@Transaction decorator and Job Buffers don't work as expected

[skid]

I didn't want to create a ticket right away because this is more of a design issue than a bug. However, it does present a problem. Consider the following code:

// controller.ts

// ...
@Get('route')
@Transaction()
async route(@Ctx() ctx: RequestContext) {
  return await this.service.importSomeProductBatch(ctx);
}
// ...

// service.ts

//...
async importSomeProductBatch(ctx: RequestContext) {
  const buff = new JobBuffer() // Buffers `update-search-index`
  this.jobQueueService.addBuffer( buffer )

  await this.processLargeImport(ctx.req.body)

  await this.jobQueueService.flush(buffer)
  this.jobQueueService.removeBuffer(buffer)
}

Jobs like update-search-index and apply-collection-filters are created when a ProductEvent or a ProductVariant event is observed. However, with the latest changes, these won't be triggered until the transaction commits. However, the transaction commits after the buffer is flushed because it's a wrapper around the entire controller method.

My solution to this is to create my own transaction wrapper which creates its own RequestContext, which is needed to use the internal Vendure services. I think that Vendure should provide a finer-grained tool for executing transactions with the current RequestContext, since it contains valuable information.

[michaelbromley]

Have you tried using the TransactionMode setting of the @Transaction() decorator? This might allow the the control that you want.

[skid]

Yes. It works like that. However, I notice that there are still a lot of update-search-index jobs scheduled when doing large imports because of this code (I assume):

vendure/packages/core/src/plugin/default-search-plugin/default-search-plugin.ts

Line 167 in f856491

const closingNotifier$ = collectionModification$.pipe(debounceTime(50));

I think the debounce makes the event trigger after the buffer has been flushed.

It's hard to figure out why, but I'm still getting around 500 update-search-index jobs added to the queue even when using buffers for apply-collection-filters and update-search-index when importing a batch of 500 products. My production setup will probably run apply-collection-filters on all collections once every 5 minutes, so I need a way to disable adding jobs to that queue completely. Adding a buffer that never flushes looks like a memory leak - is there another way to do it?

[michaelbromley]

Adding a buffer that never flushes looks like a memory leak - is there another way to do it?

Yes, your buffer's reduce() method can simply return an empty array, which will discard all collected jobs and also allow the buffer storage to get cleared about.

Not sure why you are still getting jobs going through though, unless your collect() method is somehow missing some of them?

[skid]

I think that the aggregated apply-collection-filters triggers a reindex event for each variant, and that's practically all variants.

[michaelbromley]

Yes, you need to reduce all the collected apply-collection-filters jobs into a single one, as we do here:

vendure/packages/core/src/plugin/default-search-plugin/search-job-buffer/collection-job-buffer.ts

Lines 15 to 32 in 53a1943

	reduce(collectedJobs: Array<Job<ApplyCollectionFiltersJobData>>): Array<Job<any>> {
	const collectionIdsToUpdate = collectedJobs.reduce((result, job) => {
	return [...result, ...job.data.collectionIds];
	}, [] as ID[]);
	const referenceJob = collectedJobs[0];
	const batchedCollectionJob = new Job<ApplyCollectionFiltersJobData>({
	...referenceJob,
	id: undefined,
	data: {
	collectionIds: unique(collectionIdsToUpdate),
	ctx: referenceJob.data.ctx,
	applyToChangedVariantsOnly: referenceJob.data.applyToChangedVariantsOnly,
	},
	});
	return [batchedCollectionJob];
	}

Also worth noting is that in the search plugins, we also flush the apply-collection-filters buffer first, wait for it to complete, and then flush the other update-search-index jobs:

vendure/packages/core/src/plugin/default-search-plugin/search-job-buffer/search-job-buffer.service.ts

Line 44 in becf132

async runPendingSearchUpdates(): Promise<void> {

[skid]

I'm getting these errors Error: There is already a JobBufferProcessor with the id "search-index-buffer". Ids must be unique.

I'm creating a buffer when importing a batch of products, however, the method gets called multiple times simultaneously.
My specific situation is that I'm splitting a huge payload into chunks that get imported in parallel to leverage Google Cloud Run auto scaling. However, the same process handles multiple requests and I get this error when it happens.

Not sure if this is the intended usage of buffers, but in any case, I thought it worth mentioning.