admin/twenty
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

[message-queue] Add job auto-removal (#4973)

Browse Source 
## Context

With the addition of cronjobs, the app is building a lot of jobs and
stores them indefinitely. There is no real point to keep all of them in
the queue once they have been processed (completed or failed) so we are
adding a new default option to the bull-mq driver

## Implementation
See bull-mq JobsOption doc
```typescript
    /**
     * If true, removes the job when it successfully completes
     * When given a number, it specifies the maximum amount of
     * jobs to keep, or you can provide an object specifying max
     * age and/or count to keep. It overrides whatever setting is used in the worker.
     * Default behavior is to keep the job in the completed set.
     */
    removeOnComplete?: boolean | number | KeepJobs;
    /**
     * If true, removes the job when it fails after all attempts.
     * When given a number, it specifies the maximum amount of
     * jobs to keep, or you can provide an object specifying max
     * age and/or count to keep. It overrides whatever setting is used in the worker.
     * Default behavior is to keep the job in the failed set.
     */
    removeOnFail?: boolean | number | KeepJobs;
```

removeOnFail should be a bit higher since they are the ones we are most
likely looking at when needed.
This commit is contained in:
Weiko
2024-04-15 17:33:27 +02:00
committed by GitHub
parent 11d928baa3
commit db2935b877
1 changed files with 7 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
packages/twenty-server/src/engine/integrations/message-queue/drivers/bullmq.driver.ts
View File

@ -1,4 +1,4 @@
import { Queue, QueueOptions, Worker } from 'bullmq';
import { JobsOptions, Queue, QueueOptions, Worker } from 'bullmq';
import {
QueueCronJobOptions,
@ -63,10 +63,12 @@ export class BullMQDriver implements MessageQueueDriver {
`Queue ${queueName} is not registered, make sure you have added it as a queue provider`,
);
}
const queueOptions = {
const queueOptions: JobsOptions = {
jobId: options?.id,
priority: options?.priority,
repeat: options?.repeat,
removeOnComplete: 100,
removeOnFail: 500,
};
await this.queueMap[queueName].add(jobName, data, queueOptions);
@ -93,10 +95,12 @@ export class BullMQDriver implements MessageQueueDriver {
`Queue ${queueName} is not registered, make sure you have added it as a queue provider`,
);
}
const queueOptions = {
const queueOptions: JobsOptions = {
jobId: options?.id,
priority: options?.priority,
attempts: 1 + (options?.retryLimit || 0),
removeOnComplete: 100,
removeOnFail: 500,
};
await this.queueMap[queueName].add(jobName, data, queueOptions);