From 88ac4d5ec4fd3dae0b9a5343382776db3bca54eb Mon Sep 17 00:00:00 2001 From: Jake Turner <52841588+jakeaturner@users.noreply.github.com> Date: Mon, 22 Jun 2026 21:19:49 -0700 Subject: [PATCH] feat(RAG): adds the ability to cancel all embedding jobs (#1034) --- admin/app/controllers/rag_controller.ts | 8 ++ admin/app/jobs/embed_file_job.ts | 56 ++++++++++++++ .../components/chat/KnowledgeBaseModal.tsx | 76 ++++++++++++++++--- admin/inertia/lib/api.ts | 7 ++ admin/start/routes.ts | 1 + 5 files changed, 137 insertions(+), 11 deletions(-) diff --git a/admin/app/controllers/rag_controller.ts b/admin/app/controllers/rag_controller.ts index 68cb5ae..7738d0f 100644 --- a/admin/app/controllers/rag_controller.ts +++ b/admin/app/controllers/rag_controller.ts @@ -110,6 +110,14 @@ export default class RagController { }) } + public async cancelAllJobs({ response }: HttpContext) { + const result = await EmbedFileJob.cancelAllJobs() + return response.status(200).json({ + message: `Cancelled ${result.cancelled} job${result.cancelled !== 1 ? 's' : ''}${result.filesDeleted > 0 ? `, deleted ${result.filesDeleted} file${result.filesDeleted !== 1 ? 's' : ''}` : ''}.`, + ...result, + }) + } + public async policyPromptState({ response }: HttpContext) { const result = await this.ragService.getPolicyPromptState() return response.status(200).json(result) diff --git a/admin/app/jobs/embed_file_job.ts b/admin/app/jobs/embed_file_job.ts index 93f58a9..ba52ca8 100644 --- a/admin/app/jobs/embed_file_job.ts +++ b/admin/app/jobs/embed_file_job.ts @@ -164,6 +164,22 @@ export class EmbedFileJob { await new Promise((resolve) => setTimeout(resolve, EmbedFileJob.CPU_BATCH_DELAY_MS)) } + // Bail before re-populating the queue if this job was cancelled mid-batch. + // cancelAllJobs() obliterates the queue (including this active job), but a + // worker already inside handle() would otherwise dispatch its continuation + // afterwards and silently revive a cancelled ZIM ingestion. If our own job + // key is gone, the cancel happened — skip the dispatch. Mirrors the + // "tolerate external removal" handling in safeUpdateProgress above. + const stillQueued = await QueueService.getInstance() + .getQueue(EmbedFileJob.queue) + .getJob(job.id!) + if (!stillQueued) { + logger.info( + `[EmbedFileJob] Job ${fileName} was cancelled; skipping continuation dispatch` + ) + return { success: false, cancelled: true, fileName, filePath } + } + // Dispatch next batch (not final yet). Carry forward the running // chunk count so the final batch can persist an accurate total (#933). const chunksSoFarNext = (job.data.chunksSoFar || 0) + (result.chunks || 0) @@ -443,6 +459,46 @@ export class EmbedFileJob { return { cleaned, filesDeleted } } + /** Unconditionally clear every embedding job regardless of state. + * + * cleanupFailedJobs only removes jobs explicitly tagged status === 'failed', + * which leaves stuck jobs (waiting / active / delayed / paused that never + * reached 'failed') unreachable from the UI — the operator's only recourse was + * flushing Redis by hand. This wipes the whole queue, including a locked active + * job, via obliterate({ force: true }) (plain obliterate/job.remove throw on a + * locked job). It touches only Redis, so it is safe while Qdrant/Ollama are + * offline — which is exactly when jobs pile up and wedge. */ + static async cancelAllJobs(): Promise<{ cancelled: number; filesDeleted: number }> { + const queueService = QueueService.getInstance() + const queue = queueService.getQueue(this.queue) + const jobs = await queue.getJobs(['waiting', 'active', 'delayed', 'paused', 'failed']) + + let filesDeleted = 0 + for (const job of jobs) { + const filePath = (job.data as EmbedFileJobParams).filePath + // Same guard as cleanupFailedJobs: only delete user uploads, never ZIM + // library files or Nomad docs that live outside the uploads path. + if (filePath && filePath.includes(RagService.UPLOADS_STORAGE_PATH)) { + try { + await fs.unlink(filePath) + filesDeleted++ + } catch { + // File may already be deleted — that's fine + } + } + } + + const cancelled = jobs.length + + // force: true removes the locked/active job too. An in-flight worker may keep + // running its current batch in memory; the self-exists guard in handle() + // prevents it from dispatching a continuation back into the cleared queue. + await queue.obliterate({ force: true }) + + logger.info(`[EmbedFileJob] Cancelled ${cancelled} jobs, deleted ${filesDeleted} files`) + return { cancelled, filesDeleted } + } + static async getStatus(filePath: string): Promise<{ exists: boolean status?: string diff --git a/admin/inertia/components/chat/KnowledgeBaseModal.tsx b/admin/inertia/components/chat/KnowledgeBaseModal.tsx index e9efbcf..cd47d04 100644 --- a/admin/inertia/components/chat/KnowledgeBaseModal.tsx +++ b/admin/inertia/components/chat/KnowledgeBaseModal.tsx @@ -262,6 +262,20 @@ export default function KnowledgeBaseModal({ aiAssistantName = "AI Assistant", o }, }) + const cancelAllMutation = useMutation({ + mutationFn: () => api.cancelAllEmbedJobs(), + onSuccess: (data) => { + addNotification({ type: 'success', message: data?.message || 'All embedding jobs cancelled.' }) + queryClient.invalidateQueries({ queryKey: ['embed-jobs'] }) + queryClient.invalidateQueries({ queryKey: ['failedEmbedJobs'] }) + queryClient.invalidateQueries({ queryKey: ['storedFiles'] }) + queryClient.invalidateQueries({ queryKey: ['kbFileWarnings'] }) + }, + onError: (error: any) => { + addNotification({ type: 'error', message: error?.message || 'Failed to cancel jobs.' }) + }, + }) + const startQdrantMutation = useMutation({ mutationFn: () => api.affectService(SERVICE_NAMES.QDRANT, 'start'), onSuccess: () => { @@ -358,6 +372,31 @@ export default function KnowledgeBaseModal({ aiAssistantName = "AI Assistant", o } } + const handleConfirmCancelAll = () => { + openModal( + { + cancelAllMutation.mutate() + closeModal('confirm-cancel-all-modal') + }} + onCancel={() => closeModal('confirm-cancel-all-modal')} + open={true} + confirmText='Cancel All Jobs' + cancelText='Keep Jobs' + confirmVariant='danger' + > +

+ This stops every embedding job — including ones still in progress or + stuck — and clears the processing queue. The uploaded source files for those jobs are + deleted, so you'll need to re-upload anything you still want indexed. Stored files that + already finished embedding are not affected. Are you sure you want to proceed? +

+ , + 'confirm-cancel-all-modal' + ) + } + const handleConfirmSync = () => { openModal(
-
+
- cleanupFailedMutation.mutate()} - loading={cleanupFailedMutation.isPending} - disabled={cleanupFailedMutation.isPending || qdrantOffline} - > - Clean Up Failed - +
+ cleanupFailedMutation.mutate()} + loading={cleanupFailedMutation.isPending} + disabled={cleanupFailedMutation.isPending || qdrantOffline} + > + Clean Up Failed + + {/* Not gated on qdrantOffline: clearing stuck jobs must work during + a Qdrant/Ollama outage, which is exactly when they wedge. */} + + Cancel All Jobs + +
diff --git a/admin/inertia/lib/api.ts b/admin/inertia/lib/api.ts index f15138b..9782e9e 100644 --- a/admin/inertia/lib/api.ts +++ b/admin/inertia/lib/api.ts @@ -479,6 +479,13 @@ class API { })() } + async cancelAllEmbedJobs(): Promise<{ message: string; cancelled: number; filesDeleted: number } | undefined> { + return catchInternal(async () => { + const response = await this.client.delete<{ message: string; cancelled: number; filesDeleted: number }>('/rag/jobs') + return response.data + })() + } + async checkRAGHealth() { return catchInternal(async () => { const response = await this.client.get<{ online: boolean; message?: string }>('/rag/health') diff --git a/admin/start/routes.ts b/admin/start/routes.ts index 6c19135..ca068cc 100644 --- a/admin/start/routes.ts +++ b/admin/start/routes.ts @@ -153,6 +153,7 @@ router router.get('/active-jobs', [RagController, 'getActiveJobs']) router.get('/failed-jobs', [RagController, 'getFailedJobs']) router.delete('/failed-jobs', [RagController, 'cleanupFailedJobs']) + router.delete('/jobs', [RagController, 'cancelAllJobs']) router.get('/job-status', [RagController, 'getJobStatus']) router.post('/sync', [RagController, 'scanAndSync']) router.post('/re-embed-all', [RagController, 'reembedAll'])