From fe455dd3f753a10bfda0c2accefc3f3005d81f7f Mon Sep 17 00:00:00 2001 From: Rotimi Best Date: Wed, 20 May 2026 14:07:45 +0100 Subject: [PATCH] fix: users can retriger transcription --- apps/jobs/src/workers/maintenance.ts | 52 ++++++++++++++++++- packages/db/src/queries/jobs/media-job.ts | 35 ++++++++++++- .../media-transcript/media-transcript.ts | 23 ++++++-- packages/jobs/src/payloads/maintenance.ts | 11 ++++ packages/jobs/src/queues/names.ts | 3 +- 5 files changed, 116 insertions(+), 8 deletions(-) diff --git a/apps/jobs/src/workers/maintenance.ts b/apps/jobs/src/workers/maintenance.ts index 9c7debce..aed1e5ac 100644 --- a/apps/jobs/src/workers/maintenance.ts +++ b/apps/jobs/src/workers/maintenance.ts @@ -1,12 +1,13 @@ import './../bootstrap'; -import { Worker } from 'bullmq'; +import { Queue, Worker } from 'bullmq'; -import { pruneDeadLetterJobsOlderThan } from '@cio/db/queries'; +import { pruneDeadLetterJobsOlderThan, reapStuckMediaJobs } from '@cio/db/queries'; import { JOB_NAMES, QUEUE_NAMES, ZDeadLetterCleanupPayload, + ZMediaJobReapPayload, ZRetentionCompactPayload, createRedisConnection } from '@cio/jobs'; @@ -36,11 +37,56 @@ const worker = new Worker( return { compacted: 0 }; } + if (job.name === JOB_NAMES.maintenance.mediaJobReap) { + const data = ZMediaJobReapPayload.parse(job.data ?? {}); + const cutoffIso = new Date(Date.now() - data.staleAfterMinutes * 60 * 1_000).toISOString(); + const reaped = await reapStuckMediaJobs(cutoffIso); + + if (reaped.length > 0) { + log.warn('media-job-reap-done', { + reaped: reaped.length, + cutoffIso, + ids: reaped.map((row) => row.id) + }); + } + + return { reaped: reaped.length }; + } + throw new Error(`Unknown maintenance job: ${job.name}`); }, { connection, concurrency: 1 } ); +/** + * BullMQ repeatable scheduler for the media-job reaper. `upsertJobScheduler` + * is idempotent on the scheduler id, so it's safe to call on every worker + * boot. Without this, stuck `media_job` rows would never get cleaned up. + */ +const MEDIA_JOB_REAP_INTERVAL_MS = 5 * 60 * 1_000; +// BullMQ requires a dedicated connection per Worker; give the Queue its own +// so the scheduler upsert doesn't piggyback on the worker's blocking client. +const schedulerConnection = createRedisConnection(); +const maintenanceQueue = new Queue(QUEUE_NAMES.maintenance, { connection: schedulerConnection }); + +async function registerSchedulers(): Promise { + try { + await maintenanceQueue.upsertJobScheduler( + 'media-job-reap-scheduler', + { every: MEDIA_JOB_REAP_INTERVAL_MS }, + { name: JOB_NAMES.maintenance.mediaJobReap, data: {} } + ); + log.info('maintenance-scheduler-registered', { + name: JOB_NAMES.maintenance.mediaJobReap, + everyMs: MEDIA_JOB_REAP_INTERVAL_MS + }); + } catch (err) { + log.error('maintenance-scheduler-register-failed', { error: errorMessage(err) }); + } +} + +void registerSchedulers(); + worker.on('ready', () => log.info('maintenance-worker-ready')); worker.on('failed', (job, err) => log.error('maintenance-job-failed', { jobName: job?.name, error: errorMessage(err) }) @@ -50,6 +96,8 @@ worker.on('error', (err) => log.error('maintenance-worker-error', { error: error const shutdown = async (signal: string) => { log.info('maintenance-worker-shutdown', { signal }); await worker.close(); + await maintenanceQueue.close(); + await schedulerConnection.quit(); await connection.quit(); process.exit(0); }; diff --git a/packages/db/src/queries/jobs/media-job.ts b/packages/db/src/queries/jobs/media-job.ts index 0dad886a..3ab7e7e8 100644 --- a/packages/db/src/queries/jobs/media-job.ts +++ b/packages/db/src/queries/jobs/media-job.ts @@ -1,6 +1,6 @@ import * as schema from '@db/schema'; -import { and, desc, eq, inArray, sql } from '@db/drizzle'; +import { and, desc, eq, inArray, lt, sql } from '@db/drizzle'; import { db, type DbOrTxClient } from '@db/drizzle'; import type { TJobStatus, TMediaJob, TNewMediaJob } from '@db/types'; @@ -109,6 +109,39 @@ export async function updateMediaJob( } } +/** + * Flip any `queued` / `running` media_job rows untouched since `staleBeforeIso` + * to `failed`. These belong to crashed workers, Redis-less runs, or BullMQ + * jobs that vanished — without this sweep the active-job guard would keep + * blocking new runs forever. + * + * Returns the rows that were updated so callers can log or dead-letter them. + */ +export async function reapStuckMediaJobs( + staleBeforeIso: string, + dbClient: DbOrTxClient = db +): Promise[]> { + try { + return await dbClient + .update(schema.mediaJob) + .set({ + status: 'failed', + stage: 'failed', + error: { code: 'STUCK_AFTER_RESTART', message: 'No worker progress within the stale window' }, + updatedAt: new Date().toISOString() + }) + .where(and(inArray(schema.mediaJob.status, ['queued', 'running']), lt(schema.mediaJob.updatedAt, staleBeforeIso))) + .returning({ + id: schema.mediaJob.id, + organizationId: schema.mediaJob.organizationId, + assetId: schema.mediaJob.assetId + }); + } catch (error) { + console.error('reapStuckMediaJobs error:', error); + throw new Error('Failed to reap stuck media jobs'); + } +} + export async function requestMediaJobCancel( jobId: string, orgId: string, diff --git a/packages/db/src/queries/media-transcript/media-transcript.ts b/packages/db/src/queries/media-transcript/media-transcript.ts index d4ce2cc7..3433de2d 100644 --- a/packages/db/src/queries/media-transcript/media-transcript.ts +++ b/packages/db/src/queries/media-transcript/media-transcript.ts @@ -1,6 +1,6 @@ import * as schema from '@db/schema'; -import { and, eq, inArray } from '@db/drizzle'; +import { and, eq, gte, inArray } from '@db/drizzle'; import { db, type DbOrTxClient } from '@db/drizzle'; import type { TMediaTranscript, TNewMediaTranscript } from '@db/types'; @@ -99,14 +99,28 @@ export async function listMediaTranscriptsByAssetIds( } /** - * Returns true if any media_job for this asset is still queued or running. + * Default freshness window for "active" media_job rows. Rows older than this + * are assumed to belong to a crashed worker / a Redis-less run and are + * ignored by `hasActiveMediaJobForAsset` so users can retry. A separate + * sweeper (`reapStuckMediaJobs`) flips those stale rows to `failed`. + */ +export const MEDIA_JOB_STALE_AFTER_MINUTES = 30; + +/** + * Returns true if any media_job for this asset is still queued or running and + * has been touched within `staleAfterMinutes` (default 30). Stale rows are + * treated as dead: they shouldn't block new runs because the worker that + * owned them is gone (server restart, Redis flush, lost BullMQ job, etc.). */ export async function hasActiveMediaJobForAsset( assetId: string, organizationId: string, - dbClient: DbOrTxClient = db + options: { staleAfterMinutes?: number; dbClient?: DbOrTxClient } = {} ): Promise { + const { staleAfterMinutes = MEDIA_JOB_STALE_AFTER_MINUTES, dbClient = db } = options; + try { + const freshSinceIso = new Date(Date.now() - staleAfterMinutes * 60 * 1_000).toISOString(); const [row] = await dbClient .select({ id: schema.mediaJob.id }) .from(schema.mediaJob) @@ -114,7 +128,8 @@ export async function hasActiveMediaJobForAsset( and( eq(schema.mediaJob.assetId, assetId), eq(schema.mediaJob.organizationId, organizationId), - inArray(schema.mediaJob.status, ['queued', 'running']) + inArray(schema.mediaJob.status, ['queued', 'running']), + gte(schema.mediaJob.updatedAt, freshSinceIso) ) ) .limit(1); diff --git a/packages/jobs/src/payloads/maintenance.ts b/packages/jobs/src/payloads/maintenance.ts index 398434da..a448e879 100644 --- a/packages/jobs/src/payloads/maintenance.ts +++ b/packages/jobs/src/payloads/maintenance.ts @@ -16,3 +16,14 @@ export const ZDeadLetterCleanupPayload = z.object({ olderThanDays: z.number().int().positive().default(180) }); export type TDeadLetterCleanupPayload = z.infer; + +/** + * Sweep `media_job` rows that are still `queued` / `running` but haven't been + * touched in `staleAfterMinutes`. Marks them `failed` so the active-job guard + * stops blocking new runs after a worker crash, server restart, or lost + * BullMQ job. + */ +export const ZMediaJobReapPayload = z.object({ + staleAfterMinutes: z.number().int().positive().default(30) +}); +export type TMediaJobReapPayload = z.infer; diff --git a/packages/jobs/src/queues/names.ts b/packages/jobs/src/queues/names.ts index 27136f85..50b8ff0d 100644 --- a/packages/jobs/src/queues/names.ts +++ b/packages/jobs/src/queues/names.ts @@ -37,6 +37,7 @@ export const JOB_NAMES = { }, maintenance: { retentionCompact: 'retention-compact', - deadLetterCleanup: 'dead-letter-cleanup' + deadLetterCleanup: 'dead-letter-cleanup', + mediaJobReap: 'media-job-reap' } } as const;