From 5c1ea23b811e161713ad0fb0324fd173459a44d0 Mon Sep 17 00:00:00 2001 From: Chris Sherwood Date: Mon, 10 Aug 2026 17:27:58 -0700 Subject: [PATCH] fix(queue): clear terminal jobs before re-dispatch so downloads aren't silently dropped Three dispatch sites pin a deterministic jobId. `queue.add` with an existing custom jobId returns the existing job rather than throwing, so any retained terminal record makes later dispatches a silent no-op while still reporting success to the caller. - download_model_job: only `failed` records were cleared, so once a model downloaded successfully its retained completed job blocked every later request for that model. Deleting the model and reinstalling it did nothing, and a restart did not help because the record is persisted in Redis. Adds getActiveByModelName, mirroring RunDownloadJob.getActiveByUrl. - download_service.retryFailedJob: the model branch dispatched before removing, so remove() targeted the job just enqueued under the same id instead of the old record. Flipped to remove-then-dispatch, matching the file branch. - embed_file_job: a retained failed record made re-indexing that file a no-op, returning 202 "Indexing queued" with nothing enqueued. In-flight jobs are still returned as-is, so re-clicking during an active download or embed stays idempotent. Diagnosis by @caweis in #1214, including the retry-ordering race and the embed case. Reimplemented in-house rather than ported, as noted there. Closes #1214 --- admin/app/jobs/download_model_job.ts | 44 ++++++++++++++++++++++---- admin/app/jobs/embed_file_job.ts | 27 ++++++++++++++++ admin/app/services/download_service.ts | 7 +++- 3 files changed, 71 insertions(+), 7 deletions(-) diff --git a/admin/app/jobs/download_model_job.ts b/admin/app/jobs/download_model_job.ts index c79d59e..0921950 100644 --- a/admin/app/jobs/download_model_job.ts +++ b/admin/app/jobs/download_model_job.ts @@ -160,17 +160,49 @@ export class DownloadModelJob { return await queue.getJob(jobId) } + /** + * Returns the job for this model only when it is genuinely in flight. + * + * A terminal record (completed or failed) is removed instead, because the + * jobId is a hash of the model name and `queue.add` with an existing custom + * jobId returns the existing job rather than throwing. Leaving a terminal + * record in place therefore makes every later dispatch for that model a + * silent no-op. Mirrors RunDownloadJob.getActiveByUrl. + */ + static async getActiveByModelName(modelName: string): Promise { + const job = await this.getByModelName(modelName) + if (!job) return undefined + + const state = await job.getState() + if (state === 'active' || state === 'waiting' || state === 'delayed') { + return job + } + + // Terminal state -- clean up so it doesn't block a re-download + try { + await job.remove() + } catch { + // May already be gone + } + return undefined + } + static async dispatch(params: DownloadModelJobParams) { const queueService = QueueService.getInstance() const queue = queueService.getQueue(this.queue) const jobId = this.getJobId(params.modelName) - // Clear any previous failed job so a fresh attempt can be dispatched - const existing = await queue.getJob(jobId) - if (existing) { - const state = await existing.getState() - if (state === 'failed') { - await existing.remove() + // Return an in-flight download as-is, and clear any terminal record so a + // fresh attempt can be dispatched. Previously this only cleared `failed`, + // so once a model had been downloaded successfully its retained completed + // job deduped every later request: the API still reported success and no + // worker ran, leaving the model uninstallable until Redis was flushed. + const inFlight = await this.getActiveByModelName(params.modelName) + if (inFlight) { + return { + job: inFlight, + created: false, + message: `Download already in progress for model ${params.modelName}`, } } diff --git a/admin/app/jobs/embed_file_job.ts b/admin/app/jobs/embed_file_job.ts index 65af19b..b1ae07b 100644 --- a/admin/app/jobs/embed_file_job.ts +++ b/admin/app/jobs/embed_file_job.ts @@ -389,6 +389,33 @@ export class EmbedFileJob { jobOptions.jobId = initialJobId } + // Deterministic-jobId dispatches must clear a terminal record before adding. + // `queue.add` with an existing custom jobId returns that job instead of + // throwing, so a retained failed entry (held by `removeOnFail: { count: 20 }`) + // made re-indexing that file a silent no-op: the caller got 202 "Indexing + // queued" and nothing was enqueued. In-flight jobs are still returned as-is + // so a re-click during an active embed stays idempotent. + if (!isContinuation && !force) { + const existing = await queue.getJob(initialJobId) + if (existing) { + const state = await existing.getState() + if (state === 'active' || state === 'waiting' || state === 'delayed') { + logger.info(`[EmbedFileJob] Job already in progress for file: ${params.fileName}`) + return { + job: existing, + created: false, + jobId: initialJobId, + message: `Embedding job already exists for: ${params.fileName}`, + } + } + try { + await existing.remove() + } catch { + // May already be gone + } + } + } + try { const job = await queue.add(this.key, params, jobOptions) diff --git a/admin/app/services/download_service.ts b/admin/app/services/download_service.ts index 7af5919..8302263 100644 --- a/admin/app/services/download_service.ts +++ b/admin/app/services/download_service.ts @@ -191,8 +191,13 @@ export class DownloadService { if (!modelName) { return { success: false, message: 'Cannot retry: model name not found in job data' } } - await DownloadModelJob.dispatch({ modelName }) + // Remove the old failed job first, then dispatch a fresh one. The + // model jobId is a hash of the model name, so dispatching first meant + // this remove() targeted the job we had just enqueued under that same + // id rather than the old record, deleting the retry we were creating. + // Matches the file-download branch below. await job.remove().catch(() => {}) + await DownloadModelJob.dispatch({ modelName }) return { success: true, message: `Retrying download for model ${modelName}` } }