Repository navigation
Clean up Jobqueues, minor fixes for S3 Queue - #451
Conversation
| @@ -62,8 +64,10 @@ export class JobQueueBase implements JobQueue { | |||
| resumeConsumer(): void | |||
| // eslint-disable-next-line @typescript-eslint/require-await | |||
| async resumeConsumer(): Promise<void> { | |||
There was a problem hiding this comment.
Resume is called on drain events emitted by Piscina, which results in multiple calls to sync state. This is okay, as long as dequeue code is idempotent (which it isn't, thus leading to race conditions).
We didn't face this with graphile because the runner object is unchanged over multiple sync state calls.
| // eslint-disable-next-line @typescript-eslint/await-thenable | ||
| const hadSomething = await this.readState() | ||
| this.timeout = setTimeout(() => this.syncState(), hadSomething ? 0 : this.intervalSeconds * 1000) | ||
| await this.readState() |
There was a problem hiding this comment.
Given existing semantics, readState is supposed to finish everything required with the jobs seen so far. So, it doesn't make sense to poll again right after, if it found some jobs to process.
There was a problem hiding this comment.
The idea here is that if you have 10k jobs to run, you shouldn't wait 5 sec between them (or between every 100 of them) :). So if we got a job with the last run, immediately see if there's another one. Otherwise wait a few seconds and then try again.
There was a problem hiding this comment.
Ah, that makes sense... assuming there's batching involved. I haven't seen it so far (am I missing something?) - so didn't assume such a state was possible.
Will revert.
| }) | ||
| }) | ||
|
|
||
| test('polls for new jobs', async () => { |
There was a problem hiding this comment.
these tests would fail after 60s (jest timeout) if there's no polling.
mariusandra
left a comment
mariusandra
left a comment
There was a problem hiding this comment.
some thoughts below, still haven't ran it locally
| // eslint-disable-next-line @typescript-eslint/await-thenable | ||
| const hadSomething = await this.readState() | ||
| this.timeout = setTimeout(() => this.syncState(), hadSomething ? 0 : this.intervalSeconds * 1000) | ||
| await this.readState() |
There was a problem hiding this comment.
The idea here is that if you have 10k jobs to run, you shouldn't wait 5 sec between them (or between every 100 of them) :). So if we got a job with the last run, immediately see if there's another one. Otherwise wait a few seconds and then try again.
mariusandra
left a comment
mariusandra
left a comment
There was a problem hiding this comment.
These changes make sense
…/plugin-server#451) * fix s3 consumer, cleanup graphile queue * tests * address comments
2 participants
Changes
Ensures S3 queue consumer runs.
Makes GraphileQueue adhere to JobQueueBase.
Checklist