Skip to content
This repository has been archived by the owner on Nov 4, 2021. It is now read-only.

Clean up Jobqueues, minor fixes for S3 Queue #451

Merged
merged 5 commits into from
Jun 4, 2021
Merged

Clean up Jobqueues, minor fixes for S3 Queue #451

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
66b7310
fix s3 consumer, cleanup graphile queue
neilkakkar Jun 1, 2021
a610f62
tests
neilkakkar Jun 1, 2021
dbec6f9
address comments
neilkakkar Jun 1, 2021
c0a1c54
Merge branch 'master' into jobqueues
neilkakkar Jun 3, 2021
f71cfb5
Merge branch 'master' into jobqueues
neilkakkar Jun 3, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
37 changes: 4 additions & 33 deletions src/main/job-queues/concurrent/graphile-queue.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,22 +5,18 @@ import { Pool } from 'pg'
import { EnqueuedJob, JobQueue, OnJobCallback, PluginsServerConfig } from '../../../types'
import { status } from '../../../utils/status'
import { createPostgresPool } from '../../../utils/utils'
import { JobQueueBase } from '../job-queue-base'

export class GraphileQueue implements JobQueue {
export class GraphileQueue extends JobQueueBase {
serverConfig: PluginsServerConfig
started: boolean
paused: boolean
onJob: OnJobCallback | null
runner: Runner | null
consumerPool: Pool | null
producerPool: Pool | null
workerUtilsPromise: Promise<WorkerUtils> | null

constructor(serverConfig: PluginsServerConfig) {
super()
this.serverConfig = serverConfig
this.started = false
this.paused = false
this.onJob = null
this.runner = null
this.consumerPool = null
this.producerPool = null
Expand Down Expand Up @@ -68,32 +64,7 @@ export class GraphileQueue implements JobQueue {

// consumer

async startConsumer(onJob: OnJobCallback): Promise<void> {
this.started = true
this.onJob = onJob
await this.syncState()
}

async stopConsumer(): Promise<void> {
this.started = false
await this.syncState()
}

async pauseConsumer(): Promise<void> {
this.paused = true
await this.syncState()
}

isConsumerPaused(): boolean {
return this.paused
}

async resumeConsumer(): Promise<void> {
this.paused = false
await this.syncState()
}

private async syncState(): Promise<void> {
protected async syncState(): Promise<void> {
if (this.started && !this.paused) {
if (!this.runner) {
this.consumerPool = await this.createPool()
Expand Down
16 changes: 10 additions & 6 deletions src/main/job-queues/job-queue-base.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,9 +36,11 @@ export class JobQueueBase implements JobQueue {
startConsumer(onJob: OnJobCallback): void
// eslint-disable-next-line @typescript-eslint/require-await
async startConsumer(onJob: OnJobCallback): Promise<void> {
this.started = true
this.onJob = onJob
await this.syncState()
if (!this.started) {
this.started = true
await this.syncState()
}
}

stopConsumer(): void
Expand All @@ -62,8 +64,10 @@ export class JobQueueBase implements JobQueue {
resumeConsumer(): void
// eslint-disable-next-line @typescript-eslint/require-await
async resumeConsumer(): Promise<void> {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

this.paused = false
await this.syncState()
if (this.paused) {
this.paused = false
await this.syncState()
}
}

// eslint-disable-next-line @typescript-eslint/require-await
Expand All @@ -77,8 +81,8 @@ export class JobQueueBase implements JobQueue {
clearTimeout(this.timeout)
}
// 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()
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

this.timeout = setTimeout(() => this.syncState(), this.intervalSeconds * 1000)
} else {
if (this.timeout) {
clearTimeout(this.timeout)
Expand Down
16 changes: 2 additions & 14 deletions src/main/job-queues/redlocked/s3-queue.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ import { S3Wrapper } from '../../../utils/db/s3-wrapper'
import { UUIDT } from '../../../utils/utils'
import { JobQueueBase } from '../job-queue-base'

const S3_POLL_INTERVAL = 5000
const S3_POLL_INTERVAL = 5

export class S3Queue extends JobQueueBase {
serverConfig: PluginsServerConfig
Expand All @@ -20,6 +20,7 @@ export class S3Queue extends JobQueueBase {
this.serverConfig = serverConfig
this.s3Wrapper = null
this.runner = null
neilkakkar marked this conversation as resolved.
Show resolved Hide resolved
this.intervalSeconds = S3_POLL_INTERVAL
}

// producer
Expand Down Expand Up @@ -50,19 +51,6 @@ export class S3Queue extends JobQueueBase {

// consumer

protected async syncState(): Promise<void> {
if (this.started && !this.paused) {
if (!this.runner) {
await this.connectS3()
this.runner = setTimeout(() => this.readState(), S3_POLL_INTERVAL)
}
} else {
if (this.runner) {
clearTimeout(this.runner)
}
}
}

async readState(): Promise<boolean> {
if (!this.s3Wrapper) {
throw new Error('S3 object not initialized')
Expand Down
63 changes: 63 additions & 0 deletions tests/jobs.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -151,6 +151,30 @@ describe('job queues', () => {
await waitForLogEntries(2)
expect(testConsole.read()).toEqual([['processEvent'], ['reply', 'runIn']])
})

test.only('polls for jobs in future', async () => {
neilkakkar marked this conversation as resolved.
Show resolved Hide resolved
const DELAY = 10000 // 10s
neilkakkar marked this conversation as resolved.
Show resolved Hide resolved

// return something to be picked up after a few loops (poll interval is 100ms)
const now = Date.now()

const job: EnqueuedJob = {
type: 'pluginJob',
payload: { key: 'value' },
timestamp: now + DELAY,
pluginConfigId: 2,
pluginConfigTeam: 3,
}

server.hub.jobQueueManager.enqueue(job)
const consumedJob: EnqueuedJob = await new Promise((resolve, reject) => {
server.hub.jobQueueManager.startConsumer((consumedJob) => {
resolve(consumedJob[0])
})
})

expect(consumedJob).toEqual(job)
})
})

describe('connection', () => {
Expand Down Expand Up @@ -284,5 +308,44 @@ describe('job queues', () => {
Key: `prefix/2020-01-01/20200101-123456.123Z-deadbeef.json.gz`,
})
})

test('polls for new jobs', async () => {
Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

these tests would fail after 60s (jest timeout) if there's no polling.

const DELAY = 10000 // 10s
// calls the right functions to read the enqueued job
mS3WrapperInstance.mockClear()

// return something to be picked up after a few loops (poll interval is 5s)
const now = Date.now()
const date = new Date(now + DELAY).toISOString()
const [day, time] = date.split('T')
const dayTime = `${day.split('-').join('')}-${time.split(':').join('')}`

const job: EnqueuedJob = {
type: 'pluginJob',
payload: { key: 'value' },
timestamp: now,
pluginConfigId: 2,
pluginConfigTeam: 3,
}

mS3WrapperInstance.listObjectsV2.mockReturnValue({
Contents: [{ Key: `prefix/${day}/${dayTime}-deadbeef.json.gz` }],
})
mS3WrapperInstance.getObject.mockReturnValueOnce({
Body: gzipSync(Buffer.from(JSON.stringify(job), 'utf8')),
})

const consumedJob: EnqueuedJob = await new Promise((resolve, reject) => {
hub.jobQueueManager.startConsumer((consumedJob) => {
resolve(consumedJob[0])
})
})
expect(consumedJob).toEqual(job)
await delay(10)
expect(mS3WrapperInstance.deleteObject).toBeCalledWith({
Bucket: 'bucket-name',
Key: `prefix/${day}/${dayTime}-deadbeef.json.gz`,
})
})
})
})