From e01e00f6cba7124ca15be57b03c0aeabcff94d77 Mon Sep 17 00:00:00 2001 From: Finley Ge Date: Tue, 1 Sep 2026 17:29:52 +0800 Subject: [PATCH 1/3] feat(app-delete): resume cleanup with per-app jobs --- .../dal/redis/bullmq/services/appDelete.ts | 44 +++- .../dal/test/redis/bullmq-services.test.ts | 69 +++++- packages/service/core/app/delete/index.ts | 123 ++++++++++- packages/service/core/app/delete/processor.ts | 98 +++++---- .../test/core/app/delete/processor.test.ts | 203 ++++++++++++++++++ projects/app/test/api/core/app/delete.test.ts | 56 +++-- 6 files changed, 524 insertions(+), 69 deletions(-) create mode 100644 packages/service/test/core/app/delete/processor.test.ts diff --git a/packages/dal/redis/bullmq/services/appDelete.ts b/packages/dal/redis/bullmq/services/appDelete.ts index cf8817e1c400..1514d3d3507c 100644 --- a/packages/dal/redis/bullmq/services/appDelete.ts +++ b/packages/dal/redis/bullmq/services/appDelete.ts @@ -6,8 +6,14 @@ import type { Processor, Queue, Worker } from '../types'; export type AppDeleteJobData = { teamId: string; appId: string; + /** Historical jobs without this field are treated as root jobs. */ + jobType?: 'root' | 'app'; }; +type AppDeleteAppJobInput = Omit; + +const APP_DELETE_BULK_SIZE = 200; + const appDeleteQueueOptions = { defaultJobOptions: { attempts: 10, @@ -40,19 +46,53 @@ export class AppDeleteMQService { }); } - /** 投递幂等的 App 删除任务，并延迟一秒让请求先完成。 */ + /** Add a root deletion job and delay it by one second for request completion. */ addJob(data: AppDeleteJobData) { const jobId = `${String(data.teamId)}-${String(data.appId)}`; return addOrRequeueFailedJob({ queue: this.getQueue(), name: 'delete_app', - data, + data: { ...data, jobType: 'root' }, opts: { jobId, delay: 1000 } }); } + + /** Add one app deletion job with a stable ID for restart-safe idempotency. */ + addAppJob(data: AppDeleteAppJobInput) { + const jobId = `app-${String(data.teamId)}-${String(data.appId)}`; + return addOrRequeueFailedJob({ + queue: this.getQueue(), + name: 'delete_app', + data: { ...data, jobType: 'app' }, + opts: { + jobId + } + }); + } + + /** Add app deletion jobs in BullMQ batches to avoid one Redis round trip per app. */ + async addAppJobs(data: AppDeleteAppJobInput[]) { + if (data.length === 0) return []; + + const queue = this.getQueue(); + const jobs = data.map((item) => ({ + name: 'delete_app' as const, + data: { ...item, jobType: 'app' as const }, + opts: { + jobId: `app-${String(item.teamId)}-${String(item.appId)}` + } + })); + const results = []; + + for (let index = 0; index < jobs.length; index += APP_DELETE_BULK_SIZE) { + results.push(...(await queue.addBulk(jobs.slice(index, index + APP_DELETE_BULK_SIZE)))); + } + + return results; + } } export const appDeleteMQService = new AppDeleteMQService(); diff --git a/packages/dal/test/redis/bullmq-services.test.ts b/packages/dal/test/redis/bullmq-services.test.ts index c9b89c06f10c..c8bc92adbfa2 100644 --- a/packages/dal/test/redis/bullmq-services.test.ts +++ b/packages/dal/test/redis/bullmq-services.test.ts @@ -25,6 +25,7 @@ describe('BullMQ business services', () => { it('allows queue binding injection while keeping queue contracts in the service class', async () => { const queue = { add: vi.fn().mockResolvedValue({ id: 'job-1' }), + addBulk: vi.fn().mockResolvedValue([{ id: 'job-1' }]), getJob: vi.fn().mockResolvedValue(null) }; const binding = { @@ -47,13 +48,73 @@ describe('BullMQ business services', () => { removeOnFail: { age: 30 * 24 * 60 * 60 } } }); - expect(queue.add).toHaveBeenCalledWith('delete_app', data, { - jobId: 'team-1-app-1', - delay: 1000 - }); + expect(queue.add).toHaveBeenCalledWith( + 'delete_app', + { ...data, jobType: 'root' }, + { + jobId: 'team-1-app-1', + delay: 1000 + } + ); expect(queue.getJob).toHaveBeenCalledWith('team-1-app-1'); }); + it('uses a separate stable ID for a single app deletion job', async () => { + const queue = { + add: vi.fn().mockResolvedValue({ id: 'job-app-1' }), + getJob: vi.fn().mockResolvedValue(null) + }; + const binding = { + getQueue: vi.fn(() => queue), + getWorker: vi.fn(), + getLogger: vi.fn(() => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() })) + } as unknown as BullMQBinding; + const service = new AppDeleteMQService(binding); + + await expect(service.addAppJob({ teamId: 'team-1', appId: 'app-1' })).resolves.toEqual({ + id: 'job-app-1' + }); + + expect(queue.add).toHaveBeenCalledWith( + 'delete_app', + { teamId: 'team-1', appId: 'app-1', jobType: 'app' }, + { jobId: 'app-team-1-app-1' } + ); + expect(queue.getJob).toHaveBeenCalledWith('app-team-1-app-1'); + }); + + it('adds app deletion jobs in chunks of 200', async () => { + const queue = { + add: vi.fn(), + addBulk: vi.fn().mockImplementation(async (jobs) => jobs.map((job: unknown) => job)), + getJob: vi.fn().mockResolvedValue(null) + }; + const binding = { + getQueue: vi.fn(() => queue), + getWorker: vi.fn(), + getLogger: vi.fn(() => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() })) + } as unknown as BullMQBinding; + const service = new AppDeleteMQService(binding); + const data = Array.from({ length: 201 }, (_, index) => ({ + teamId: 'team-1', + appId: `app-${index}` + })); + + await service.addAppJobs(data); + + expect(queue.addBulk).toHaveBeenCalledTimes(2); + expect(queue.addBulk.mock.calls[0][0]).toHaveLength(200); + expect(queue.addBulk.mock.calls[1][0]).toHaveLength(1); + expect(queue.addBulk.mock.calls[0][0][0]).toEqual({ + name: 'delete_app', + data: { teamId: 'team-1', appId: 'app-0', jobType: 'app' }, + opts: { jobId: 'app-team-1-app-0' } + }); + expect(queue.addBulk.mock.calls[1][0][0].opts).toEqual({ + jobId: 'app-team-1-app-200' + }); + }); + it('uses failed-job recovery for Dataset deletion jobs', async () => { const queue = { add: vi.fn().mockResolvedValue({ id: 'job-2' }), diff --git a/packages/service/core/app/delete/index.ts b/packages/service/core/app/delete/index.ts index 05f4365e78bd..eafff2b16576 100644 --- a/packages/service/core/app/delete/index.ts +++ b/packages/service/core/app/delete/index.ts @@ -1,12 +1,129 @@ import { appDeleteProcessor } from './processor'; import { appDeleteMQService, type AppDeleteJobData } from '@fastgpt/dal/redis/bullmq'; +import { MongoApp } from '../schema'; +import { batchRunSettled } from '@fastgpt/global/common/system/utils'; +import { getLogger, LogCategories } from '../../../common/logger'; +import { setCron } from '../../../common/system/cron'; export type { AppDeleteJobData } from '@fastgpt/dal/redis/bullmq'; -// 创建工作进程 +const APP_DELETE_RESUME_BATCH_SIZE = 200; +const APP_DELETE_RESUME_CONCURRENCY = 5; +const APP_DELETE_RESUME_CRON = '*/5 * * * *'; +const logger = getLogger(LogCategories.MODULE.APP.FOLDER); + +let recoveryCronRegistered = false; +let recoveryPromise: Promise | undefined; + +/** + * Initialize the app deletion worker and asynchronously resume soft-deleted apps + * whose cleanup was not completed. Recovery does not block worker creation, and + * one failed enqueue does not prevent other apps in the same batch from proceeding. + */ export const initAppDeleteWorker = () => { - return appDeleteMQService.getWorker(appDeleteProcessor); + const worker = appDeleteMQService.getWorker(appDeleteProcessor); + + registerAppDeleteRecoveryCron(); + + resumeMarkedAppDeleteJobs().catch((error) => { + logger.error('Failed to resume marked app delete jobs', { error }); + }); + + return worker; }; -// 添加删除任务 +/** + * Keep recovery alive after transient Redis or Mongo failures during startup. + * Stable app job IDs make repeated scans idempotent across workers and pods. + */ +const registerAppDeleteRecoveryCron = () => { + if (recoveryCronRegistered) return; + + recoveryCronRegistered = true; + setCron(APP_DELETE_RESUME_CRON, () => { + resumeMarkedAppDeleteJobs().catch((error) => { + logger.error('Failed to resume marked app delete jobs', { error }); + }); + }); +}; + +/** + * Scan soft-deleted apps and resume their single-app deletion jobs. + * Cursor pagination and bounded concurrency prevent startup from saturating Mongo or Redis. + */ +async function resumeMarkedAppDeleteJobsInternal(): Promise { + const cursor = MongoApp.find( + { + deleteTime: { + $exists: true, + $ne: null + } + }, + { + _id: 1, + teamId: 1 + } + ) + .lean() + .cursor({ batchSize: APP_DELETE_RESUME_BATCH_SIZE }); + + let totalMarked = 0; + let resumedCount = 0; + let failedCount = 0; + let batch: { teamId: string; appId: string }[] = []; + + /** Flush one bounded batch so startup recovery does not overload Redis or Mongo. */ + const flushBatch = async () => { + if (batch.length === 0) return; + + const currentBatch = batch; + batch = []; + const results = await batchRunSettled( + currentBatch, + (item) => appDeleteMQService.addAppJob(item), + APP_DELETE_RESUME_CONCURRENCY + ); + resumedCount += results.filter((result) => result.success).length; + failedCount += results.filter((result) => !result.success).length; + }; + + for await (const app of cursor) { + totalMarked += 1; + batch.push({ + teamId: String(app.teamId), + appId: String(app._id) + }); + + if (batch.length >= APP_DELETE_RESUME_BATCH_SIZE) { + await flushBatch(); + } + } + + await flushBatch(); + + logger.info('Marked app delete jobs resumed', { + totalMarked, + resumedCount, + failedCount + }); +} + +/** + * Resume marked app cleanup with process-local overlap protection. + * Multiple callers share one scan while distributed callers converge on stable job IDs. + */ +export function resumeMarkedAppDeleteJobs(): Promise { + if (recoveryPromise) return recoveryPromise; + + recoveryPromise = resumeMarkedAppDeleteJobsInternal().finally(() => { + recoveryPromise = undefined; + }); + return recoveryPromise; +} + +/** Add a root app deletion job. */ export const addAppDeleteJob = (data: AppDeleteJobData) => appDeleteMQService.addJob(data); + +/** Add a single-app deletion job for startup recovery and other recovery flows. */ +export const addAppDeleteAppJob = (data: Omit) => + appDeleteMQService.addAppJob(data); diff --git a/packages/service/core/app/delete/processor.ts b/packages/service/core/app/delete/processor.ts index c19cecfbe1b6..52315ff2dc47 100644 --- a/packages/service/core/app/delete/processor.ts +++ b/packages/service/core/app/delete/processor.ts @@ -1,37 +1,68 @@ import type { Processor } from '@fastgpt/dal/redis/bullmq'; import type { AppDeleteJobData } from './index'; +import { appDeleteMQService } from '@fastgpt/dal/redis/bullmq'; import { findAppAndAllChildren, deleteAppDataProcessor } from '../controller'; -import { batchRun } from '@fastgpt/global/common/system/utils'; -import type { AppSchemaType } from '@fastgpt/global/core/app/type'; import { MongoApp } from '../schema'; import { getLogger, LogCategories } from '../../../common/logger'; const logger = getLogger(LogCategories.MODULE.APP.FOLDER); -const deleteApps = async ({ teamId, apps }: { teamId: string; apps: AppSchemaType[] }) => { - // 每个 App 使用独立 Source Lease；任务内串行用于限制外部资源清理压力。 - const results = await batchRun( - apps, - async (app) => { - await deleteAppDataProcessor({ app, teamId }); +/** + * Clean one app using only the fields required by the existing deletion processor. + * Missing apps are treated as an idempotent completion; live apps fail the safety check. + */ +const deleteSingleApp = async ({ teamId, appId }: { teamId: string; appId: string }) => { + const startTime = Date.now(); + const app = await MongoApp.findOne( + { + _id: appId, + teamId }, - 1 - ); + '_id teamId type avatar deleteTime' + ).lean(); + + if (!app) { + logger.warn('App not found for deletion', { teamId, appId }); + return; + } + + // Recheck the soft-delete marker so stale jobs cannot remove a live app. + if (!app.deleteTime) { + logger.warn('App delete safety check mismatch', { + teamId, + appId, + markedCount: 0, + totalCount: 1 + }); + throw new Error('App delete safety check mismatch'); + } - return results.flat(); + await deleteAppDataProcessor({ app, teamId }); + + logger.info('App delete completed', { + teamId, + appId, + durationMs: Date.now() - startTime + }); }; export const appDeleteProcessor: Processor = async (job) => { - const { teamId, appId } = job.data; + const { teamId, appId, jobType = 'root' } = job.data; const startTime = Date.now(); logger.info('App delete started', { teamId, appId }); try { - // 1. 查找应用及其所有子应用 + if (jobType === 'app') { + await deleteSingleApp({ teamId, appId }); + return; + } + + // 1. Find the app subtree using only fields needed to create child jobs. const apps = await findAppAndAllChildren({ teamId, - appId + appId, + fields: '_id teamId parentId deleteTime' }); if (!apps || apps.length === 0) { @@ -39,42 +70,31 @@ export const appDeleteProcessor: Processor = async (job) => { return; } - // 2. 安全检查：确保所有要删除的应用都已标记为 deleteTime - const markedForDelete = await MongoApp.find( - { - _id: { $in: apps.map((app) => app._id) }, - teamId, - deleteTime: { $ne: null } - }, - { _id: 1 } - ).lean(); - - if (markedForDelete.length !== apps.length) { + // 2. The root task only splits work, but the complete subtree must be soft-deleted. + const unmarkedApps = apps.filter((app) => !app.deleteTime); + if (unmarkedApps.length > 0) { logger.warn('App delete safety check mismatch', { - markedCount: markedForDelete.length, + markedCount: apps.length - unmarkedApps.length, totalCount: apps.length, - markedAppIds: markedForDelete.map((app) => app._id), - totalAppIds: apps.map((app) => app._id) + unmarkedCount: unmarkedApps.length }); throw new Error('App delete safety check mismatch'); } - const childrenLen = apps.length - 1; - const appIds = apps.map((app) => app._id); - - // 3. 执行真正的删除操作（只删除已经标记为 deleteTime 的数据） - await deleteApps({ - teamId, - apps - }); + // 3. Split the work without cleaning any app resources in this job. + await appDeleteMQService.addAppJobs( + apps.map((app) => ({ + teamId, + appId: String(app._id) + })) + ); logger.info('App delete completed', { teamId, appId, - childCount: childrenLen, + childCount: apps.length - 1, durationMs: Date.now() - startTime, - totalApps: appIds.length, - appIds + totalApps: apps.length }); } catch (error: any) { logger.error('App delete failed', { teamId, appId, error }); diff --git a/packages/service/test/core/app/delete/processor.test.ts b/packages/service/test/core/app/delete/processor.test.ts new file mode 100644 index 000000000000..6e104c30b787 --- /dev/null +++ b/packages/service/test/core/app/delete/processor.test.ts @@ -0,0 +1,203 @@ +import { beforeEach, describe, expect, it, vi } from 'vitest'; +import type { Job } from '@fastgpt/dal/redis/bullmq'; +import type { AppDeleteJobData } from '@fastgpt/dal/redis/bullmq'; + +const mocks = vi.hoisted(() => ({ + findAppAndAllChildren: vi.fn(), + deleteAppDataProcessor: vi.fn(), + findOne: vi.fn(), + find: vi.fn(), + addAppJobs: vi.fn(), + addAppJob: vi.fn(), + getWorker: vi.fn(), + setCron: vi.fn(), + logger: { + info: vi.fn(), + warn: vi.fn(), + error: vi.fn() + } +})); + +vi.mock('@fastgpt/service/core/app/controller', () => ({ + findAppAndAllChildren: mocks.findAppAndAllChildren, + deleteAppDataProcessor: mocks.deleteAppDataProcessor +})); + +vi.mock('@fastgpt/service/core/app/schema', () => ({ + MongoApp: { + findOne: mocks.findOne, + find: mocks.find + } +})); + +vi.mock('@fastgpt/service/common/logger', () => ({ + getLogger: () => mocks.logger, + LogCategories: { MODULE: { APP: { FOLDER: 'app.folder' } } } +})); + +vi.mock('@fastgpt/service/common/system/cron', () => ({ + setCron: mocks.setCron +})); + +vi.mock('@fastgpt/dal/redis/bullmq', () => ({ + appDeleteMQService: { + addAppJobs: mocks.addAppJobs, + addAppJob: mocks.addAppJob, + getWorker: mocks.getWorker + } +})); + +import { appDeleteProcessor } from '../../../../core/app/delete/processor'; +import { initAppDeleteWorker, resumeMarkedAppDeleteJobs } from '../../../../core/app/delete'; + +const createJob = (id: string, data: AppDeleteJobData) => + ({ id, data }) as unknown as Job; + +describe('appDeleteProcessor', () => { + beforeEach(() => { + vi.clearAllMocks(); + mocks.findAppAndAllChildren.mockResolvedValue([ + { _id: 'root', teamId: 'team-1', parentId: null, deleteTime: new Date() }, + { _id: 'child-1', teamId: 'team-1', parentId: 'root', deleteTime: new Date() }, + { _id: 'child-2', teamId: 'team-1', parentId: 'root', deleteTime: new Date() } + ]); + mocks.addAppJobs.mockResolvedValue([]); + mocks.addAppJob.mockResolvedValue({ id: 'app-delete-job' }); + mocks.findOne.mockReturnValue({ + lean: vi.fn().mockResolvedValue({ + _id: 'app-1', + teamId: 'team-1', + type: 'simple', + avatar: '', + deleteTime: new Date() + }) + }); + mocks.deleteAppDataProcessor.mockResolvedValue(undefined); + }); + + it('splits a root deletion job into single-app jobs without cleaning app data', async () => { + await appDeleteProcessor( + createJob('root-job', { teamId: 'team-1', appId: 'root', jobType: 'root' }) + ); + + expect(mocks.findAppAndAllChildren).toHaveBeenCalledWith({ + teamId: 'team-1', + appId: 'root', + fields: '_id teamId parentId deleteTime' + }); + expect(mocks.addAppJobs).toHaveBeenCalledWith([ + { teamId: 'team-1', appId: 'root' }, + { teamId: 'team-1', appId: 'child-1' }, + { teamId: 'team-1', appId: 'child-2' } + ]); + expect(mocks.findOne).not.toHaveBeenCalled(); + expect(mocks.deleteAppDataProcessor).not.toHaveBeenCalled(); + }); + + it('treats a historical job without jobType as a root job', async () => { + await appDeleteProcessor(createJob('legacy-root-job', { teamId: 'team-1', appId: 'root' })); + + expect(mocks.addAppJobs).toHaveBeenCalledTimes(1); + expect(mocks.deleteAppDataProcessor).not.toHaveBeenCalled(); + }); + + it('does not log all subtree IDs when the root safety check fails', async () => { + mocks.findAppAndAllChildren.mockResolvedValue([ + { _id: 'root', teamId: 'team-1', parentId: null, deleteTime: null }, + { _id: 'child-1', teamId: 'team-1', parentId: 'root', deleteTime: null } + ]); + + await expect( + appDeleteProcessor( + createJob('unsafe-root-job', { teamId: 'team-1', appId: 'root', jobType: 'root' }) + ) + ).rejects.toThrow('App delete safety check mismatch'); + + expect(mocks.logger.warn).toHaveBeenCalledWith('App delete safety check mismatch', { + markedCount: 0, + totalCount: 2, + unmarkedCount: 2 + }); + expect(mocks.logger.warn.mock.calls[0][1]).not.toHaveProperty('unmarkedAppIds'); + }); + + it('cleans exactly one marked app for a single-app job', async () => { + await appDeleteProcessor( + createJob('app-job', { teamId: 'team-1', appId: 'app-1', jobType: 'app' }) + ); + + expect(mocks.findOne).toHaveBeenCalledWith( + { _id: 'app-1', teamId: 'team-1' }, + '_id teamId type avatar deleteTime' + ); + expect(mocks.deleteAppDataProcessor).toHaveBeenCalledWith({ + app: expect.objectContaining({ _id: 'app-1' }), + teamId: 'team-1' + }); + expect(mocks.findAppAndAllChildren).not.toHaveBeenCalled(); + }); + + it('rejects an unmarked single-app job before deleting external data', async () => { + mocks.findOne.mockReturnValue({ + lean: vi.fn().mockResolvedValue({ + _id: 'app-1', + teamId: 'team-1', + type: 'simple', + avatar: '', + deleteTime: null + }) + }); + + await expect( + appDeleteProcessor( + createJob('unmarked-app-job', { teamId: 'team-1', appId: 'app-1', jobType: 'app' }) + ) + ).rejects.toThrow('App delete safety check mismatch'); + expect(mocks.deleteAppDataProcessor).not.toHaveBeenCalled(); + }); + + it('resumes marked apps through a cursor with bounded batch processing', async () => { + const cursor = (async function* () { + yield { _id: 'app-1', teamId: 'team-1' }; + yield { _id: 'app-2', teamId: 'team-2' }; + })(); + const lean = vi.fn().mockReturnValue({ + cursor: vi.fn().mockReturnValue(cursor) + }); + mocks.find.mockReturnValue({ lean }); + + await resumeMarkedAppDeleteJobs(); + + expect(mocks.find).toHaveBeenCalledWith( + { deleteTime: { $exists: true, $ne: null } }, + { _id: 1, teamId: 1 } + ); + expect(mocks.addAppJob).toHaveBeenNthCalledWith(1, { + teamId: 'team-1', + appId: 'app-1' + }); + expect(mocks.addAppJob).toHaveBeenNthCalledWith(2, { + teamId: 'team-2', + appId: 'app-2' + }); + expect(mocks.logger.info).toHaveBeenCalledWith('Marked app delete jobs resumed', { + totalMarked: 2, + resumedCount: 2, + failedCount: 0 + }); + }); + + it('registers periodic recovery when the worker starts', async () => { + const cursor = (async function* () {})(); + mocks.find.mockReturnValue({ + lean: vi.fn().mockReturnValue({ cursor: vi.fn().mockReturnValue(cursor) }) + }); + const worker = { name: 'app-delete-worker' }; + mocks.getWorker.mockReturnValue(worker); + + expect(initAppDeleteWorker()).toBe(worker); + expect(mocks.setCron).toHaveBeenCalledWith('*/5 * * * *', expect.any(Function)); + + await resumeMarkedAppDeleteJobs(); + }); +}); diff --git a/projects/app/test/api/core/app/delete.test.ts b/projects/app/test/api/core/app/delete.test.ts index c9b6434db4f8..9c8cb9e24fc5 100644 --- a/projects/app/test/api/core/app/delete.test.ts +++ b/projects/app/test/api/core/app/delete.test.ts @@ -47,10 +47,16 @@ vi.mock('@fastgpt/dal/redis/bullmq', () => { removeOnFail: { age: 30 * 24 * 60 * 60 } } }) - .add('delete_app', data, { - jobId: `${data.teamId}-${data.appId}`, - delay: 1000 - }) + .add( + 'delete_app', + { ...data, jobType: 'root' }, + { + jobId: `${data.teamId}-${data.appId}`, + delay: 1000 + } + ), + addAppJobs: vi.fn().mockResolvedValue([]), + addAppJob: vi.fn().mockResolvedValue({ id: 'app-delete-job' }) }, QueueNames: { appDelete: 'appDelete' @@ -109,10 +115,14 @@ describe('App Delete Queue', () => { } }); - expect(mockQueue.add).toHaveBeenCalledWith('delete_app', jobData, { - jobId: 'team-123-app-123', - delay: 1000 - }); + expect(mockQueue.add).toHaveBeenCalledWith( + 'delete_app', + { ...jobData, jobType: 'root' }, + { + jobId: 'team-123-app-123', + delay: 1000 + } + ); expect(result).toEqual({ id: 'job-123' }); }); @@ -132,7 +142,7 @@ describe('App Delete Queue', () => { expect(mockQueue.add).toHaveBeenCalledWith( 'delete_app', - jobData, + { ...jobData, jobType: 'root' }, expect.objectContaining({ jobId: 'team-xyz-app-abc' }) @@ -183,7 +193,8 @@ describe('App Delete API Integration', () => { 'delete_app', { teamId: rootUser.teamId, - appId: String(testApp._id) + appId: String(testApp._id), + jobType: 'root' }, { jobId: `${rootUser.teamId}-${testApp._id}`, @@ -345,7 +356,7 @@ describe('App Delete Data Cleanup Verification', () => { // 3. 执行删除处理器（模拟队列任务执行） const mockJob = { - data: { teamId, appId }, + data: { teamId, appId, jobType: 'app' }, id: 'test-job-id' }; @@ -387,13 +398,18 @@ describe('App Delete Data Cleanup Verification', () => { { deleteTime: new Date() } ); - // 执行删除（应该级联删除子应用） - const mockJob = { - data: { teamId, appId: String(parentApp._id) }, + // Each app is cleaned by its own queue job after the root job is split. + const parentJob = { + data: { teamId, appId: String(parentApp._id), jobType: 'app' as const }, id: 'test-nested-job' }; + const childJob = { + data: { teamId, appId: String(childApp._id), jobType: 'app' as const }, + id: 'test-nested-child-job' + }; - await appDeleteProcessor(mockJob); + await appDeleteProcessor(parentJob); + await appDeleteProcessor(childJob); // 验证父应用和子应用都被删除 expect(await MongoApp.countDocuments({ _id: parentApp._id })).toBe(0); @@ -409,7 +425,7 @@ describe('App Delete Data Cleanup Verification', () => { }); }); - it('should reject external cleanup when a nested app is not marked for deletion', async () => { + it('should reject external cleanup when the target app is not marked for deletion', async () => { const parentApp = await MongoApp.create({ name: 'Safety Parent App', teamId, @@ -425,11 +441,9 @@ describe('App Delete Data Cleanup Verification', () => { parentId: parentApp._id, modules: [] }); - await MongoApp.updateOne({ _id: parentApp._id }, { deleteTime: new Date() }); - await expect( appDeleteProcessor({ - data: { teamId, appId: String(parentApp._id) }, + data: { teamId, appId: String(parentApp._id), jobType: 'app' }, id: 'test-delete-safety-job' }) ).rejects.toThrow('App delete safety check mismatch'); @@ -457,7 +471,7 @@ describe('App Delete Data Cleanup Verification', () => { // 删除第一个应用 const mockJob1 = { - data: { teamId, appId }, + data: { teamId, appId, jobType: 'app' }, id: 'test-batch-job-1' }; @@ -498,7 +512,7 @@ describe('App Delete Data Cleanup Verification', () => { await MongoApp.updateOne({ _id: workflowToolAppId }, { deleteTime: new Date() }); await appDeleteProcessor({ - data: { teamId, appId: workflowToolAppId }, + data: { teamId, appId: workflowToolAppId, jobType: 'app' }, id: 'test-workflow-tool-cleanup-job' }); From 035dab8ac858f6fbd6346c3ab8fecb58a8239fba Mon Sep 17 00:00:00 2001 From: Finley Ge Date: Tue, 1 Sep 2026 18:59:13 +0800 Subject: [PATCH 2/3] feat(app-delete): enqueue ordered task flows --- packages/dal/redis/bullmq/binding.ts | 7 +- .../dal/redis/bullmq/flow-producer-manager.ts | 66 +++++ packages/dal/redis/bullmq/index.ts | 4 + packages/dal/redis/bullmq/runtime.ts | 22 +- .../dal/redis/bullmq/services/appDelete.ts | 63 ++--- packages/dal/redis/bullmq/types.ts | 6 + .../dal/test/redis/bullmq-binding.test.ts | 9 + .../dal/test/redis/bullmq-services.test.ts | 87 +++--- packages/dal/test/redis/bullmq.test.ts | 52 +++- packages/service/core/app/delete/index.ts | 263 ++++++++++++++---- packages/service/core/app/delete/processor.ts | 47 +--- .../service/support/user/team/delete/utils.ts | 14 +- .../test/core/app/delete/processor.test.ts | 130 +++++++-- projects/app/src/pages/api/core/app/del.ts | 9 +- projects/app/test/api/core/app/delete.test.ts | 150 ++++------ 15 files changed, 622 insertions(+), 307 deletions(-) create mode 100644 packages/dal/redis/bullmq/flow-producer-manager.ts diff --git a/packages/dal/redis/bullmq/binding.ts b/packages/dal/redis/bullmq/binding.ts index ebcda41371b0..a05a2555e0cf 100644 --- a/packages/dal/redis/bullmq/binding.ts +++ b/packages/dal/redis/bullmq/binding.ts @@ -1,7 +1,7 @@ import { getRedisRuntime } from '../runtime'; import { getRedisBullMQRuntime } from './context'; import type { QueueNames } from './names'; -import type { Processor, Queue, QueueOptions, Worker, WorkerOptions } from './types'; +import type { FlowProducer, Processor, Queue, QueueOptions, Worker, WorkerOptions } from './types'; const defaultWorkerOpts: Omit = { removeOnComplete: { @@ -59,6 +59,11 @@ export class BullMQBinding { ...opts }); } + + /** Returns the Runtime-owned FlowProducer for atomically creating dependent job trees. */ + getFlowProducer(): FlowProducer { + return this.getRuntime().getFlowProducer(); + } } /** 进程级 BullMQ 绑定。 */ diff --git a/packages/dal/redis/bullmq/flow-producer-manager.ts b/packages/dal/redis/bullmq/flow-producer-manager.ts new file mode 100644 index 000000000000..ec852cd890ae --- /dev/null +++ b/packages/dal/redis/bullmq/flow-producer-manager.ts @@ -0,0 +1,66 @@ +import { FlowProducer } from 'bullmq'; +import type { RedisRuntime } from '../runtime/connection'; +import type { RedisRuntimeLogger } from '../types'; +import { closeWithTimeout, forceDisconnect } from './close'; +import type { BullMQDisconnectable } from './types'; + +/** Manages the shared FlowProducer lifecycle for a BullMQ Runtime. */ +export class BullMQFlowProducerManager { + private flowProducer: FlowProducer | undefined; + + constructor( + private readonly options: { + redisRuntime: RedisRuntime; + logger: RedisRuntimeLogger; + closeTimeoutMs: number; + } + ) {} + + /** Creates one FlowProducer and reuses it for all flows in this Runtime. */ + getFlowProducer() { + if (this.flowProducer) return this.flowProducer; + + const connection = this.options.redisRuntime.createQueueConnection(); + try { + const flowProducer = new FlowProducer({ connection }); + const errorHandler = (error: Error) => { + this.options.logger.error('BullMQ flow producer error', { error }); + }; + flowProducer.on('error', errorHandler); + this.flowProducer = flowProducer; + return flowProducer; + } catch (error) { + void this.options.redisRuntime.releaseConnection(connection).catch((releaseError) => { + this.options.logger.warn( + 'Failed to release Redis connection after FlowProducer creation error', + { error: releaseError } + ); + }); + throw error; + } + } + + async close() { + const flowProducer = this.flowProducer; + this.flowProducer = undefined; + if (!flowProducer) return; + + await closeWithTimeout({ + operation: () => flowProducer.close(), + resource: 'BullMQ flow producer', + timeoutMs: this.options.closeTimeoutMs + }).catch((error) => { + this.options.logger.warn('BullMQ flow producer close failed', { error }); + const connection = (flowProducer as unknown as { connection?: BullMQDisconnectable }) + .connection; + forceDisconnect({ + name: 'flow-producer', + resource: 'flow producer connection', + disconnect: connection + ? () => connection.disconnect(false) + : () => flowProducer.disconnect(), + logger: this.options.logger + }); + }); + } +} diff --git a/packages/dal/redis/bullmq/index.ts b/packages/dal/redis/bullmq/index.ts index 4854bd99ce6b..ff86db9cc47c 100644 --- a/packages/dal/redis/bullmq/index.ts +++ b/packages/dal/redis/bullmq/index.ts @@ -2,6 +2,7 @@ export { bullMQ, BullMQBinding } from './binding'; export { getConfiguredRedisBullMQRuntime, getRedisBullMQRuntime } from './context'; export { QueueNames } from './names'; export { RedisBullMQRuntime } from './runtime'; +export { BullMQFlowProducerManager } from './flow-producer-manager'; export { addOrRequeueFailedJob } from './job-recovery'; export { DelayedError, UnrecoverableError } from 'bullmq'; export * from './services'; @@ -9,7 +10,10 @@ export type { BullMQRuntimeState, BullMQWorkerLifecycleOptions, ConnectionOptions, + FlowJob, + FlowProducer, Job, + JobNode, JobSchedulerJson, Processor, Queue, diff --git a/packages/dal/redis/bullmq/runtime.ts b/packages/dal/redis/bullmq/runtime.ts index 618d22c77f75..2cc2480557c9 100644 --- a/packages/dal/redis/bullmq/runtime.ts +++ b/packages/dal/redis/bullmq/runtime.ts @@ -1,4 +1,4 @@ -import type { Queue, Processor, QueueOptions, Worker, WorkerOptions } from 'bullmq'; +import type { FlowProducer, Processor, Queue, QueueOptions, Worker, WorkerOptions } from 'bullmq'; import type { RedisRuntime } from '../runtime/connection'; import { DEFAULT_CLOSE_TIMEOUT_MS, @@ -6,6 +6,7 @@ import { silentBullMQLogger } from './constants'; import { BullMQQueueManager } from './queue-manager'; +import { BullMQFlowProducerManager } from './flow-producer-manager'; import type { BullMQRuntimeState, BullMQWorkerLifecycleOptions, @@ -20,6 +21,7 @@ export class RedisBullMQRuntime { private readonly logger: RedisRuntimeLogger; private readonly queueManager: BullMQQueueManager; + private readonly flowProducerManager: BullMQFlowProducerManager; private readonly workerManager: BullMQWorkerManager; private readonly unregisterBeforeCloseHook: () => void; private state: BullMQRuntimeState = 'running'; @@ -44,6 +46,11 @@ export class RedisBullMQRuntime { logger, closeTimeoutMs }); + this.flowProducerManager = new BullMQFlowProducerManager({ + redisRuntime, + logger, + closeTimeoutMs + }); this.workerManager = new BullMQWorkerManager({ redisRuntime, logger, @@ -83,6 +90,12 @@ export class RedisBullMQRuntime { return this.workerManager.getWorker(name, processor, opts); } + /** Returns the Runtime-shared FlowProducer, which owns a dedicated queue connection. */ + getFlowProducer(): FlowProducer { + this.assertRunning(); + return this.flowProducerManager.getFlowProducer(); + } + close() { if (this.closePromise) return this.closePromise; @@ -101,6 +114,13 @@ export class RedisBullMQRuntime { } // Worker 关闭失败也不能跳过 Queue，否则队列连接会被 Redis Runtime 强制回收。 + try { + await this.flowProducerManager.close(); + } catch (error) { + firstError = error; + hasError = true; + } + try { await this.queueManager.close(); } catch (error) { diff --git a/packages/dal/redis/bullmq/services/appDelete.ts b/packages/dal/redis/bullmq/services/appDelete.ts index 1514d3d3507c..e964bc1dbd82 100644 --- a/packages/dal/redis/bullmq/services/appDelete.ts +++ b/packages/dal/redis/bullmq/services/appDelete.ts @@ -1,19 +1,16 @@ import { bullMQ, type BullMQBinding } from '../binding'; import { addOrRequeueFailedJob } from '../job-recovery'; import { QueueNames } from '../names'; -import type { Processor, Queue, Worker } from '../types'; +import type { FlowJob, JobNode, Processor, Queue, Worker } from '../types'; export type AppDeleteJobData = { teamId: string; appId: string; - /** Historical jobs without this field are treated as root jobs. */ - jobType?: 'root' | 'app'; + /** Distinguishes task roots, internal Flow steps, and pre-Flow jobs. */ + jobType?: 'task' | 'step' | 'root' | 'app'; + taskId?: string; }; -type AppDeleteAppJobInput = Omit; - -const APP_DELETE_BULK_SIZE = 200; - const appDeleteQueueOptions = { defaultJobOptions: { attempts: 10, @@ -26,6 +23,22 @@ const appDeleteQueueOptions = { } }; +const appDeleteFlowJobOptions = appDeleteQueueOptions.defaultJobOptions; + +/** Apply the App deletion queue retry and retention policy to every node in a Flow tree. */ +const applyFlowJobOptions = (flow: FlowJob): FlowJob => ({ + ...flow, + opts: { + ...appDeleteFlowJobOptions, + ...flow.opts + }, + ...(flow.children + ? { + children: flow.children.map((child) => applyFlowJobOptions(child)) + } + : {}) +}); + /** App 删除队列的业务合同和生命周期入口。 */ export class AppDeleteMQService { constructor(private readonly binding: BullMQBinding = bullMQ) {} @@ -60,38 +73,16 @@ export class AppDeleteMQService { }); } - /** Add one app deletion job with a stable ID for restart-safe idempotency. */ - addAppJob(data: AppDeleteAppJobInput) { - const jobId = `app-${String(data.teamId)}-${String(data.appId)}`; - return addOrRequeueFailedJob({ - queue: this.getQueue(), - name: 'delete_app', - data: { ...data, jobType: 'app' }, - opts: { - jobId - } - }); + /** Returns the FlowProducer used to atomically submit deletion task Flows. */ + getFlowProducer() { + return this.binding.getFlowProducer(); } - /** Add app deletion jobs in BullMQ batches to avoid one Redis round trip per app. */ - async addAppJobs(data: AppDeleteAppJobInput[]) { - if (data.length === 0) return []; - - const queue = this.getQueue(); - const jobs = data.map((item) => ({ - name: 'delete_app' as const, - data: { ...item, jobType: 'app' as const }, - opts: { - jobId: `app-${String(item.teamId)}-${String(item.appId)}` - } - })); - const results = []; - - for (let index = 0; index < jobs.length; index += APP_DELETE_BULK_SIZE) { - results.push(...(await queue.addBulk(jobs.slice(index, index + APP_DELETE_BULK_SIZE)))); - } + /** Atomically submits task Flows; callers define step order through the dependency chain. */ + addFlows(flows: FlowJob[]): Promise { + if (flows.length === 0) return Promise.resolve([]); - return results; + return this.getFlowProducer().addBulk(flows.map((flow) => applyFlowJobOptions(flow))); } } diff --git a/packages/dal/redis/bullmq/types.ts b/packages/dal/redis/bullmq/types.ts index 2f6b07b91d92..2e8ea13d5a82 100644 --- a/packages/dal/redis/bullmq/types.ts +++ b/packages/dal/redis/bullmq/types.ts @@ -1,6 +1,9 @@ import type { ConnectionOptions, + FlowJob, + FlowProducer, Job, + JobNode, JobSchedulerJson, Processor, Queue, @@ -42,7 +45,10 @@ export type WorkerListenerSnapshot = { export type { ConnectionOptions, + FlowJob, + FlowProducer, Job, + JobNode, JobSchedulerJson, Processor, Queue, diff --git a/packages/dal/test/redis/bullmq-binding.test.ts b/packages/dal/test/redis/bullmq-binding.test.ts index b7c06477ed50..10c4b216d9f0 100644 --- a/packages/dal/test/redis/bullmq-binding.test.ts +++ b/packages/dal/test/redis/bullmq-binding.test.ts @@ -4,6 +4,7 @@ vi.unmock('@fastgpt/dal/redis/bullmq'); const bullMQMocks = vi.hoisted(() => { const queue = { id: 'queue' }; + const flowProducer = { id: 'flow-producer' }; const worker = { id: 'worker' }; const redisRuntime = { id: 'redis-runtime', @@ -11,12 +12,14 @@ const bullMQMocks = vi.hoisted(() => { }; const runtime = { getQueue: vi.fn(() => queue), + getFlowProducer: vi.fn(() => flowProducer), getWorker: vi.fn(() => worker), getLogger: vi.fn(() => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() })) }; return { queue, + flowProducer, worker, redisRuntime, runtime, @@ -38,6 +41,7 @@ describe('DAL BullMQ binding', () => { beforeEach(async () => { vi.resetModules(); bullMQMocks.runtime.getQueue.mockClear(); + bullMQMocks.runtime.getFlowProducer.mockClear(); bullMQMocks.runtime.getWorker.mockClear(); bullMQMocks.getRedisBullMQRuntime.mockClear(); bullMQ = await import('@fastgpt/dal/redis/bullmq'); @@ -78,4 +82,9 @@ describe('DAL BullMQ binding', () => { }) ); }); + + it('delegates FlowProducer access to the DAL BullMQ Runtime', () => { + expect(bullMQ.bullMQ.getFlowProducer()).toBe(bullMQMocks.flowProducer); + expect(bullMQMocks.runtime.getFlowProducer).toHaveBeenCalledTimes(1); + }); }); diff --git a/packages/dal/test/redis/bullmq-services.test.ts b/packages/dal/test/redis/bullmq-services.test.ts index c8bc92adbfa2..1ef007eeccc0 100644 --- a/packages/dal/test/redis/bullmq-services.test.ts +++ b/packages/dal/test/redis/bullmq-services.test.ts @@ -31,6 +31,7 @@ describe('BullMQ business services', () => { const binding = { getQueue: vi.fn(() => queue), getWorker: vi.fn(), + getFlowProducer: vi.fn(), getLogger: vi.fn(() => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() })) } as unknown as BullMQBinding; const service = new AppDeleteMQService(binding); @@ -59,60 +60,54 @@ describe('BullMQ business services', () => { expect(queue.getJob).toHaveBeenCalledWith('team-1-app-1'); }); - it('uses a separate stable ID for a single app deletion job', async () => { - const queue = { - add: vi.fn().mockResolvedValue({ id: 'job-app-1' }), - getJob: vi.fn().mockResolvedValue(null) - }; - const binding = { - getQueue: vi.fn(() => queue), - getWorker: vi.fn(), - getLogger: vi.fn(() => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() })) - } as unknown as BullMQBinding; - const service = new AppDeleteMQService(binding); - - await expect(service.addAppJob({ teamId: 'team-1', appId: 'app-1' })).resolves.toEqual({ - id: 'job-app-1' - }); - - expect(queue.add).toHaveBeenCalledWith( - 'delete_app', - { teamId: 'team-1', appId: 'app-1', jobType: 'app' }, - { jobId: 'app-team-1-app-1' } - ); - expect(queue.getJob).toHaveBeenCalledWith('app-team-1-app-1'); - }); - - it('adds app deletion jobs in chunks of 200', async () => { - const queue = { - add: vi.fn(), - addBulk: vi.fn().mockImplementation(async (jobs) => jobs.map((job: unknown) => job)), - getJob: vi.fn().mockResolvedValue(null) + it('adds one task Flow with retry options on every node', async () => { + const flowProducer = { + addBulk: vi.fn().mockResolvedValue([{ job: { id: 'task-1' } }]) }; const binding = { - getQueue: vi.fn(() => queue), + getQueue: vi.fn(), getWorker: vi.fn(), + getFlowProducer: vi.fn(() => flowProducer), getLogger: vi.fn(() => ({ info: vi.fn(), warn: vi.fn(), error: vi.fn() })) } as unknown as BullMQBinding; const service = new AppDeleteMQService(binding); - const data = Array.from({ length: 201 }, (_, index) => ({ - teamId: 'team-1', - appId: `app-${index}` - })); - await service.addAppJobs(data); + await expect( + service.addFlows([ + { + name: 'delete_app_task', + queueName: 'appDelete', + data: { jobType: 'task' }, + opts: { jobId: 'task-1' }, + children: [ + { + name: 'delete_app_step', + queueName: 'appDelete', + data: { jobType: 'step' }, + opts: { jobId: 'task-1:step:app-1' } + } + ] + } + ]) + ).resolves.toEqual([{ job: { id: 'task-1' } }]); - expect(queue.addBulk).toHaveBeenCalledTimes(2); - expect(queue.addBulk.mock.calls[0][0]).toHaveLength(200); - expect(queue.addBulk.mock.calls[1][0]).toHaveLength(1); - expect(queue.addBulk.mock.calls[0][0][0]).toEqual({ - name: 'delete_app', - data: { teamId: 'team-1', appId: 'app-0', jobType: 'app' }, - opts: { jobId: 'app-team-1-app-0' } - }); - expect(queue.addBulk.mock.calls[1][0][0].opts).toEqual({ - jobId: 'app-team-1-app-200' - }); + expect(flowProducer.addBulk).toHaveBeenCalledWith([ + expect.objectContaining({ + opts: expect.objectContaining({ + jobId: 'task-1', + attempts: 10, + removeOnComplete: true + }), + children: [ + expect.objectContaining({ + opts: expect.objectContaining({ + jobId: 'task-1:step:app-1', + attempts: 10 + }) + }) + ] + }) + ]); }); it('uses failed-job recovery for Dataset deletion jobs', async () => { diff --git a/packages/dal/test/redis/bullmq.test.ts b/packages/dal/test/redis/bullmq.test.ts index 2741f59fc246..dfea63790549 100644 --- a/packages/dal/test/redis/bullmq.test.ts +++ b/packages/dal/test/redis/bullmq.test.ts @@ -10,9 +10,11 @@ import { RedisRuntime, type RedisClient } from '../../redis/runtime'; const bullMQMocks = vi.hoisted(() => ({ queues: [] as Array<{ name: string; options: Record }>, + flowProducers: [] as Array<{ options: Record }>, workers: [] as Array<{ name: string; options: Record }>, closeOrder: [] as string[], queueConstructorFailures: 0, + flowProducerConstructorFailures: 0, workerConstructorFailures: 0, onWorkerClose: undefined as (() => void) | undefined })); @@ -71,7 +73,27 @@ vi.mock('bullmq', async () => { } } + class MockFlowProducer extends EventEmitter { + readonly close = vi.fn(async () => { + bullMQMocks.closeOrder.push('flow-producer'); + }); + readonly disconnect = vi.fn(async () => undefined); + readonly connection = { + disconnect: vi.fn(async () => undefined) + }; + + constructor(readonly options: Record) { + super(); + if (bullMQMocks.flowProducerConstructorFailures > 0) { + bullMQMocks.flowProducerConstructorFailures -= 1; + throw new Error('flow producer constructor failed'); + } + bullMQMocks.flowProducers.push(this); + } + } + return { + FlowProducer: MockFlowProducer, Queue: MockQueue, UnrecoverableError: class UnrecoverableError extends Error {}, Worker: MockWorker @@ -102,9 +124,11 @@ const createBullMQRuntime = (options?: Partial) => { describe('RedisBullMQRuntime', () => { beforeEach(() => { bullMQMocks.queues.length = 0; + bullMQMocks.flowProducers.length = 0; bullMQMocks.workers.length = 0; bullMQMocks.closeOrder.length = 0; bullMQMocks.queueConstructorFailures = 0; + bullMQMocks.flowProducerConstructorFailures = 0; bullMQMocks.workerConstructorFailures = 0; bullMQMocks.onWorkerClose = undefined; }); @@ -122,6 +146,16 @@ describe('RedisBullMQRuntime', () => { expect(bullMQMocks.workers[0]?.options).toMatchObject({ concurrency: 2 }); }); + it('creates and reuses one FlowProducer per Runtime', () => { + const { redisRuntime, runtime } = createBullMQRuntime(); + + const flowProducer = runtime.getFlowProducer(); + + expect(runtime.getFlowProducer()).toBe(flowProducer); + expect(bullMQMocks.flowProducers).toHaveLength(1); + expect(redisRuntime.createQueueConnection).toHaveBeenCalledTimes(1); + }); + it('releases the Runtime connection when a Queue or Worker constructor fails', async () => { const { redisRuntime, runtime } = createBullMQRuntime(); bullMQMocks.queueConstructorFailures = 1; @@ -129,15 +163,19 @@ describe('RedisBullMQRuntime', () => { bullMQMocks.workerConstructorFailures = 1; expect(() => runtime.getWorker('worker-failure', vi.fn())).toThrow('worker constructor failed'); + + bullMQMocks.flowProducerConstructorFailures = 1; + expect(() => runtime.getFlowProducer()).toThrow('flow producer constructor failed'); await Promise.resolve(); - expect(redisRuntime.releaseConnection).toHaveBeenCalledTimes(2); + expect(redisRuntime.releaseConnection).toHaveBeenCalledTimes(3); }); it('registers the Redis hook and closes workers before queues', async () => { const { redisRuntime, runtime } = createBullMQRuntime(); const processor = vi.fn(); const queue = runtime.getQueue('datasetSync'); + runtime.getFlowProducer(); const worker = runtime.getWorker('datasetSync', processor); const queueError = vi.fn(); queue.on('error', queueError); @@ -150,7 +188,11 @@ describe('RedisBullMQRuntime', () => { expect(runtime.close()).toBe(closePromise); await closePromise; - expect(bullMQMocks.closeOrder).toEqual(['worker:datasetSync:true', 'queue:datasetSync']); + expect(bullMQMocks.closeOrder).toEqual([ + 'worker:datasetSync:true', + 'flow-producer', + 'queue:datasetSync' + ]); expect(worker.listenerCount('error')).toBe(0); expect(queue.listenerCount('error')).toBe(1); expect(queue.listeners('error')).toContain(queueError); @@ -181,7 +223,8 @@ describe('RedisBullMQRuntime', () => { const redisRuntime = new RedisRuntime({ redisUrl: 'redis://localhost', clientFactory: (options) => { - const client = new RuntimeRedisClient(options, clients.length === 0 ? 'queue' : 'worker'); + const role = clients.length === 0 ? 'queue' : clients.length === 1 ? 'flow' : 'worker'; + const client = new RuntimeRedisClient(options, role); clients.push(client); return client as unknown as RedisClient; } @@ -189,14 +232,17 @@ describe('RedisBullMQRuntime', () => { const bullMQRuntime = new RedisBullMQRuntime({ redisRuntime }); bullMQRuntime.getQueue('runtime-order'); + bullMQRuntime.getFlowProducer(); bullMQRuntime.getWorker('runtime-order', vi.fn()); await redisRuntime.close(); expect(bullMQMocks.closeOrder).toEqual([ 'worker:runtime-order:true', + 'flow-producer', 'queue:runtime-order', 'redis:queue', + 'redis:flow', 'redis:worker' ]); expect(bullMQRuntime.getState()).toBe('closed'); diff --git a/packages/service/core/app/delete/index.ts b/packages/service/core/app/delete/index.ts index eafff2b16576..2504906c1c44 100644 --- a/packages/service/core/app/delete/index.ts +++ b/packages/service/core/app/delete/index.ts @@ -1,5 +1,12 @@ import { appDeleteProcessor } from './processor'; -import { appDeleteMQService, type AppDeleteJobData } from '@fastgpt/dal/redis/bullmq'; +import { + appDeleteMQService, + QueueNames, + type AppDeleteJobData, + type FlowJob, + type Job +} from '@fastgpt/dal/redis/bullmq'; +import { findAppAndAllChildren } from '../controller'; import { MongoApp } from '../schema'; import { batchRunSettled } from '@fastgpt/global/common/system/utils'; import { getLogger, LogCategories } from '../../../common/logger'; @@ -7,18 +14,176 @@ import { setCron } from '../../../common/system/cron'; export type { AppDeleteJobData } from '@fastgpt/dal/redis/bullmq'; +type AppDeleteTaskInput = { + teamId: string; + appId: string; +}; + +type MarkedApp = { + _id: unknown; + teamId: unknown; + parentId?: unknown; +}; + const APP_DELETE_RESUME_BATCH_SIZE = 200; const APP_DELETE_RESUME_CONCURRENCY = 5; +const APP_DELETE_FLOW_BULK_SIZE = 50; const APP_DELETE_RESUME_CRON = '*/5 * * * *'; const logger = getLogger(LogCategories.MODULE.APP.FOLDER); let recoveryCronRegistered = false; let recoveryPromise: Promise | undefined; +const getTaskId = ({ teamId, appId }: AppDeleteTaskInput) => + `app-delete-task-${String(teamId)}-${String(appId)}`; + /** - * Initialize the app deletion worker and asynchronously resume soft-deleted apps - * whose cleanup was not completed. Recovery does not block worker creation, and - * one failed enqueue does not prevent other apps in the same batch from proceeding. + * Build one task Flow as a linked list. BullMQ processes children before parents, so reversing + * the resource list makes descendants complete before their parents and keeps the task root last. + */ +export const buildAppDeleteFlow = ({ + teamId, + appId, + apps +}: AppDeleteTaskInput & { apps: Array<{ _id: unknown }> }): FlowJob => { + const taskId = getTaskId({ teamId, appId }); + const stepChain = apps + .slice() + .reverse() + .reduce((nextStep, app) => { + const resourceId = String(app._id); + const step: FlowJob = { + name: 'delete_app_step', + queueName: QueueNames.appDelete, + data: { + teamId, + appId: resourceId, + taskId, + jobType: 'step' + }, + opts: { + jobId: `${taskId}:step:${resourceId}`, + failParentOnFailure: true, + ...(nextStep ? {} : { delay: 1000 }) + } + }; + + if (nextStep) step.children = [nextStep]; + return step; + }, undefined); + + return { + name: 'delete_app_task', + queueName: QueueNames.appDelete, + data: { + teamId, + appId, + taskId, + jobType: 'task' + }, + opts: { + jobId: taskId + }, + ...(stepChain ? { children: [stepChain] } : {}) + }; +}; + +/** Load the marked app subtree and construct its ordered deletion Flow. */ +const loadDeleteTaskFlow = async (data: AppDeleteTaskInput) => { + const apps = await findAppAndAllChildren({ + teamId: data.teamId, + appId: data.appId, + fields: '_id teamId parentId deleteTime' + }); + + const unmarkedApps = apps.filter((app) => !app.deleteTime); + if (unmarkedApps.length > 0) { + logger.warn('App delete safety check mismatch', { + markedCount: apps.length - unmarkedApps.length, + totalCount: apps.length, + unmarkedCount: unmarkedApps.length + }); + throw new Error('App delete safety check mismatch'); + } + + return { + flow: buildAppDeleteFlow({ ...data, apps }) + }; +}; + +/** + * Remove only a terminal task Flow before rebuilding it. Active and waiting tasks are left alone, + * allowing multiple recovery callers to converge on the same stable task ID. + */ +const prepareTaskFlow = async (data: AppDeleteTaskInput) => { + const taskId = getTaskId(data); + const queue = appDeleteMQService.getQueue(); + const getExistingJob = async () => { + const job = await queue.getJob(taskId); + if (!job) return; + + const state = await job.getState(); + if (state !== 'unknown') return { job, state }; + + const latestJob = await queue.getJob(taskId); + if (!latestJob) return; + + const latestState = await latestJob.getState(); + if (latestState === 'unknown') { + throw new Error(`BullMQ app delete task is in an unknown state: ${taskId}`); + } + return { job: latestJob, state: latestState }; + }; + + const existing = await getExistingJob(); + if (existing) { + if ( + ['waiting-children', 'waiting', 'active', 'delayed', 'prioritized', 'paused'].includes( + existing.state + ) + ) { + return { job: existing.job }; + } + await existing.job.remove({ removeChildren: true }); + } + + return loadDeleteTaskFlow(data); +}; + +/** Submit task Flows in bounded atomic batches so recovery cannot create one resource job at a time. */ +const addAppDeleteTasks = async (tasks: AppDeleteTaskInput[]) => { + const flows: FlowJob[] = []; + const jobs: Job[] = []; + for (const task of tasks) { + const prepared = await prepareTaskFlow(task); + if ('job' in prepared) { + jobs.push(prepared.job); + } else { + flows.push(prepared.flow); + } + } + + for (let index = 0; index < flows.length; index += APP_DELETE_FLOW_BULK_SIZE) { + const nodes = await appDeleteMQService.addFlows( + flows.slice(index, index + APP_DELETE_FLOW_BULK_SIZE) + ); + jobs.push(...nodes.map((node) => node.job)); + } + return jobs; +}; + +/** Add one deletion task Flow. The public contract remains root App based, not resource based. */ +export const addAppDeleteJob = async (data: AppDeleteTaskInput) => { + const [job] = await addAppDeleteTasks([data]); + return job; +}; + +/** Add multiple root deletion tasks through BullMQ FlowProducer.addBulk. */ +export const addAppDeleteJobs = (data: AppDeleteTaskInput[]) => addAppDeleteTasks(data); + +/** + * Initialize the App deletion worker and asynchronously resume soft-deleted task roots whose Flow + * was not created or reached a terminal failure. Recovery does not block worker creation. */ export const initAppDeleteWorker = () => { const worker = appDeleteMQService.getWorker(appDeleteProcessor); @@ -32,10 +197,7 @@ export const initAppDeleteWorker = () => { return worker; }; -/** - * Keep recovery alive after transient Redis or Mongo failures during startup. - * Stable app job IDs make repeated scans idempotent across workers and pods. - */ +/** Keep recovery alive after transient Redis or Mongo failures during startup. */ const registerAppDeleteRecoveryCron = () => { if (recoveryCronRegistered) return; @@ -47,10 +209,16 @@ const registerAppDeleteRecoveryCron = () => { }); }; -/** - * Scan soft-deleted apps and resume their single-app deletion jobs. - * Cursor pagination and bounded concurrency prevent startup from saturating Mongo or Redis. - */ +/** Select marked apps whose parent is not also marked, producing one task root per subtree. */ +const getMarkedTaskRoots = (apps: MarkedApp[]) => { + const markedIds = new Set(apps.map((app) => `${String(app.teamId)}:${String(app._id)}`)); + return apps.filter((app) => { + if (!app.parentId) return true; + return !markedIds.has(`${String(app.teamId)}:${String(app.parentId)}`); + }); +}; + +/** Scan only marked tree roots and rebuild missing or failed task Flows with bounded concurrency. */ async function resumeMarkedAppDeleteJobsInternal(): Promise { const cursor = MongoApp.find( { @@ -61,57 +229,59 @@ async function resumeMarkedAppDeleteJobsInternal(): Promise { }, { _id: 1, - teamId: 1 + teamId: 1, + parentId: 1 } ) - .lean() + .lean() .cursor({ batchSize: APP_DELETE_RESUME_BATCH_SIZE }); let totalMarked = 0; + let rootCount = 0; let resumedCount = 0; let failedCount = 0; - let batch: { teamId: string; appId: string }[] = []; - - /** Flush one bounded batch so startup recovery does not overload Redis or Mongo. */ - const flushBatch = async () => { - if (batch.length === 0) return; - - const currentBatch = batch; - batch = []; - const results = await batchRunSettled( - currentBatch, - (item) => appDeleteMQService.addAppJob(item), - APP_DELETE_RESUME_CONCURRENCY - ); - resumedCount += results.filter((result) => result.success).length; - failedCount += results.filter((result) => !result.success).length; - }; + const markedApps: MarkedApp[] = []; for await (const app of cursor) { totalMarked += 1; - batch.push({ - teamId: String(app.teamId), - appId: String(app._id) - }); + markedApps.push(app); + } - if (batch.length >= APP_DELETE_RESUME_BATCH_SIZE) { - await flushBatch(); - } + const roots = getMarkedTaskRoots(markedApps); + rootCount = roots.length; + const tasks = roots.map((root) => ({ + teamId: String(root.teamId), + appId: String(root._id) + })); + + const taskBatches = []; + for (let index = 0; index < tasks.length; index += APP_DELETE_FLOW_BULK_SIZE) { + taskBatches.push(tasks.slice(index, index + APP_DELETE_FLOW_BULK_SIZE)); } - await flushBatch(); + const results = await batchRunSettled( + taskBatches, + (batch) => addAppDeleteJobs(batch), + APP_DELETE_RESUME_CONCURRENCY + ); + + for (const [index, result] of results.entries()) { + if (result.success) { + resumedCount += taskBatches[index].length; + } else { + failedCount += taskBatches[index].length; + } + } - logger.info('Marked app delete jobs resumed', { + logger.info('Marked app delete tasks resumed', { totalMarked, + rootCount, resumedCount, failedCount }); } -/** - * Resume marked app cleanup with process-local overlap protection. - * Multiple callers share one scan while distributed callers converge on stable job IDs. - */ +/** Process-local overlap protection; distributed callers converge on stable Flow task IDs. */ export function resumeMarkedAppDeleteJobs(): Promise { if (recoveryPromise) return recoveryPromise; @@ -120,10 +290,3 @@ export function resumeMarkedAppDeleteJobs(): Promise { }); return recoveryPromise; } - -/** Add a root app deletion job. */ -export const addAppDeleteJob = (data: AppDeleteJobData) => appDeleteMQService.addJob(data); - -/** Add a single-app deletion job for startup recovery and other recovery flows. */ -export const addAppDeleteAppJob = (data: Omit) => - appDeleteMQService.addAppJob(data); diff --git a/packages/service/core/app/delete/processor.ts b/packages/service/core/app/delete/processor.ts index 52315ff2dc47..84cc13220a1f 100644 --- a/packages/service/core/app/delete/processor.ts +++ b/packages/service/core/app/delete/processor.ts @@ -1,9 +1,9 @@ import type { Processor } from '@fastgpt/dal/redis/bullmq'; import type { AppDeleteJobData } from './index'; -import { appDeleteMQService } from '@fastgpt/dal/redis/bullmq'; -import { findAppAndAllChildren, deleteAppDataProcessor } from '../controller'; +import { deleteAppDataProcessor } from '../controller'; import { MongoApp } from '../schema'; import { getLogger, LogCategories } from '../../../common/logger'; +import { addAppDeleteJob } from './index'; const logger = getLogger(LogCategories.MODULE.APP.FOLDER); @@ -53,48 +53,29 @@ export const appDeleteProcessor: Processor = async (job) => { logger.info('App delete started', { teamId, appId }); try { - if (jobType === 'app') { - await deleteSingleApp({ teamId, appId }); + if (jobType === 'task') { + logger.info('App delete task completed', { + teamId, + appId, + taskId: job.data.taskId, + durationMs: Date.now() - startTime + }); return; } - // 1. Find the app subtree using only fields needed to create child jobs. - const apps = await findAppAndAllChildren({ - teamId, - appId, - fields: '_id teamId parentId deleteTime' - }); - - if (!apps || apps.length === 0) { - logger.warn('App not found for deletion', { teamId, appId }); + if (jobType === 'step' || jobType === 'app') { + await deleteSingleApp({ teamId, appId }); return; } - // 2. The root task only splits work, but the complete subtree must be soft-deleted. - const unmarkedApps = apps.filter((app) => !app.deleteTime); - if (unmarkedApps.length > 0) { - logger.warn('App delete safety check mismatch', { - markedCount: apps.length - unmarkedApps.length, - totalCount: apps.length, - unmarkedCount: unmarkedApps.length - }); - throw new Error('App delete safety check mismatch'); - } - - // 3. Split the work without cleaning any app resources in this job. - await appDeleteMQService.addAppJobs( - apps.map((app) => ({ - teamId, - appId: String(app._id) - })) - ); + // Legacy root jobs become a bridge to the task Flow. New requests enqueue the Flow directly. + await addAppDeleteJob({ teamId, appId }); logger.info('App delete completed', { teamId, appId, - childCount: apps.length - 1, durationMs: Date.now() - startTime, - totalApps: apps.length + legacy: true }); } catch (error: any) { logger.error('App delete failed', { teamId, appId, error }); diff --git a/packages/service/support/user/team/delete/utils.ts b/packages/service/support/user/team/delete/utils.ts index 4be050b174c3..e875109df881 100644 --- a/packages/service/support/user/team/delete/utils.ts +++ b/packages/service/support/user/team/delete/utils.ts @@ -1,9 +1,9 @@ import { MongoApp } from '../../../../core/app/schema'; import { deleteAppsImmediate } from '../../../../core/app/controller'; -import { addAppDeleteJob } from '../../../../core/app/delete'; +import { addAppDeleteJobs } from '../../../../core/app/delete'; export const onDelAllApp = async (teamId: string) => { - // 正常只投递根应用；如果历史数据留下孤立子应用，则把孤立应用作为自己的根补偿投递。 + // Normally only roots are submitted; orphaned children become compensating roots. const apps = await MongoApp.find( { teamId @@ -32,11 +32,11 @@ export const onDelAllApp = async (teamId: string) => { } ); - // 添加到删除队列 - for (const app of deleteRootApps) { - await addAppDeleteJob({ + // Add all root task Flows atomically in bounded batches. + await addAppDeleteJobs( + deleteRootApps.map((app) => ({ teamId, appId: String(app._id) - }); - } + })) + ); }; diff --git a/packages/service/test/core/app/delete/processor.test.ts b/packages/service/test/core/app/delete/processor.test.ts index 6e104c30b787..46767e77a80e 100644 --- a/packages/service/test/core/app/delete/processor.test.ts +++ b/packages/service/test/core/app/delete/processor.test.ts @@ -7,8 +7,8 @@ const mocks = vi.hoisted(() => ({ deleteAppDataProcessor: vi.fn(), findOne: vi.fn(), find: vi.fn(), - addAppJobs: vi.fn(), - addAppJob: vi.fn(), + addFlows: vi.fn(), + getQueue: vi.fn(), getWorker: vi.fn(), setCron: vi.fn(), logger: { @@ -41,14 +41,19 @@ vi.mock('@fastgpt/service/common/system/cron', () => ({ vi.mock('@fastgpt/dal/redis/bullmq', () => ({ appDeleteMQService: { - addAppJobs: mocks.addAppJobs, - addAppJob: mocks.addAppJob, + addFlows: mocks.addFlows, + getQueue: mocks.getQueue, getWorker: mocks.getWorker - } + }, + QueueNames: { appDelete: 'appDelete' } })); import { appDeleteProcessor } from '../../../../core/app/delete/processor'; -import { initAppDeleteWorker, resumeMarkedAppDeleteJobs } from '../../../../core/app/delete'; +import { + addAppDeleteJob, + initAppDeleteWorker, + resumeMarkedAppDeleteJobs +} from '../../../../core/app/delete'; const createJob = (id: string, data: AppDeleteJobData) => ({ id, data }) as unknown as Job; @@ -61,8 +66,8 @@ describe('appDeleteProcessor', () => { { _id: 'child-1', teamId: 'team-1', parentId: 'root', deleteTime: new Date() }, { _id: 'child-2', teamId: 'team-1', parentId: 'root', deleteTime: new Date() } ]); - mocks.addAppJobs.mockResolvedValue([]); - mocks.addAppJob.mockResolvedValue({ id: 'app-delete-job' }); + mocks.addFlows.mockResolvedValue([{ job: { id: 'app-delete-task' } }]); + mocks.getQueue.mockReturnValue({ getJob: vi.fn().mockResolvedValue(null) }); mocks.findOne.mockReturnValue({ lean: vi.fn().mockResolvedValue({ _id: 'app-1', @@ -75,7 +80,7 @@ describe('appDeleteProcessor', () => { mocks.deleteAppDataProcessor.mockResolvedValue(undefined); }); - it('splits a root deletion job into single-app jobs without cleaning app data', async () => { + it('bridges a legacy root deletion job into one task Flow', async () => { await appDeleteProcessor( createJob('root-job', { teamId: 'team-1', appId: 'root', jobType: 'root' }) ); @@ -85,10 +90,26 @@ describe('appDeleteProcessor', () => { appId: 'root', fields: '_id teamId parentId deleteTime' }); - expect(mocks.addAppJobs).toHaveBeenCalledWith([ - { teamId: 'team-1', appId: 'root' }, - { teamId: 'team-1', appId: 'child-1' }, - { teamId: 'team-1', appId: 'child-2' } + expect(mocks.addFlows).toHaveBeenCalledWith([ + expect.objectContaining({ + name: 'delete_app_task', + data: expect.objectContaining({ jobType: 'task' }), + children: [ + expect.objectContaining({ + data: expect.objectContaining({ appId: 'root', jobType: 'step' }), + children: [ + expect.objectContaining({ + data: expect.objectContaining({ appId: 'child-1', jobType: 'step' }), + children: [ + expect.objectContaining({ + data: expect.objectContaining({ appId: 'child-2', jobType: 'step' }) + }) + ] + }) + ] + }) + ] + }) ]); expect(mocks.findOne).not.toHaveBeenCalled(); expect(mocks.deleteAppDataProcessor).not.toHaveBeenCalled(); @@ -97,7 +118,22 @@ describe('appDeleteProcessor', () => { it('treats a historical job without jobType as a root job', async () => { await appDeleteProcessor(createJob('legacy-root-job', { teamId: 'team-1', appId: 'root' })); - expect(mocks.addAppJobs).toHaveBeenCalledTimes(1); + expect(mocks.addFlows).toHaveBeenCalledTimes(1); + expect(mocks.deleteAppDataProcessor).not.toHaveBeenCalled(); + }); + + it('completes a task root after all child steps have finished', async () => { + await appDeleteProcessor( + createJob('task-job', { + teamId: 'team-1', + appId: 'root', + taskId: 'task-1', + jobType: 'task' + }) + ); + + expect(mocks.findAppAndAllChildren).not.toHaveBeenCalled(); + expect(mocks.findOne).not.toHaveBeenCalled(); expect(mocks.deleteAppDataProcessor).not.toHaveBeenCalled(); }); @@ -158,8 +194,9 @@ describe('appDeleteProcessor', () => { it('resumes marked apps through a cursor with bounded batch processing', async () => { const cursor = (async function* () { - yield { _id: 'app-1', teamId: 'team-1' }; - yield { _id: 'app-2', teamId: 'team-2' }; + yield { _id: 'app-1', teamId: 'team-1', parentId: null }; + yield { _id: 'child-1', teamId: 'team-1', parentId: 'app-1' }; + yield { _id: 'app-2', teamId: 'team-2', parentId: null }; })(); const lean = vi.fn().mockReturnValue({ cursor: vi.fn().mockReturnValue(cursor) @@ -170,23 +207,62 @@ describe('appDeleteProcessor', () => { expect(mocks.find).toHaveBeenCalledWith( { deleteTime: { $exists: true, $ne: null } }, - { _id: 1, teamId: 1 } + { _id: 1, teamId: 1, parentId: 1 } ); - expect(mocks.addAppJob).toHaveBeenNthCalledWith(1, { - teamId: 'team-1', - appId: 'app-1' - }); - expect(mocks.addAppJob).toHaveBeenNthCalledWith(2, { - teamId: 'team-2', - appId: 'app-2' - }); - expect(mocks.logger.info).toHaveBeenCalledWith('Marked app delete jobs resumed', { - totalMarked: 2, + expect(mocks.addFlows).toHaveBeenCalledTimes(1); + expect(mocks.addFlows.mock.calls[0]?.[0]).toHaveLength(2); + expect(mocks.logger.info).toHaveBeenCalledWith('Marked app delete tasks resumed', { + totalMarked: 3, + rootCount: 2, resumedCount: 2, failedCount: 0 }); }); + it('reuses an active task instead of building another Flow', async () => { + const existingJob = { + getState: vi.fn().mockResolvedValue('waiting-children') + }; + const getJob = vi.fn().mockResolvedValue(existingJob); + mocks.getQueue.mockReturnValue({ getJob }); + + await expect(addAppDeleteJob({ teamId: 'team-1', appId: 'root' })).resolves.toBe(existingJob); + + expect(getJob).toHaveBeenCalledWith('app-delete-task-team-1-root'); + expect(mocks.findAppAndAllChildren).not.toHaveBeenCalled(); + expect(mocks.addFlows).not.toHaveBeenCalled(); + }); + + it('removes a failed task Flow before rebuilding it', async () => { + const existingJob = { + getState: vi.fn().mockResolvedValue('failed'), + remove: vi.fn().mockResolvedValue(undefined) + }; + mocks.getQueue.mockReturnValue({ getJob: vi.fn().mockResolvedValue(existingJob) }); + + await addAppDeleteJob({ teamId: 'team-1', appId: 'root' }); + + expect(existingJob.remove).toHaveBeenCalledWith({ removeChildren: true }); + expect(mocks.findAppAndAllChildren).toHaveBeenCalledTimes(1); + expect(mocks.addFlows).toHaveBeenCalledTimes(1); + }); + + it('does not rebuild a task while its state is still unknown', async () => { + const existingJob = { + getState: vi.fn().mockResolvedValue('unknown') + }; + const getJob = vi.fn().mockResolvedValue(existingJob); + mocks.getQueue.mockReturnValue({ getJob }); + + await expect(addAppDeleteJob({ teamId: 'team-1', appId: 'root' })).rejects.toThrow( + 'BullMQ app delete task is in an unknown state' + ); + + expect(getJob).toHaveBeenCalledTimes(2); + expect(mocks.findAppAndAllChildren).not.toHaveBeenCalled(); + expect(mocks.addFlows).not.toHaveBeenCalled(); + }); + it('registers periodic recovery when the worker starts', async () => { const cursor = (async function* () {})(); mocks.find.mockReturnValue({ diff --git a/projects/app/src/pages/api/core/app/del.ts b/projects/app/src/pages/api/core/app/del.ts index 1ba2dae351fc..8c158a3f1e48 100644 --- a/projects/app/src/pages/api/core/app/del.ts +++ b/projects/app/src/pages/api/core/app/del.ts @@ -54,14 +54,11 @@ async function handler(req: NextApiRequest): Promise { teamId, appIds: deleteAppsList.map((app) => app._id) }); - - // Add to delete queue for async cleanup - await addAppDeleteJob({ - teamId, - appId - }); }); + // Add the task Flow after the soft-delete transaction commits. + await addAppDeleteJob({ teamId, appId }); + (async () => { addAuditLog({ tmbId, diff --git a/projects/app/test/api/core/app/delete.test.ts b/projects/app/test/api/core/app/delete.test.ts index 9c8cb9e24fc5..e5af0b09a55a 100644 --- a/projects/app/test/api/core/app/delete.test.ts +++ b/projects/app/test/api/core/app/delete.test.ts @@ -1,6 +1,5 @@ import { describe, expect, it, beforeEach, vi, afterEach } from 'vitest'; -import type { AppDeleteJobData } from '@fastgpt/service/core/app/delete'; -import { addAppDeleteJob } from '@fastgpt/service/core/app/delete'; +import { buildAppDeleteFlow } from '@fastgpt/service/core/app/delete'; import { appDeleteProcessor } from '@fastgpt/service/core/app/delete/processor'; import handler from '@/pages/api/core/app/del'; import { MongoApp } from '@fastgpt/service/core/app/schema'; @@ -34,29 +33,19 @@ vi.mock('@fastgpt/dal/redis/bullmq', () => { return { bullMQ, appDeleteMQService: { - addJob: (data: AppDeleteJobData) => - bullMQ - .getQueue('appDelete', { - defaultJobOptions: { - attempts: 10, - backoff: { - type: 'exponential', - delay: 5000 - }, - removeOnComplete: true, - removeOnFail: { age: 30 * 24 * 60 * 60 } - } - }) - .add( - 'delete_app', - { ...data, jobType: 'root' }, - { - jobId: `${data.teamId}-${data.appId}`, - delay: 1000 - } - ), - addAppJobs: vi.fn().mockResolvedValue([]), - addAppJob: vi.fn().mockResolvedValue({ id: 'app-delete-job' }) + getQueue: () => + bullMQ.getQueue('appDelete', { + defaultJobOptions: { + attempts: 10, + backoff: { + type: 'exponential', + delay: 5000 + }, + removeOnComplete: true, + removeOnFail: { age: 30 * 24 * 60 * 60 } + } + }), + addFlows: vi.fn().mockResolvedValue([{ job: { id: 'app-delete-task' } }]) }, QueueNames: { appDelete: 'appDelete' @@ -76,9 +65,10 @@ vi.mock('@fastgpt/service/common/file/image/controller', () => ({ })); // Import mocked modules for type access -import { bullMQ, QueueNames } from '@fastgpt/dal/redis/bullmq'; +import { bullMQ, QueueNames, appDeleteMQService } from '@fastgpt/dal/redis/bullmq'; const mockGetQueue = vi.mocked(bullMQ.getQueue); +const mockAddFlows = vi.mocked(appDeleteMQService.addFlows); describe('App Delete Queue', () => { beforeEach(() => { @@ -89,64 +79,32 @@ describe('App Delete Queue', () => { vi.restoreAllMocks(); }); - describe('addAppDeleteJob', () => { - it('should add job to queue with correct parameters', async () => { - const mockQueue = { - add: vi.fn().mockResolvedValue({ id: 'job-123' }) - }; - mockGetQueue.mockReturnValue(mockQueue as any); - - const jobData: AppDeleteJobData = { - teamId: 'team-123', - appId: 'app-123' - }; - - const result = await addAppDeleteJob(jobData); - - expect(mockGetQueue).toHaveBeenCalledWith(QueueNames.appDelete, { - defaultJobOptions: { - attempts: 10, - backoff: { - type: 'exponential', - delay: 5000 - }, - removeOnComplete: true, - removeOnFail: { age: 30 * 24 * 60 * 60 } - } + describe('buildAppDeleteFlow', () => { + it('builds a task flow with descendants before parents', () => { + const flow = buildAppDeleteFlow({ + teamId: 'team-1', + appId: 'root', + apps: [{ _id: 'root' }, { _id: 'child' }, { _id: 'grandchild' }] }); - expect(mockQueue.add).toHaveBeenCalledWith( - 'delete_app', - { ...jobData, jobType: 'root' }, - { - jobId: 'team-123-app-123', - delay: 1000 - } - ); - - expect(result).toEqual({ id: 'job-123' }); - }); - - it('should use correct jobId format for preventing duplicates', async () => { - const mockQueue = { - add: vi.fn().mockResolvedValue({ id: 'job-456' }) - }; - mockGetQueue.mockReturnValue(mockQueue as any); - - const jobData: AppDeleteJobData = { - teamId: 'team-xyz', - appId: 'app-abc' - }; - - await addAppDeleteJob(jobData); - - expect(mockQueue.add).toHaveBeenCalledWith( - 'delete_app', - { ...jobData, jobType: 'root' }, - expect.objectContaining({ - jobId: 'team-xyz-app-abc' - }) - ); + expect(flow).toMatchObject({ + name: 'delete_app_task', + queueName: QueueNames.appDelete, + data: { teamId: 'team-1', appId: 'root', jobType: 'task' }, + opts: { jobId: 'app-delete-task-team-1-root' }, + children: [ + { + opts: { failParentOnFailure: true }, + data: { appId: 'root', jobType: 'step' }, + children: [ + { + data: { appId: 'child', jobType: 'step' }, + children: [{ data: { appId: 'grandchild', jobType: 'step' } }] + } + ] + } + ] + }); }); }); }); @@ -171,7 +129,7 @@ describe('App Delete API Integration', () => { // Mock the queue to avoid actual background deletion const mockQueue = { - add: vi.fn().mockResolvedValue({ id: 'job-123' }) + getJob: vi.fn().mockResolvedValue(null) }; mockGetQueue.mockReturnValue(mockQueue as any); @@ -189,18 +147,16 @@ describe('App Delete API Integration', () => { expect(deletedApp?.deleteTime).not.toBeNull(); // Verify queue job was added - expect(mockQueue.add).toHaveBeenCalledWith( - 'delete_app', - { - teamId: rootUser.teamId, - appId: String(testApp._id), - jobType: 'root' - }, - { - jobId: `${rootUser.teamId}-${testApp._id}`, - delay: 1000 - } - ); + expect(mockAddFlows).toHaveBeenCalledWith([ + expect.objectContaining({ + name: 'delete_app_task', + data: expect.objectContaining({ + teamId: rootUser.teamId, + appId: String(testApp._id), + jobType: 'task' + }) + }) + ]); // Cleanup await MongoApp.deleteOne({ _id: testApp._id }); @@ -228,7 +184,7 @@ describe('App Delete API Integration', () => { // Mock the queue const mockQueue = { - add: vi.fn().mockResolvedValue({ id: 'job-folder' }) + getJob: vi.fn().mockResolvedValue(null) }; mockGetQueue.mockReturnValue(mockQueue as any); @@ -273,7 +229,7 @@ describe('App Delete API Integration', () => { }); const mockQueue = { - add: vi.fn().mockResolvedValue({ id: 'job-workflow-tool' }) + getJob: vi.fn().mockResolvedValue(null) }; mockGetQueue.mockReturnValue(mockQueue as any); From f163dfc1008d91d82d97fec36a6fb28f0562ea8c Mon Sep 17 00:00:00 2001 From: Finley Ge Date: Tue, 1 Sep 2026 19:16:41 +0800 Subject: [PATCH 3/3] chore(app-delete): align flow implementation with standards --- .../dal/redis/bullmq/flow-producer-manager.ts | 7 +-- packages/service/core/app/delete/index.ts | 62 +++++++++---------- 2 files changed, 31 insertions(+), 38 deletions(-) diff --git a/packages/dal/redis/bullmq/flow-producer-manager.ts b/packages/dal/redis/bullmq/flow-producer-manager.ts index ec852cd890ae..a8bca36c8761 100644 --- a/packages/dal/redis/bullmq/flow-producer-manager.ts +++ b/packages/dal/redis/bullmq/flow-producer-manager.ts @@ -2,7 +2,6 @@ import { FlowProducer } from 'bullmq'; import type { RedisRuntime } from '../runtime/connection'; import type { RedisRuntimeLogger } from '../types'; import { closeWithTimeout, forceDisconnect } from './close'; -import type { BullMQDisconnectable } from './types'; /** Manages the shared FlowProducer lifecycle for a BullMQ Runtime. */ export class BullMQFlowProducerManager { @@ -51,14 +50,10 @@ export class BullMQFlowProducerManager { timeoutMs: this.options.closeTimeoutMs }).catch((error) => { this.options.logger.warn('BullMQ flow producer close failed', { error }); - const connection = (flowProducer as unknown as { connection?: BullMQDisconnectable }) - .connection; forceDisconnect({ name: 'flow-producer', resource: 'flow producer connection', - disconnect: connection - ? () => connection.disconnect(false) - : () => flowProducer.disconnect(), + disconnect: () => flowProducer.disconnect(), logger: this.options.logger }); }); diff --git a/packages/service/core/app/delete/index.ts b/packages/service/core/app/delete/index.ts index 2504906c1c44..aef8e729fcec 100644 --- a/packages/service/core/app/delete/index.ts +++ b/packages/service/core/app/delete/index.ts @@ -88,29 +88,6 @@ export const buildAppDeleteFlow = ({ }; }; -/** Load the marked app subtree and construct its ordered deletion Flow. */ -const loadDeleteTaskFlow = async (data: AppDeleteTaskInput) => { - const apps = await findAppAndAllChildren({ - teamId: data.teamId, - appId: data.appId, - fields: '_id teamId parentId deleteTime' - }); - - const unmarkedApps = apps.filter((app) => !app.deleteTime); - if (unmarkedApps.length > 0) { - logger.warn('App delete safety check mismatch', { - markedCount: apps.length - unmarkedApps.length, - totalCount: apps.length, - unmarkedCount: unmarkedApps.length - }); - throw new Error('App delete safety check mismatch'); - } - - return { - flow: buildAppDeleteFlow({ ...data, apps }) - }; -}; - /** * Remove only a terminal task Flow before rebuilding it. Active and waiting tasks are left alone, * allowing multiple recovery callers to converge on the same stable task ID. @@ -118,6 +95,28 @@ const loadDeleteTaskFlow = async (data: AppDeleteTaskInput) => { const prepareTaskFlow = async (data: AppDeleteTaskInput) => { const taskId = getTaskId(data); const queue = appDeleteMQService.getQueue(); + const loadDeleteTaskFlow = async () => { + const apps = await findAppAndAllChildren({ + teamId: data.teamId, + appId: data.appId, + fields: '_id teamId parentId deleteTime' + }); + + const unmarkedApps = apps.filter((app) => !app.deleteTime); + if (unmarkedApps.length > 0) { + logger.warn('App delete safety check mismatch', { + markedCount: apps.length - unmarkedApps.length, + totalCount: apps.length, + unmarkedCount: unmarkedApps.length + }); + throw new Error('App delete safety check mismatch'); + } + + return { + flow: buildAppDeleteFlow({ ...data, apps }) + }; + }; + const getExistingJob = async () => { const job = await queue.getJob(taskId); if (!job) return; @@ -209,15 +208,6 @@ const registerAppDeleteRecoveryCron = () => { }); }; -/** Select marked apps whose parent is not also marked, producing one task root per subtree. */ -const getMarkedTaskRoots = (apps: MarkedApp[]) => { - const markedIds = new Set(apps.map((app) => `${String(app.teamId)}:${String(app._id)}`)); - return apps.filter((app) => { - if (!app.parentId) return true; - return !markedIds.has(`${String(app.teamId)}:${String(app.parentId)}`); - }); -}; - /** Scan only marked tree roots and rebuild missing or failed task Flows with bounded concurrency. */ async function resumeMarkedAppDeleteJobsInternal(): Promise { const cursor = MongoApp.find( @@ -247,6 +237,14 @@ async function resumeMarkedAppDeleteJobsInternal(): Promise { markedApps.push(app); } + const getMarkedTaskRoots = (apps: MarkedApp[]) => { + const markedIds = new Set(apps.map((app) => `${String(app.teamId)}:${String(app._id)}`)); + return apps.filter((app) => { + if (!app.parentId) return true; + return !markedIds.has(`${String(app.teamId)}:${String(app.parentId)}`); + }); + }; + const roots = getMarkedTaskRoots(markedApps); rootCount = roots.length; const tasks = roots.map((root) => ({