diff --git a/src/group-queue.test.ts b/src/group-queue.test.ts index 6201110..0e90223 100644 --- a/src/group-queue.test.ts +++ b/src/group-queue.test.ts @@ -123,6 +123,49 @@ describe('GroupQueue', () => { expect(processMessages).toHaveBeenCalledTimes(3); }); + it('reserves one foreground slot so queued tasks do not block new messages', async () => { + let resolveTaskOne!: () => void; + let resolveTaskTwo!: () => void; + let messageStarted = false; + + const processMessages = vi.fn(async (groupJid: string) => { + if (groupJid === 'group3@g.us') { + messageStarted = true; + } + return true; + }); + + queue.setProcessMessagesFn(processMessages); + + queue.enqueueTask('group1@g.us', 'task-1', async () => { + await new Promise((resolve) => { + resolveTaskOne = resolve; + }); + }); + await vi.advanceTimersByTimeAsync(10); + + queue.enqueueTask('group2@g.us', 'task-2', async () => { + await new Promise((resolve) => { + resolveTaskTwo = resolve; + }); + }); + queue.enqueueMessageCheck('group3@g.us'); + await vi.advanceTimersByTimeAsync(10); + + expect(messageStarted).toBe(true); + expect(processMessages).toHaveBeenCalledWith('group3@g.us', { + runId: expect.any(String), + reason: 'messages', + }); + + resolveTaskOne!(); + await vi.advanceTimersByTimeAsync(10); + expect(messageStarted).toBe(true); + + resolveTaskTwo!(); + await vi.advanceTimersByTimeAsync(10); + }); + // --- Tasks prioritized over messages --- it('drains tasks before messages for same group', async () => { diff --git a/src/group-queue.ts b/src/group-queue.ts index cbf6f96..18994ef 100644 --- a/src/group-queue.ts +++ b/src/group-queue.ts @@ -17,6 +17,8 @@ export interface GroupRunContext { const MAX_RETRIES = 5; const BASE_RETRY_MS = 5000; +const MAX_CONCURRENT_TASKS = + MAX_CONCURRENT_AGENTS > 1 ? MAX_CONCURRENT_AGENTS - 1 : 1; interface GroupState { active: boolean; @@ -46,6 +48,7 @@ export interface GroupStatus { export class GroupQueue { private groups = new Map(); private activeCount = 0; + private activeTaskCount = 0; private waitingGroups: string[] = []; private processMessagesFn: | ((groupJid: string, context: GroupRunContext) => Promise) @@ -156,14 +159,22 @@ export class GroupQueue { return; } - if (this.activeCount >= MAX_CONCURRENT_AGENTS) { + if ( + this.activeCount >= MAX_CONCURRENT_AGENTS || + this.activeTaskCount >= MAX_CONCURRENT_TASKS + ) { state.pendingTasks.push({ id: taskId, groupJid, fn }); if (!this.waitingGroups.includes(groupJid)) { this.waitingGroups.push(groupJid); } logger.debug( - { groupJid, taskId, activeCount: this.activeCount }, - 'At concurrency limit, task queued', + { + groupJid, + taskId, + activeCount: this.activeCount, + activeTaskCount: this.activeTaskCount, + }, + 'At task concurrency limit, task queued', ); return; } @@ -350,6 +361,7 @@ export class GroupQueue { state.runningTaskId = task.id; state.startedAt = Date.now(); this.activeCount++; + this.activeTaskCount++; logger.debug( { groupJid, taskId: task.id, activeCount: this.activeCount }, @@ -370,6 +382,7 @@ export class GroupQueue { state.processName = null; state.ipcDir = null; this.activeCount--; + this.activeTaskCount--; this.drainGroup(groupJid); } } @@ -449,11 +462,39 @@ export class GroupQueue { this.waitingGroups.length > 0 && this.activeCount < MAX_CONCURRENT_AGENTS ) { - const nextJid = this.waitingGroups.shift()!; + const nextMessageIndex = this.waitingGroups.findIndex((jid) => { + const state = this.getGroup(jid); + return state.pendingMessages; + }); + const nextIndex = + nextMessageIndex !== -1 + ? nextMessageIndex + : this.waitingGroups.findIndex((jid) => { + const state = this.getGroup(jid); + return ( + state.pendingTasks.length > 0 && + this.activeTaskCount < MAX_CONCURRENT_TASKS + ); + }); + + if (nextIndex === -1) { + return; + } + + const [nextJid] = this.waitingGroups.splice(nextIndex, 1); const state = this.getGroup(nextJid); - // Prioritize tasks over messages - if (state.pendingTasks.length > 0) { + if (state.pendingMessages) { + this.runForGroup(nextJid, 'drain').catch((err) => + logger.error( + { groupJid: nextJid, err }, + 'Unhandled error in runForGroup (waiting)', + ), + ); + } else if ( + state.pendingTasks.length > 0 && + this.activeTaskCount < MAX_CONCURRENT_TASKS + ) { const task = state.pendingTasks.shift()!; this.runTask(nextJid, task).catch((err) => logger.error( @@ -461,13 +502,6 @@ export class GroupQueue { 'Unhandled error in runTask (waiting)', ), ); - } else if (state.pendingMessages) { - this.runForGroup(nextJid, 'drain').catch((err) => - logger.error( - { groupJid: nextJid, err }, - 'Unhandled error in runForGroup (waiting)', - ), - ); } // If neither pending, skip this group }