Big refactor #2 + Jobs Jobs Jobs

src/config/config.ts

@@ -59,8 +59,8 @@ export function getDefaultConfig(): PluginsServerConfig {
         DISTINCT_ID_LRU_SIZE: 10000,
         INTERNAL_MMDB_SERVER_PORT: 0,
         PLUGIN_SERVER_IDLE: false,
-        RETRY_QUEUES: '',
-        RETRY_QUEUE_GRAPHILE_URL: '',
+        JOB_QUEUES: '',
+        JOB_QUEUE_GRAPHILE_URL: '',
         ENABLE_PERSISTENT_CONSOLE: false, // TODO: remove when persistent console ships in main repo
         STALENESS_RESTART_SECONDS: 0,
     }
@@ -104,8 +104,8 @@ export function getConfigHelp(): Record<keyof PluginsServerConfig, string> {
         DISTINCT_ID_LRU_SIZE: 'size of persons distinct ID LRU cache',
         INTERNAL_MMDB_SERVER_PORT: 'port of the internal server used for IP location (0 means random)',
         PLUGIN_SERVER_IDLE: 'whether to disengage the plugin server, e.g. for development',
-        RETRY_QUEUES: 'retry queue engine and fallback queues',
-        RETRY_QUEUE_GRAPHILE_URL: 'use a different postgres connection in the graphile retry queue',
+        JOB_QUEUES: 'retry queue engine and fallback queues',
+        JOB_QUEUE_GRAPHILE_URL: 'use a different postgres connection in the graphile retry queue',
         STALENESS_RESTART_SECONDS: 'trigger a restart if no event ingested for this duration',
     }
 }

src/main/job-queues/fs-queue.ts

@@ -1,4 +1,4 @@
-import { EnqueuedRetry, JobQueue, OnRetryCallback } from '../../types'
+import { EnqueuedJob, JobQueue, OnJobCallback } from '../../types'
 import Timeout = NodeJS.Timeout
 import * as fs from 'fs'
 import * as path from 'path'
@@ -17,20 +17,22 @@ export class FsQueue implements JobQueue {
         this.started = false
         this.interval = null
         this.filename = filename || path.join(process.cwd(), 'tmp', 'fs-queue.txt')
+    }
 
+    connectProducer(): void {
         fs.mkdirSync(path.dirname(this.filename), { recursive: true })
         fs.writeFileSync(this.filename, '')
     }
 
-    enqueue(retry: EnqueuedRetry): Promise<void> | void {
-        fs.appendFileSync(this.filename, `${JSON.stringify(retry)}\n`)
+    enqueue(job: EnqueuedJob): Promise<void> | void {
+        fs.appendFileSync(this.filename, `${JSON.stringify(job)}\n`)
     }
 
-    quit(): void {
+    disconnectProducer(): void {
         // nothing to do
     }
 
-    startConsumer(onRetry: OnRetryCallback): void {
+    startConsumer(onJob: OnJobCallback): void {
         fs.writeFileSync(this.filename, '')
         this.started = true
         this.interval = setInterval(() => {
@@ -43,14 +45,14 @@ export class FsQueue implements JobQueue {
                 .toString()
                 .split('\n')
                 .filter((a) => a)
-                .map((s) => JSON.parse(s) as EnqueuedRetry)
+                .map((s) => JSON.parse(s) as EnqueuedJob)
 
             const newQueue = queue.filter((element) => element.timestamp < timestamp)
             if (newQueue.length > 0) {
                 const oldQueue = queue.filter((element) => element.timestamp >= timestamp)
                 fs.writeFileSync(this.filename, `${oldQueue.map((q) => JSON.stringify(q)).join('\n')}\n`)
 
-                void onRetry(newQueue)
+                void onJob(newQueue)
             }
         }, 1000)
     }

src/main/job-queues/graphile-queue.ts

@@ -1,49 +1,44 @@
 import { makeWorkerUtils, run, Runner, WorkerUtils, WorkerUtilsOptions } from 'graphile-worker'
 
-import { EnqueuedRetry, JobQueue, OnRetryCallback, PluginsServer } from '../../types'
+import { EnqueuedJob, JobQueue, OnJobCallback, PluginsServer } from '../../types'
 
 export class GraphileQueue implements JobQueue {
     pluginsServer: PluginsServer
     started: boolean
     paused: boolean
-    onRetry: OnRetryCallback | null
+    onJob: OnJobCallback | null
     runner: Runner | null
-    workerUtils: WorkerUtils | null
+    workerUtilsPromise: Promise<WorkerUtils> | null
 
     constructor(pluginsServer: PluginsServer) {
         this.pluginsServer = pluginsServer
         this.started = false
         this.paused = false
-        this.onRetry = null
+        this.onJob = null
         this.runner = null
-        this.workerUtils = null
+        this.workerUtilsPromise = null
     }
 
-    async enqueue(retry: EnqueuedRetry): Promise<void> {
-        if (!this.workerUtils) {
-            this.workerUtils = await makeWorkerUtils(
-                this.pluginsServer.RETRY_QUEUE_GRAPHILE_URL
-                    ? {
-                          connectionString: this.pluginsServer.RETRY_QUEUE_GRAPHILE_URL,
-                      }
-                    : ({
-                          pgPool: this.pluginsServer.postgres,
-                      } as WorkerUtilsOptions)
-            )
-            await this.workerUtils.migrate()
-        }
-        await this.workerUtils.addJob('retryTask', retry, { runAt: new Date(retry.timestamp), maxAttempts: 1 })
+    async connectProducer(): Promise<void> {
+        await (await this.getWorkerUtils()).migrate()
+    }
+
+    async enqueue(retry: EnqueuedJob): Promise<void> {
+        await (await this.getWorkerUtils()).addJob('pluginJob', retry, {
+            runAt: new Date(retry.timestamp),
+            maxAttempts: 1,
+        })
     }
 
-    async quit(): Promise<void> {
-        const oldWorkerUtils = this.workerUtils
-        this.workerUtils = null
+    async disconnectProducer(): Promise<void> {
+        const oldWorkerUtils = await this.workerUtilsPromise
+        this.workerUtilsPromise = null
         await oldWorkerUtils?.release()
     }
 
-    async startConsumer(onRetry: OnRetryCallback): Promise<void> {
+    async startConsumer(onJob: OnJobCallback): Promise<void> {
         this.started = true
-        this.onRetry = onRetry
+        this.onJob = onJob
         await this.syncState()
     }
 
@@ -66,19 +61,19 @@ export class GraphileQueue implements JobQueue {
         await this.syncState()
     }
 
-    async syncState(): Promise<void> {
+    private async syncState(): Promise<void> {
         if (this.started && !this.paused) {
             if (!this.runner) {
                 this.runner = await run({
-                    connectionString: this.pluginsServer.DATABASE_URL,
+                    ...this.getConnectionOptions(),
                     concurrency: 1,
                     // Install signal handlers for graceful shutdown on SIGINT, SIGTERM, etc
                     noHandleSignals: false,
                     pollInterval: 100,
                     // you can set the taskList or taskDirectory but not both
                     taskList: {
-                        retryTask: (payload) => {
-                            void this.onRetry?.([payload as EnqueuedRetry])
+                        pluginJob: (payload) => {
+                            void this.onJob?.([payload as EnqueuedJob])
                         },
                     },
                 })
@@ -91,4 +86,21 @@ export class GraphileQueue implements JobQueue {
             }
         }
     }
+
+    private getConnectionOptions(): Partial<WorkerUtilsOptions> {
+        return this.pluginsServer.JOB_QUEUE_GRAPHILE_URL
+            ? {
+                  connectionString: this.pluginsServer.JOB_QUEUE_GRAPHILE_URL,
+              }
+            : ({
+                  pgPool: this.pluginsServer.postgres,
+              } as Partial<WorkerUtilsOptions>)
+    }
+
+    private async getWorkerUtils(): Promise<WorkerUtils> {
+        if (!this.workerUtilsPromise) {
+            this.workerUtilsPromise = makeWorkerUtils(this.getConnectionOptions())
+        }
+        return await this.workerUtilsPromise
+    }
 }

src/main/job-queues/job-queue-consumer.ts

@@ -1,35 +1,35 @@
 import Piscina from '@posthog/piscina'
 
-import { JobQueueConsumerControl, OnRetryCallback, PluginsServer } from '../../types'
+import { JobQueueConsumerControl, OnJobCallback, PluginsServer } from '../../types'
 import { startRedlock } from '../../utils/redlock'
 import { status } from '../../utils/status'
 import { pauseQueueIfWorkerFull } from '../ingestion-queues/queue'
 
-export const LOCKED_RESOURCE = 'plugin-server:locks:retry-queue-consumer'
+export const LOCKED_RESOURCE = 'plugin-server:locks:job-queue-consumer'
 
 export async function startJobQueueConsumer(server: PluginsServer, piscina: Piscina): Promise<JobQueueConsumerControl> {
-    status.info('🔄', 'Starting retry queue consumer, trying to get lock...')
+    status.info('🔄', 'Starting job queue consumer, trying to get lock...')
 
-    const onRetry: OnRetryCallback = async (retries) => {
-        pauseQueueIfWorkerFull(server.retryQueueManager.pauseConsumer, server, piscina)
-        for (const retry of retries) {
-            await piscina.runTask({ task: 'retry', args: { retry } })
+    const onJob: OnJobCallback = async (jobs) => {
+        pauseQueueIfWorkerFull(server.jobQueueManager.pauseConsumer, server, piscina)
+        for (const job of jobs) {
+            await piscina.runTask({ task: 'runJob', args: { job } })
         }
     }
 
     const unlock = await startRedlock({
         server,
         resource: LOCKED_RESOURCE,
         onLock: async () => {
-            status.info('🔄', 'Retry queue consumer lock aquired')
-            await server.retryQueueManager.startConsumer(onRetry)
+            status.info('🔄', 'Job queue consumer lock aquired')
+            await server.jobQueueManager.startConsumer(onJob)
         },
         onUnlock: async () => {
-            status.info('🔄', 'Stopping retry queue consumer')
-            await server.retryQueueManager.stopConsumer()
+            status.info('🔄', 'Stopping job queue consumer')
+            await server.jobQueueManager.stopConsumer()
         },
         ttl: server.SCHEDULE_LOCK_TTL,
     })
 
-    return { stop: () => unlock(), resume: () => server.retryQueueManager.resumeConsumer() }
+    return { stop: () => unlock(), resume: () => server.jobQueueManager.resumeConsumer() }
 }

src/main/job-queues/job-queue-manager.ts

@@ -1,6 +1,7 @@
 import * as Sentry from '@sentry/node'
 
-import { EnqueuedRetry, JobQueue, OnRetryCallback, PluginsServer } from '../../types'
+import { EnqueuedJob, JobQueue, OnJobCallback, PluginsServer } from '../../types'
+import { status } from '../../utils/status'
 import { FsQueue } from './fs-queue'
 import { GraphileQueue } from './graphile-queue'
 
@@ -17,35 +18,50 @@ const queues: Record<JobQueueType, (server: PluginsServer) => JobQueue> = {
 export class JobQueueManager implements JobQueue {
     pluginsServer: PluginsServer
     jobQueues: JobQueue[]
+    jobQueueTypes: JobQueueType[]
 
     constructor(pluginsServer: PluginsServer) {
         this.pluginsServer = pluginsServer
 
-        this.jobQueues = pluginsServer.RETRY_QUEUES.split(',')
+        this.jobQueueTypes = pluginsServer.JOB_QUEUES.split(',')
             .map((q) => q.trim() as JobQueueType)
             .filter((q) => !!q)
-            .map(
-                (queue): JobQueue => {
-                    if (queues[queue]) {
-                        return queues[queue](pluginsServer)
-                    } else {
-                        throw new Error(`Unknown retry queue "${queue}"`)
-                    }
+
+        this.jobQueues = this.jobQueueTypes.map(
+            (queue): JobQueue => {
+                if (queues[queue]) {
+                    return queues[queue](pluginsServer)
+                } else {
+                    throw new Error(`Unknown job queue "${queue}"`)
+                }
+            }
+        )
+    }
+
+    async connectProducer(): Promise<void> {
+        await Promise.all(
+            this.jobQueues.map(async (jobQueue, index) => {
+                try {
+                    await jobQueue.connectProducer()
+                    status.info('💂', `Connected to job queue producer: ${this.jobQueueTypes[index]}`)
+                } catch (error) {
+                    Sentry.captureException(error)
                 }
-            )
+            })
+        )
     }
 
-    async enqueue(retry: EnqueuedRetry): Promise<void> {
-        for (const retryQueue of this.jobQueues) {
+    async enqueue(job: EnqueuedJob): Promise<void> {
+        for (const jobQueue of this.jobQueues) {
             try {
-                await retryQueue.enqueue(retry)
+                await jobQueue.enqueue(job)
                 return
             } catch (error) {
                 // if one fails, take the next queue
                 Sentry.captureException(error, {
                     extra: {
-                        retry: JSON.stringify(retry),
-                        queue: retryQueue.toString(),
+                        job: JSON.stringify(job),
+                        queue: jobQueue.toString(),
                         queues: this.jobQueues.map((q) => q.toString()),
                     },
                 })
@@ -54,12 +70,12 @@ export class JobQueueManager implements JobQueue {
         throw new Error('No JobQueue available')
     }
 
-    async quit(): Promise<void> {
-        await Promise.all(this.jobQueues.map((r) => r.quit()))
+    async disconnectProducer(): Promise<void> {
+        await Promise.all(this.jobQueues.map((r) => r.disconnectProducer()))
     }
 
-    async startConsumer(onRetry: OnRetryCallback): Promise<void> {
-        await Promise.all(this.jobQueues.map((r) => r.startConsumer(onRetry)))
+    async startConsumer(onJob: OnJobCallback): Promise<void> {
+        await Promise.all(this.jobQueues.map((r) => r.startConsumer(onJob)))
     }
 
     async stopConsumer(): Promise<void> {