Merge 5c1ea23b81 into e6acf9938a

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<Job | undefined> {
+    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}`,
       }
     }
 

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)
 

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}` }
         }