diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 5c479f76c7..da71ea6152 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -16000,7 +16000,12 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { it("reorders rapid Slack callbacks by provider time before one conversation drain", async () => { const fixture = await seedCompany(); const runtime = new FakeChatSdkRuntime(); - const deferred: Array<() => void> = []; + const deferred: Array<() => void | Promise> = []; + let releaseFirstDelivery!: () => void; + let signalFirstDelivery!: () => void; + const firstDeliveryEntered = new Promise((resolve) => { signalFirstDelivery = resolve; }); + const firstDeliveryReleased = new Promise((resolve) => { releaseFirstDelivery = resolve; }); + let firstDelivery = true; const wakeup = vi.fn(async () => ({ accepted: true })); const service = chatChannelService(db, { deferWebhookProcessing: true, @@ -16009,6 +16014,12 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { publicBaseUrl: "https://paperclip.example", runtime: runtime as unknown as ChatSdkRuntime, scheduleDeferredWork: (task) => deferred.push(task), + reachAuthorizationBarrier: async () => { + if (!firstDelivery) return; + firstDelivery = false; + signalFirstDelivery(); + await firstDeliveryReleased; + }, }); const endpoint = await service.create( fixture.companyId, @@ -16127,69 +16138,54 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { publicBaseUrl: "https://paperclip.example", runtime: new FakeChatSdkRuntime() as unknown as ChatSdkRuntime, }); - deferred.shift()?.(); - // Simulate another server process reconciling the same durable rows at - // the same time as the webhook process's deferred drain. - await competingService.processPendingDeliveries(); - await vi.waitFor(async () => { - const rows = await db + let drainSettled = false; + const draining = Promise.resolve(deferred.shift()?.()).then(() => { drainSettled = true; }); + try { + // Hold the deferred owner after it acquires the conversation lease. The + // competing sweep must finish without overtaking that owner. + await firstDeliveryEntered; + await competingService.processPendingDeliveries(); + expect(drainSettled).toBe(false); + expect(wakeup.mock.calls.filter((call) => call[0] === fixture.assignedAgentId)).toHaveLength(0); + releaseFirstDelivery(); + await draining; + expect(drainSettled).toBe(true); + const conversations = await db .select() .from(chatConversations) .where(eq(chatConversations.endpointId, endpoint.id)); - expect(rows).toHaveLength(1); - }); - const [conversation] = await db - .select() - .from(chatConversations) - .where(eq(chatConversations.endpointId, endpoint.id)); - await vi.waitFor(async () => { - const rows = await db - .select({ id: issueComments.id }) + expect(conversations).toHaveLength(1); + const [conversation] = conversations; + const comments = await db + .select({ id: issueComments.id, body: issueComments.body }) .from(issueComments) - .where(eq(issueComments.issueId, conversation.issueId)); - expect(rows).toHaveLength(8); - }); - const comments = await db - .select({ id: issueComments.id, body: issueComments.body }) - .from(issueComments) - .where(eq(issueComments.issueId, conversation.issueId)) - .orderBy(asc(issueComments.createdAt), asc(issueComments.id)); - expect(comments.map((comment) => comment.body)).toEqual([ - "@maya acknowledge quickly", - "follow-up 3", - "follow-up 4", - "follow-up 5", - "and include the rollback status", - "follow-up 6", - "follow-up 7", - "follow-up 8", - ]); - // Comment admission commits before the durable wake. Wait for this - // company's last wake too, not merely its already-visible last comment. - // The competing sweep may legitimately reconcile another fixture company. - await vi.waitFor(() => { - const calls = wakeup.mock.calls.filter( - (call) => call[0] === fixture.assignedAgentId, - ); + .where(eq(issueComments.issueId, conversation.issueId)) + .orderBy(asc(issueComments.createdAt), asc(issueComments.id)); + expect(comments.map((comment) => comment.body)).toEqual([ + "@maya acknowledge quickly", + "follow-up 3", + "follow-up 4", + "follow-up 5", + "and include the rollback status", + "follow-up 6", + "follow-up 7", + "follow-up 8", + ]); + // Joining the owner includes all comment admissions, durable wakeups and + // lease release. These assertions do not race a one-second polling budget. + expect(comments).toHaveLength(8); + const calls = wakeup.mock.calls.filter((call) => call[0] === fixture.assignedAgentId); expect(calls).toHaveLength(8); - expect(calls.map((call) => call[1]?.payload?.wakeCommentId)).toEqual( - comments.map((comment) => comment.id), - ); - }); - // The last comment and wakeup commit inside the lease. Under full-suite - // load the assertions above can observe those effects one microtask before - // the deferred owner's `finally` deletes its lease. Require prompt eventual - // release; a real leak would remain for the much longer lease TTL. - await vi.waitFor(async () => { + expect(calls.map((call) => call[1]?.payload?.wakeCommentId)).toEqual(comments.map((comment) => comment.id)); expect( - await db - .select() - .from(chatEndpointLeases) - .where(eq(chatEndpointLeases.endpointId, endpoint.id)), + await db.select().from(chatEndpointLeases).where(eq(chatEndpointLeases.endpointId, endpoint.id)), ).toHaveLength(0); - }); - await competingService.shutdown(); - await service.shutdown(); + } finally { + releaseFirstDelivery(); + await draining; + await competingService.shutdown(); + await service.shutdown(); + } }); it("stops a conversation drain after its lease renewal fails", async () => { @@ -61921,8 +61917,8 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ).resolves.toMatchObject({ ok: true }); expect(deferred).toHaveLength(1); await drainDeferred(); - await vi.waitFor(() => expect(deferred).toHaveLength(1)); - await drainDeferred(); + // Awaiting each callback also drains the lifecycle work it queues. + expect(deferred).toHaveLength(0); await vi.waitFor(async () => { await expect( db diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index c8b2628a22..1fe4eac26f 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -1467,8 +1467,8 @@ export interface ChatChannelServiceOptions { /** Optional verified ingress origin; never used for board or identity links. */ webhookPublicBaseUrl?: string | null; runtime?: ChatSdkRuntime; - /** Testable scheduler hook; production defaults to the next event-loop turn. */ - scheduleDeferredWork?: (task: () => void) => void; + /** Testable scheduler hook; its callback settles after the tracked work finishes. */ + scheduleDeferredWork?: (task: () => void | Promise) => void; /** Test boundary after selecting due Slack status work and before claiming. */ slackSessionSyncSelectionBarrier?: () => Promise; /** Narrow fault-injection boundary for the one-time setup-secret audit. */ @@ -3441,6 +3441,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { }); backgroundMessageTasks.add(pending); void pending.finally(() => backgroundMessageTasks.delete(pending)); + return pending; }); }