fix(cron): address review issues from PR #1447

zynx · zynx · commit 492a2f548d03 · 2026-03-19T17:27:19.000+08:00
- Fix setProcessing(true) busy-state leak: move into executeJob after task acquisition via onAcquired callback, preserving onceIdle ordering guarantee - Remove FallbackConversationRepository dead-code class (no instantiation point, no fallback logic, no tests); remove from vitest coverage include - Add missing searchMessages mock to WorkerTaskManager.test.ts makeRepo() Review follow-up for #1447
diff --git a/src/process/database/FallbackConversationRepository.ts b/src/process/database/FallbackConversationRepository.ts
diff --git a/src/process/services/cron/CronService.ts b/src/process/services/cron/CronService.ts
@@ -380,13 +380,12 @@ export class CronService {
     let lastError: string | undefined;
 
     try {
-      // Mark conversation as busy BEFORE registering the idle callback,
-      // so onceIdle registers a deferred callback instead of firing immediately.
-      this.executor.setProcessing(conversationId, true);
-      this.registerCompletionNotification(job);
-
-      // Delegate task-get + sendMessage to the executor.
-      await this.executor.executeJob(job);
+      // executeJob marks the conversation busy only after task acquisition succeeds.
+      // The onAcquired callback registers the completion notification while the
+      // conversation is already busy, preventing premature onceIdle fires.
+      await this.executor.executeJob(job, () => {
+        this.registerCompletionNotification(job);
+      });
 
       // Success
       this.retryCounts.delete(job.id);
diff --git a/src/process/services/cron/ICronJobExecutor.ts b/src/process/services/cron/ICronJobExecutor.ts
@@ -9,8 +9,11 @@ import type { CronJob } from './CronStore';
 export interface ICronJobExecutor {
   /** Returns true if the conversation already has an active run in progress. */
   isConversationBusy(conversationId: string): boolean;
-  /** Execute the job's payload against the target conversation. */
-  executeJob(job: CronJob): Promise<void>;
+  /** Execute the job's payload against the target conversation.
+   * @param onAcquired - Called after task acquisition succeeds, before sendMessage.
+   *   Use this hook to register completion notifications while guaranteeing that
+   *   busy-state has already been set (avoiding premature onceIdle fires). */
+  executeJob(job: CronJob, onAcquired?: () => void): Promise<void>;
   /** Register a callback to fire once the conversation becomes idle. */
   onceIdle(conversationId: string, callback: () => Promise<void>): void;
   /** Mark the conversation as busy/not-busy. */
diff --git a/src/process/services/cron/WorkerTaskManagerJobExecutor.ts b/src/process/services/cron/WorkerTaskManagerJobExecutor.ts
@@ -24,7 +24,7 @@ export class WorkerTaskManagerJobExecutor implements ICronJobExecutor {
     return this.busyGuard.isProcessing(conversationId);
   }
 
-  async executeJob(job: CronJob): Promise<void> {
+  async executeJob(job: CronJob, onAcquired?: () => void): Promise<void> {
     const { conversationId } = job.metadata;
     const messageText = job.target.payload.text;
     const msgId = uuid();
@@ -45,6 +45,14 @@ export class WorkerTaskManagerJobExecutor implements ICronJobExecutor {
       task = await this.taskManager.getOrBuildTask(conversationId, { yoloMode: true });
     }
 
+    // Mark busy only after task acquisition succeeds. This ensures that if
+    // getOrBuildTask throws (conversation deleted), setProcessing(true) is never
+    // called and no "busy" state leaks into subsequent runs.
+    this.busyGuard.setProcessing(conversationId, true);
+    // Notify caller so it can register onceIdle callbacks while the conversation
+    // is already marked busy (prevents premature idle fires).
+    onAcquired?.();
+
     const workspace = (task as { workspace?: string }).workspace;
     const workspaceFiles = workspace ? await copyFilesToDirectory(workspace, [], false) : [];
 
diff --git a/tests/unit/WorkerTaskManager.test.ts b/tests/unit/WorkerTaskManager.test.ts
@@ -16,6 +16,7 @@ function makeRepo(overrides?: Partial<IConversationRepository>): IConversationRe
     insertMessage: vi.fn(),
     getUserConversations: vi.fn(),
     listAllConversations: vi.fn(() => []),
+    searchMessages: vi.fn(() => ({ items: [], total: 0, page: 0, pageSize: 20, hasMore: false })),
     ...overrides,
   };
 }
diff --git a/tests/unit/cronService.test.ts b/tests/unit/cronService.test.ts
@@ -232,7 +232,7 @@ describe('CronService', () => {
     // Advance exactly one interval period to fire the timer once.
     await vi.advanceTimersByTimeAsync(60000);
 
-    expect(executor.executeJob).toHaveBeenCalledWith(job);
+    expect(executor.executeJob).toHaveBeenCalledWith(job, expect.any(Function));
     expect(repo.update).toHaveBeenCalledWith(
       'j1',
       expect.objectContaining({ state: expect.objectContaining({ lastStatus: 'ok' }) })
diff --git a/vitest.config.ts b/vitest.config.ts
@@ -67,7 +67,6 @@ export default defineConfig({
         'src/process/bridge/conversationBridge.ts',
         // Decoupling layer (interfaces + implementations)
         'src/process/database/SqliteConversationRepository.ts',
-        'src/process/database/FallbackConversationRepository.ts',
         'src/process/database/SqliteChannelRepository.ts',
         'src/process/services/ConversationServiceImpl.ts',
         'src/process/services/cron/SqliteCronRepository.ts',

Original file line number	Diff line number	Diff line change
`@@ -16,6 +16,7 @@ function makeRepo(overrides?: Partial<IConversationRepository>): IConversationRe`
`16`	`16`	`insertMessage: vi.fn(),`
`17`	`17`	`getUserConversations: vi.fn(),`
`18`	`18`	`listAllConversations: vi.fn(() => []),`
	`19`	`+ searchMessages: vi.fn(() => ({ items: [], total: 0, page: 0, pageSize: 20, hasMore: false })),`
`19`	`20`	`...overrides,`
`20`	`21`	`};`
`21`	`22`	`}`