perf: keep jobs in waiting list when queue is paused #2769
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,27 @@ | ||
| --- | ||
| description: Tips and hints on how to migrate to v6. | ||
| --- | ||
|
|
||
| # Migration to v6 | ||
|
|
||
| Make sure to pass **skipMigrationsExecution** option in any of our instances as false in order to execute all necessary changes when coming from an older version | ||
|
|
||
| ## Migration of deprecated paused key | ||
|
|
||
| If you have paused queues after upgrading to this version. These jobs will be moved to wait state when initializing any of our instances (Worker, Queue, QueueEvents or FlowProducer). | ||
|
|
||
| Paused key is not longer needed as this state is already represented by queue meta key. It also improve the process of pausing or resuming a queue as we don't need to rename any key. | ||
|
There was a problem hiding this comment. by meta key -> inside the meta key |
||
|
|
||
| ## Remove legacy markers | ||
|
There was a problem hiding this comment. I think we are going to need more detailed information on how to safely perform the migration. |
||
|
|
||
| When migrating from versions before v5. | ||
| It's recommended to do this process: | ||
|
|
||
| 1. Pause your queues. | ||
| 2. Upgrade to v6. | ||
| 3. Instantiate any instance where migrations will be executed. | ||
| 4. Resume your queues. | ||
|
|
||
| This way you will prevent that your workers pick a legacy marker. | ||
|
|
||
| A second option would be to do incremental upgrades. | ||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,116 @@ | ||
| import { RedisClient } from '../interfaces'; | ||
|
|
||
| export interface MigrationOptions { | ||
| prefix: string; | ||
| queueName: string; | ||
| } | ||
|
|
||
| export type MigrationFunction = ( | ||
| client: RedisClient, | ||
| opts: MigrationOptions, | ||
| ) => Promise<void>; | ||
|
|
||
| export const checkPendingMigrations = async ( | ||
| client: RedisClient, | ||
| opts: MigrationOptions, | ||
| ) => { | ||
| const migrationsKey = getRedisKeyFromOpts(opts, 'migrations'); | ||
| const existingMigrations = await client.zrange(migrationsKey, 0, -1); | ||
| return migrations.some( | ||
| migration => | ||
| !existingMigrations.includes(`${migration.version}-${migration.name}`), | ||
| ); | ||
| }; | ||
|
|
||
| export const migrations: { | ||
| name: string; | ||
| version: string; | ||
| migrate: MigrationFunction; | ||
| }[] = [ | ||
| { | ||
| name: 'remove-legacy-markers', | ||
| version: '6.0.0', | ||
| migrate: async (client: RedisClient, opts: MigrationOptions) => { | ||
| const keys: (string | number)[] = [ | ||
| getRedisKeyFromOpts(opts, 'wait'), | ||
| getRedisKeyFromOpts(opts, 'paused'), | ||
| getRedisKeyFromOpts(opts, 'meta'), | ||
| getRedisKeyFromOpts(opts, 'completed'), | ||
| getRedisKeyFromOpts(opts, 'failed'), | ||
| ]; | ||
| const args = [getRedisKeyFromOpts(opts, '')]; | ||
|
|
||
| await (<any>client).removeLegacyMarkers(keys.concat(args)); | ||
| }, | ||
| }, | ||
| { | ||
| name: 'migrate-paused-jobs', | ||
| version: '6.0.0', | ||
| migrate: async (client: RedisClient, opts: MigrationOptions) => { | ||
| let cursor = 0; | ||
| do { | ||
| const keys: (string | number)[] = [ | ||
| getRedisKeyFromOpts(opts, 'paused'), | ||
| getRedisKeyFromOpts(opts, 'wait'), | ||
| ]; | ||
| const args = [1000]; | ||
| cursor = await (<any>client).migrateDeprecatedPausedKey( | ||
| keys.concat(args), | ||
| ); | ||
| } while (cursor); | ||
| }, | ||
| }, | ||
| ]; | ||
|
|
||
| /** | ||
| * Run Migrations. | ||
| * | ||
| * This method is used to run possibly existing migrations for the queue. | ||
| * | ||
| * Normally, if there are pending migrations, the Queue, Worker and QueueEvents instances | ||
| * will throw an error when they are instantiated. Use then this method to run the migrations | ||
| * before instantiating the instances. | ||
| * | ||
| * @param redisClient The Redis client instance | ||
| * @param opts The options for the migration | ||
| * | ||
| * @sa https://docs.bullmq.io/guide/migrations | ||
| */ | ||
| export const runMigrations = async ( | ||
| redisClient: RedisClient, | ||
| opts: { | ||
| prefix?: string; | ||
| queueName: string; | ||
| }, | ||
| ) => { | ||
| const prefix = opts.prefix || 'bull'; | ||
| const migrationsKey = getRedisKeyFromOpts({ prefix, ...opts }, 'migrations'); | ||
|
|
||
| // The migrations key is a ZSET with the migration timestamp as the score | ||
| for (const migration of migrations) { | ||
| const migrationId = `${migration.version}-${migration.name}`; | ||
| const pendingMigration = !!(await redisClient.zscore( | ||
| migrationsKey, | ||
| migrationId, | ||
| )); | ||
| if (pendingMigration) { | ||
| continue; | ||
| } | ||
| console.log(`[BULLMQ] Running migration ${migrationId}`); | ||
| try { | ||
| await migration.migrate(redisClient, { | ||
| prefix, | ||
| queueName: opts.queueName, | ||
| }); | ||
| await redisClient.zadd(migrationsKey, Date.now(), migrationId); | ||
| } catch (err) { | ||
| console.error(`[BULLMQ] Migration ${migrationId} failed: ${err}`); | ||
| break; | ||
| } | ||
| console.log(`[BULLMQ] Migration ${migrationId} completed`); | ||
| } | ||
| }; | ||
|
|
||
| function getRedisKeyFromOpts(opts: MigrationOptions, key: string): string { | ||
| return `${opts.prefix}:${opts.queueName}:${key}`; | ||
| } |
Oops, something went wrong.
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
do you think default should be perform the migration? I think it makes more sense to have the reverse, default is not to do the migration, you must explicitly do it, In fact it would be better to give an error if the queue has not been migrated yet and require you to manually migrate first as part of your deployment steps.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
So basically we should have only one way to migrate which would be using the run migrations utility, and not implicitly when the queue is instantiated.