From e733bddc45e00efb1819fffa8677f7b0d07802c5 Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Sun, 11 Oct 2026 06:08:05 -0500 Subject: [PATCH] perf(chat): stop polling provider action and session sync queues (#15822) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Chat stores pending provider actions in durable database rows. > - Three action queues still scan every second when empty. > - These scans keep idle databases active. > - This pull request wakes those queues after a commit or at their next deadline. > - Empty queues stop querying while existing recovery rules remain in place. ## Linked Issues or Issue Description Refs #15808, #15802, #15625. Related sleep work: #15750, #15757. **What existing behavior does this improve?** Provider messages and notices, accepted GitHub webhook receipts, and Slack session status synchronization. **Current behavior** Chat maintenance scans these three queues every second. It also rechecks paused Slack sync rows every 30 seconds. **Proposed behavior** Reuse the transaction-aware action workers and existing chat_actions rows. Wake after commits. Restore unfinished work on startup. Use stored retry and stale-claim deadlines. Stop scanning when empty. Park paused Slack sync rows until an endpoint change. ## What Changed - Add three topics to the existing action worker registration. No tables, migrations, triggers, or external services. - Signal producers, claims, retry updates, terminal settlements, and operator resolution inside their owning transaction. Provider I/O remains outside transactions. - Share pending-state and deadline predicates between dispatch and scheduling. Exclude terminal history and uncertain provider sends. - Preserve GitHub receipt attempt fences and provider-send quarantine. Preserve Slack's 30-minute refresh while a session is processing. - Remove these workers from periodic app maintenance. Update the sleep inventory and add regression coverage. ## Verification - Passed: 13 focused tests across scheduler, app wiring, and real PostgreSQL integration. Covers actual commit wakes with deferred fast paths disabled, GitHub retries and startup recovery, stale-send quarantine, Slack publication-to-sync delivery, paused deadlines, and idle queues. - Passed: pnpm typecheck. - Passed: full chat integration and related coordinator suites, **1,274 tests across five files**. - Passed: pnpm build. - Greptile: **5/5**, no actionable findings, completed on `dc284cd275e5f6df624d60b27954421f2bcdbee2`. - Hosted CI: **54 checks passed**, two intentionally skipped Storybook checks, no failures or pending checks on `dc284cd275e5f6df624d60b27954421f2bcdbee2`. Provider responses are simulated. PostgreSQL, transactions, service dispatch, and scheduling are real. No live Slack or GitHub App was used. ## Risks - Writers must emit the existing process-local transaction signal. Independent database clients and direct SQL still need explicit integration or startup recovery. - Accepted GitHub receipts are covered. Finding callbacks GitHub never delivered still uses the separate provider recovery lane. - Other chat maintenance and persistent transports still run. This does not establish whole-instance sleep. - Existing claim, revision, credential, and authorization checks remain authoritative during concurrent foreground and scheduled dispatch. ## Model Used OpenAI Codex (GPT-6), with repository inspection, code execution, and automated tests. ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [x] All Paperclip CI gates are green - [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge Co-authored-by: Paperclip --- ...-instance-sleep-and-scheduler-inventory.md | 16 +- .../app-private-hostname-gate.test.ts | 6 + server/src/__tests__/chat-action-work.test.ts | 7 +- .../chat-channels.integration.test.ts | 171 ++++++++- server/src/app.ts | 6 +- server/src/services/chat-action-work.ts | 18 +- server/src/services/chat-channels.ts | 341 +++++++++--------- .../src/services/chat-slack-session-outbox.ts | 3 + .../src/services/chat-work-notifications.ts | 3 + .../services/delivery-work-notifications.ts | 3 + 10 files changed, 382 insertions(+), 192 deletions(-) diff --git a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md index d818caa41c6..e40716791f3 100644 --- a/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md +++ b/doc/plans/2026-10-08-instance-sleep-and-scheduler-inventory.md @@ -434,9 +434,19 @@ Longer polling intervals alone do not complete an item. Startup restores existing rows, and empty workers disarm. No new tables. Tests cover an idle hour, real producer notifications, PostgreSQL deadline selection, stale claims, outer-transaction wakes, and callback-driven setup. - Remaining **1s** maintenance: provider runtimes, other provider action - outboxes, run milestone/wake-notice projection, GitHub operation recovery, - Teams protocol maintenance, and Slack session sync. GitHub pre-ingress recovery + Three more existing action queues now use commit/deadline workers: provider + messages/notices, accepted GitHub webhook receipts, and Slack session sync. + Foreground claims, retries, operator resolution, and settlement writes signal + in their transaction. Shared predicates select pending rows and deadlines; + terminal history and uncertain provider sends have no deadline. Stale provider + sends remain quarantined, while GitHub receipts retain bounded replay and + their existing attempt fences. Slack sync retains the 30-minute refresh for + running sessions and pauses without rescheduling until an endpoint change. + Tests exercise real producers with deferred fast paths disabled, retry and + startup recovery, stale-send quarantine, and zero scans after settlement. + Remaining **1s** maintenance: provider runtimes, Telegram maintenance, failed + chat-run retries, run milestone/wake-notice projection, GitHub operation + recovery, and Teams protocol maintenance. GitHub pre-ingress recovery must still inspect provider history: a lost callback has no local queue row. This slice does **not** yet make the whole chat subsystem quiet. Independent DB roots/direct SQL require startup recovery, as with the other migrated queues. diff --git a/server/src/__tests__/app-private-hostname-gate.test.ts b/server/src/__tests__/app-private-hostname-gate.test.ts index 61893417dfd..8f087241913 100644 --- a/server/src/__tests__/app-private-hostname-gate.test.ts +++ b/server/src/__tests__/app-private-hostname-gate.test.ts @@ -106,6 +106,12 @@ describe("createChatReconciliationCoordinator", () => { // The service integration tests hold real publication workers while this // scheduled method returns; app shutdown must also join those workers. expect(source).toContain("await chatChannels.shutdown()"); + expect(source).not.toContain("chatChannels.processPendingSlackSessionSyncs()"); + const service = readFileSync(new URL("../services/chat-channels.ts", import.meta.url), "utf8"); + const actionMaintenance = service.slice(service.indexOf("async function processPendingChatMaintenance("), + service.indexOf("async function processPendingDeliveries(")); + expect(actionMaintenance).not.toContain("processPendingProviderEffects("); + expect(actionMaintenance).not.toContain("processPendingGitHubWebhookIngress("); }); it.each([false, true])( diff --git a/server/src/__tests__/chat-action-work.test.ts b/server/src/__tests__/chat-action-work.test.ts index af541a60c07..b96541e79ee 100644 --- a/server/src/__tests__/chat-action-work.test.ts +++ b/server/src/__tests__/chat-action-work.test.ts @@ -19,6 +19,9 @@ function setup() { const coordinator = createDeliveryWorkCoordinator({ owner, canRun: enabled, canReconcile: enabled, onError }); coordinators.push(coordinator); const service = { + processPendingProviderEffects: vi.fn(async () => {}), + processPendingGitHubWebhookIngress: vi.fn(async () => {}), + processPendingSlackSessionSyncs: vi.fn(async () => {}), processPendingSlackBoardMessages: vi.fn(async () => {}), processPendingSlackTaskStarts: vi.fn(async () => {}), processPendingReceiptReactions: vi.fn(async () => {}), @@ -27,7 +30,7 @@ function setup() { nextChatActionAt: vi.fn(async (): Promise => null), nextVerificationMessageAt: vi.fn(async (): Promise => null), }; - const runs = [service.processPendingSlackBoardMessages, service.processPendingSlackTaskStarts, + const runs = [service.processPendingProviderEffects, service.processPendingGitHubWebhookIngress, service.processPendingSlackSessionSyncs, service.processPendingSlackBoardMessages, service.processPendingSlackTaskStarts, service.processPendingReceiptReactions, service.processPendingSlackSessionStops, service.processPendingVerificationMessages]; return { owner, service, runs, coordinator, enabled, onError, start: () => registerChatActionWork(coordinator, service, enabled) }; } @@ -37,7 +40,7 @@ describe("chat action scheduling", () => { await s.start().ready; await vi.advanceTimersByTimeAsync(3_600_000); for (const run of s.runs) expect(run).toHaveBeenCalledTimes(1); - expect(s.service.nextChatActionAt).toHaveBeenCalledTimes(4); + expect(s.service.nextChatActionAt).toHaveBeenCalledTimes(7); expect(s.service.nextVerificationMessageAt).toHaveBeenCalledTimes(1); expect(s.coordinator.nextWakeAt()).toBeNull(); expect(vi.getTimerCount()).toBe(0); diff --git a/server/src/__tests__/chat-channels.integration.test.ts b/server/src/__tests__/chat-channels.integration.test.ts index 118b47a8a81..a7d9fad74fa 100644 --- a/server/src/__tests__/chat-channels.integration.test.ts +++ b/server/src/__tests__/chat-channels.integration.test.ts @@ -13250,6 +13250,49 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await service.shutdown(); }); + it("drains accepted GitHub webhooks from commit wakes and durable retry deadlines", async () => { + const fixture = await seedCompany(); + const { endpoint, runtime, service, webhookSecret } = await configuredGitHubEndpoint(fixture, { + deferWebhookProcessing: true, scheduleDeferredWork: () => undefined, + }); + const providerRuntime = runtime.endpoints.get(endpoint.id)!; + let attempts = 0; + providerRuntime.webhookHook = async () => { + providerRuntime.webhookResponse = new Response("result", { status: ++attempts === 1 ? 503 : 202 }); + }; + const delivery = randomUUID(); + const send = () => service.handleWebhook(endpoint.publicId, "github", signedGitHubWebhookRequest({ + delivery, event: "installation_repositories", payload: { action: "added", installation: { id: 2468 } }, webhookSecret, + })); + const onError = vi.fn(); + let coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + const receipt = () => db.select().from(chatActions).where(and(eq(chatActions.endpointId, endpoint.id), + eq(chatActions.providerActionId, `github_webhook_ingress:${delivery}`))).then(rows => rows[0]); + try { + await registerChatActionWork(coordinator, service, () => true).ready; + expect((await send()).status).toBe(202); + await expect.poll(async () => (await receipt())?.status, { timeout: 10_000 }).toBe("processed"); + expect(attempts).toBe(2); + expect((await receipt())?.result).toMatchObject({ attempts: 2, httpStatus: 202 }); + // A duplicate callback wakes the lane but cannot repeat a settled effect. + await send(); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 }); + expect(attempts).toBe(2); + await coordinator.stop(); + // Work committed while no worker exists is recovered by the next startup. + const restartDelivery = randomUUID(); + await service.handleWebhook(endpoint.publicId, "github", signedGitHubWebhookRequest({ + delivery: restartDelivery, event: "installation_repositories", + payload: { action: "added", installation: { id: 2468 } }, webhookSecret, + })); + expect(attempts).toBe(2); + coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + await registerChatActionWork(coordinator, service, () => true).ready; + await expect.poll(() => attempts, { timeout: 10_000 }).toBe(3); + expect(onError).not.toHaveBeenCalled(); + } finally { await coordinator.stop(); } + }); + it("safely re-arms an exact terminal GitHub delivery after an operator redelivery", async () => { const fixture = await seedCompany(); const { endpoint, runtime, service, webhookSecret } = @@ -32819,7 +32862,7 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { .update(chatEndpoints) .set({ status: "active" }) .where(eq(chatEndpoints.id, endpoint.id)); - const channelId = `C-LANE-${endpoint.id}`; + const channelId = `C${endpoint.id.replaceAll("-", "").toUpperCase()}`; await db.insert(chatEndpointResources).values({ companyId: fixture.companyId, endpointId: endpoint.id, @@ -32892,6 +32935,45 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { return { ...fixture, storage, runtime, service, lanes, enqueue, cleanup }; } + it("wakes Slack session sync from publication commit and leaves settled action queues quiet", async () => { + const f = await publicationLaneFixture(1); + const lane = f.lanes[0]!; + const onError = vi.fn(); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError }); + try { + await registerChatActionWork(coordinator, f.service, () => true).ready; + await f.enqueue(0, "A completed answer"); + await f.service.processPendingPublications(); + await expect.poll(async () => db.select({ status: chatActions.status }).from(chatActions) + .where(and(eq(chatActions.endpointId, lane.endpoint.id), eq(chatActions.kind, "slack_session_sync"))), + { timeout: 10_000 }).toEqual([{ status: "processed" }]); + await vi.waitFor(() => expect(coordinator.nextWakeAt()).toBeNull(), { timeout: 10_000 }); + const scans = vi.spyOn(db, "select"); + try { + await new Promise(resolve => setTimeout(resolve, 1200)); + expect(scans).not.toHaveBeenCalled(); + } finally { scans.mockRestore(); } + expect(onError).not.toHaveBeenCalled(); + } finally { await coordinator.stop(); await f.cleanup(); } + }); + + it("recovers stale provider sends on startup without replaying uncertain effects", async () => { + const f = await publicationLaneFixture(1); + const lane = f.lanes[0]!; + const [action] = await db.insert(chatActions).values({ companyId: f.companyId, endpointId: lane.endpoint.id, + kind: "provider_effect", providerActionId: randomUUID(), status: "processing", result: { attempts: 1 }, + updatedAt: new Date(Date.now() - 120_000), payload: {} }).returning(); + const before = lane.providerRuntime.posts.length; + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true }); + try { + await registerChatActionWork(coordinator, f.service, () => true).ready; + await expect.poll(async () => db.select({ status: chatActions.status }).from(chatActions) + .where(eq(chatActions.id, action!.id)), { timeout: 10_000 }).toEqual([{ status: "delivery_unknown" }]); + expect(await f.service.nextChatActionAt("provider_effect")).toBeNull(); + expect(lane.providerRuntime.posts).toHaveLength(before); + } finally { await coordinator.stop(); await f.cleanup(); } + }); + it("restores action retry and stale-claim deadlines and parks paused or uncertain work", async () => { const f = await publicationLaneFixture(1); const lane = f.lanes[0]!; @@ -32929,6 +33011,21 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { await set("delivery_unknown"); expect(await f.service.nextChatActionAt(kind)).toBeNull(); expect(await due()).toBe(false); + } else if (kind === "slack_session_sync") { + expect(await due()).toBe(true); + await set("received", { retryAt: later.toISOString() }); + expect(await f.service.nextChatActionAt(kind)).toBe(later.getTime()); + expect(await due()).toBe(false); + await set("processing"); + expect(await f.service.nextChatActionAt(kind)).toBe(at.getTime() + 60_000); + for (const status of ["paused", "attention"]) { + await db.update(chatEndpoints).set({ status }).where(eq(chatEndpoints.id, lane.endpoint.id)); + expect(await f.service.nextChatActionAt(kind)).toBeNull(); + expect(await due()).toBe(false); + } + await db.update(chatEndpoints).set({ status: "active" }).where(eq(chatEndpoints.id, lane.endpoint.id)); + await set("failed", { retryable: true }); + expect(await f.service.nextChatActionAt(kind)).toBeNull(); } else { expect(await due()).toBe(true); await set("processing"); @@ -53999,6 +54096,75 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { ); }); + it("delivers queued provider notices from commit wakes without the deferred fast path", async () => { + const fixture = await seedCompany(); + const { callbacks, endpoint, service, runtime } = + await configuredSlackEndpoint(fixture, { scheduleDeferredWork: () => undefined }); + if (!callbacks.onSlashCommand) { + throw new Error("Slack slash command callback was not registered"); + } + await db.insert(chatEndpointResources).values({ + companyId: fixture.companyId, + endpointId: endpoint.id, + type: "channel", + providerResourceId: "C-CONTROL-GUIDANCE", + label: "control-guidance", + availability: "available", + enabled: true, + }); + const command = endpoint.setup.command; + if (!command) throw new Error("Slack endpoint did not expose its command"); + const post = vi.fn(async () => ({ + id: "unexpected-channel-control-post", + threadId: "slack:C-CONTROL-GUIDANCE:root", + })); + const postEphemeral = vi.fn(async () => ({ + id: "channel-control-guidance", + threadId: "slack:C-CONTROL-GUIDANCE:root", + })); + const providerRuntime = runtime.endpoints.get(endpoint.id)!; + vi.spyOn(providerRuntime, "thread").mockReturnValue({ post, postEphemeral } as never); + const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true }); + onTestFinished(async () => { await coordinator.stop(); }); + await registerChatActionWork(coordinator, service, () => true).ready; + for (const control of ["status"] as const) { + await callbacks.onSlashCommand({ + endpointId: endpoint.id, + provider: "slack", + event: { + channel: { + id: "slack:C-CONTROL-GUIDANCE", + name: "control-guidance", + isDM: false, + post, + postEphemeral, + } as never, + command, + text: control, + triggerId: `channel-control-${control}`, + user: { + userId: "U-CHANNEL-CONTROLLER", + userName: "channel-controller", + fullName: "Channel Controller", + isBot: false, + isMe: false, + isSystem: false, + }, + raw: { command, text: control }, + adapter: {} as never, + openModal: async () => undefined, + }, + }); + } + await expect.poll(async () => db.select({ status: chatActions.status }).from(chatActions) + .where(and(eq(chatActions.endpointId, endpoint.id), eq(chatActions.kind, "provider_effect"))), + { timeout: 10_000 }).toEqual([{ status: "processed" }]); + expect(postEphemeral).toHaveBeenCalledTimes(1); + expect(post).not.toHaveBeenCalled(); + expect(await service.nextChatActionAt("provider_effect")).toBeNull(); + await coordinator.stop(); + }); + it("returns ephemeral guidance for exact Slack controls in channels without creating tasks or actions", async () => { const fixture = await seedCompany(); const { callbacks, endpoint, service } = @@ -54438,6 +54604,9 @@ describeEmbeddedPostgres("chat channel control-plane integration", () => { const fixture = await seedCompany(); const { callbacks, endpoint, runtime, service } = await configuredTelegramEndpoint(fixture); + const workWake = vi.fn(); + const stopWorkObserver = subscribeDeliveryWork(db, DELIVERY_QUEUES.chatProviderEffects, workWake); + onTestFinished(() => { stopWorkObserver(); expect(workWake).toHaveBeenCalled(); }); if (!callbacks.onSlashCommand) { throw new Error("Telegram slash command callback was not registered"); } diff --git a/server/src/app.ts b/server/src/app.ts index 003d587c551..238a5bad698 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -326,7 +326,7 @@ export function createChatReconciliationCoordinator(input: { projectRunMilestones: () => Promise; flushPublications: () => Promise; processPendingSlackFileUploadReceipts?: () => Promise; - processPendingSlackSessionSyncs: () => Promise; + processPendingSlackSessionSyncs?: () => Promise; onError: (lane: ChatReconciliationLane, error: unknown) => void; }) { let stopped = false; @@ -373,7 +373,7 @@ export function createChatReconciliationCoordinator(input: { milestoneReconciliation.poll(); publicationReconciliation.poll(); if (input.processPendingSlackFileUploadReceipts) start("Slack file receipts", input.processPendingSlackFileUploadReceipts); - start("Slack session status", input.processPendingSlackSessionSyncs); + if (input.processPendingSlackSessionSyncs) start("Slack session status", input.processPendingSlackSessionSyncs); }, notifyPublications() { milestoneReconciliation.notify(); @@ -1271,8 +1271,6 @@ export async function createApp( publicBaseUrl: opts.authPublicBaseUrl, }), flushPublications: () => reconcileChatPublicationMaintenance(), - processPendingSlackSessionSyncs: () => - chatChannels.processPendingSlackSessionSyncs(), onError: (lane, err) => { logger.error({ err, lane }, `Failed to reconcile chat ${lane}`); }, diff --git a/server/src/services/chat-action-work.ts b/server/src/services/chat-action-work.ts index 42993950548..f794cc6a0c2 100644 --- a/server/src/services/chat-action-work.ts +++ b/server/src/services/chat-action-work.ts @@ -23,6 +23,10 @@ export function chatActionDeadline(kind: ScheduledChatAction) { when ${chatActions.status} = 'admitting' then ${updated} + ${SLACK_COMMAND_ADMISSION_STALE_MS} when ${chatActions.status} = 'resolving' then ${updated} + ${SLACK_COMMAND_EXPLICIT_RETRY_STALE_MS} when ${chatActions.status} = 'validating' then ${updated} + ${SLACK_COMMAND_POST_STALE_MS} end`; + if (kind === "slack_session_sync") return sql`case + when ${chatEndpoints.status} in ('paused', 'attention') then null + when ${chatActions.status} = 'received' then ${retry} + when ${chatActions.status} = 'processing' then ${updated} + ${PROVIDER_EFFECT_STALE_MS} end`; return sql`case when ${chatActions.status} = 'received' then 0 when ${chatActions.status} = 'processing' then ${updated} + ${PROVIDER_EFFECT_STALE_MS} @@ -30,13 +34,15 @@ export function chatActionDeadline(kind: ScheduledChatAction) { } function pendingChatAction(kind: ScheduledChatAction) { - // Filter history before the endpoint join and aggregate. Failed reactions - // and stops only remain live when the saved result permits another attempt. + // Filter history before the endpoint join and aggregate. Failed actions + // only remain live when the saved result permits another attempt. const status = kind === "slack_board_message" ? eq(chatActions.status, "received") : kind === "slash_task_start" ? inArray(chatActions.status, ["queued", "provider_confirmed", "admitting", "resolving", "validating"]) - : or(inArray(chatActions.status, ["received", "processing"]), + : kind === "slack_session_sync" + ? inArray(chatActions.status, ["received", "processing"]) + : or(inArray(chatActions.status, ["received", "processing"]), and(eq(chatActions.status, "failed"), sql`${chatActions.result}->>'retryable' = 'true'`)); return and(eq(chatActions.kind, kind), status); } @@ -54,6 +60,9 @@ export async function nextChatActionAt(db: Db, kind: ScheduledChatAction): Promi } type ChatActionService = { + processPendingProviderEffects(): Promise; + processPendingGitHubWebhookIngress(): Promise; + processPendingSlackSessionSyncs(): Promise; processPendingSlackBoardMessages(): Promise; processPendingSlackTaskStarts(): Promise; processPendingReceiptReactions(): Promise; @@ -69,6 +78,9 @@ export function registerChatActionWork( canRun: () => boolean, ) { const actions = [ + { kind: "provider_effect", run: service.processPendingProviderEffects }, + { kind: "github_webhook_ingress", run: service.processPendingGitHubWebhookIngress }, + { kind: "slack_session_sync", run: service.processPendingSlackSessionSyncs }, { kind: "slack_board_message", run: service.processPendingSlackBoardMessages }, { kind: "slash_task_start", run: service.processPendingSlackTaskStarts }, { kind: "receipt_reaction", run: service.processPendingReceiptReactions }, diff --git a/server/src/services/chat-channels.ts b/server/src/services/chat-channels.ts index d81f4ea8744..12bff8621d9 100644 --- a/server/src/services/chat-channels.ts +++ b/server/src/services/chat-channels.ts @@ -4581,6 +4581,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { runtimeContext: LifecycleRuntimeFence; }, ) { + await notifyChatActionWork(database, "provider_effect"); const [inserted] = await database .insert(chatActions) .values({ @@ -4718,6 +4719,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { ["verifying", "active"], ) : null; + await notifyChatActionWork(tx, "provider_effect"); const [ownedAction] = await tx .update(chatActions) .set({ @@ -4819,6 +4821,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const failure = "Provider delivery could not be confirmed after the worker stopped. The effect will not be replayed automatically."; await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "provider_effect"); const [quarantined] = await tx .update(chatActions) .set({ @@ -5093,6 +5096,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { action: typeof chatActions.$inferSelect, attempt: number, ): Promise { + await notifyChatActionWork(tx, "provider_effect"); const [cancelled] = await tx .update(chatActions) .set({ @@ -5161,14 +5165,17 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { if (!action) return "failed"; const payload = providerEffectPayload(action.payload); if (!payload) { - await db - .update(chatActions) - .set({ - status: "failed", - result: { code: "provider_effect_payload_invalid" }, - updatedAt: new Date(), - }) - .where(eq(chatActions.id, action.id)); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "provider_effect"); + return tx + .update(chatActions) + .set({ + status: "failed", + result: { code: "provider_effect_payload_invalid" }, + updatedAt: new Date(), + }) + .where(eq(chatActions.id, action.id)); + }); return "failed"; } if (action.status === "processed") return "processed"; @@ -5238,6 +5245,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { payload, ) : false; + await notifyChatActionWork(tx, "provider_effect"); const [transportClaim] = await tx .update(chatActions) .set({ @@ -5355,6 +5363,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { }, ["verifying", "active"], ); + await notifyChatActionWork(tx, "provider_effect"); const [completed] = await tx .update(chatActions) .set({ @@ -5481,23 +5490,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const actions = await db .select() .from(chatActions) - .where( - and( - eq(chatActions.kind, "provider_effect"), - or( - eq(chatActions.status, "received"), - and( - eq(chatActions.status, "processing"), - lte(chatActions.updatedAt, staleBefore), - ), - and( - eq(chatActions.status, "failed"), - sql`coalesce(${chatActions.result}->>'retryable', 'false') = 'true'`, - sql`(${chatActions.result}->>'retryAt' is null or (${chatActions.result}->>'retryAt')::timestamptz <= ${now.toISOString()}::timestamptz)`, - ), - ), - ), - ) + .where(dueChatAction("provider_effect", now.getTime())) .orderBy(asc(chatActions.createdAt)) .limit(limit); for (const action of actions) { @@ -25203,6 +25196,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { runtimeGeneration: context.generation, credentialFingerprint: context.credentialFingerprint, }; + await notifyChatActionWork(tx, "github_webhook_ingress"); const inserted = await tx .insert(chatActions) .values({ @@ -25229,6 +25223,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { // currently authenticated body after a terminal processing failure. // Successful, cancelled, retryable, and in-flight receipts remain // immutable deduplication fences. + await notifyChatActionWork(tx, "github_webhook_ingress"); const [rearmed] = await tx .update(chatActions) .set({ @@ -25312,32 +25307,35 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { Date.now() + Math.min(60_000, 1_000 * 2 ** Math.max(0, input.attempts - 1)), ); - await db - .update(chatActions) - .set({ - status: cancelled ? "cancelled" : processed ? "processed" : "failed", - payload: terminal - ? redactedGitHubWebhookIngressPayload(input.payload) - : input.payload, - result: { - attempts: input.attempts, - ...(input.cancelledCode ? { code: input.cancelledCode } : {}), - ...(input.response - ? { httpStatus: input.response.status } - : { error: redactError(input.error) }), - retryable: !terminal, - ...(!terminal ? { retryAt: retryAt.toISOString() } : {}), - }, - updatedAt: new Date(), - }) - .where( - and( - eq(chatActions.id, actionId), - eq(chatActions.kind, "github_webhook_ingress"), - eq(chatActions.status, "processing"), - sql`(${chatActions.result}->>'attempts')::int = ${input.attempts}`, - ), - ); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "github_webhook_ingress"); + return tx + .update(chatActions) + .set({ + status: cancelled ? "cancelled" : processed ? "processed" : "failed", + payload: terminal + ? redactedGitHubWebhookIngressPayload(input.payload) + : input.payload, + result: { + attempts: input.attempts, + ...(input.cancelledCode ? { code: input.cancelledCode } : {}), + ...(input.response + ? { httpStatus: input.response.status } + : { error: redactError(input.error) }), + retryable: !terminal, + ...(!terminal ? { retryAt: retryAt.toISOString() } : {}), + }, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.id, actionId), + eq(chatActions.kind, "github_webhook_ingress"), + eq(chatActions.status, "processing"), + sql`(${chatActions.result}->>'attempts')::int = ${input.attempts}`, + ), + ); + }); } async function processGitHubWebhookIngress( @@ -25361,18 +25359,21 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { return githubIngressAcceptedResponse(); const payload = githubWebhookIngressPayload(action.payload); if (!payload) { - await db - .update(chatActions) - .set({ - status: "failed", - payload: {}, - result: { - code: "github_webhook_ingress_payload_invalid", - retryable: false, - }, - updatedAt: new Date(), - }) - .where(eq(chatActions.id, action.id)); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "github_webhook_ingress"); + return tx + .update(chatActions) + .set({ + status: "failed", + payload: {}, + result: { + code: "github_webhook_ingress_payload_invalid", + retryable: false, + }, + updatedAt: new Date(), + }) + .where(eq(chatActions.id, action.id)); + }); return githubIngressAcceptedResponse(); } const now = new Date(); @@ -25407,21 +25408,24 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { sql`coalesce((${chatActions.result}->>'attempts')::int, 0) = ${previousAttempts}`, ) : eq(chatActions.status, "received"); - const [claimed] = await db - .update(chatActions) - .set({ - status: "processing", - result: { attempts }, - updatedAt: now, - }) - .where( - and( - eq(chatActions.id, action.id), - eq(chatActions.kind, "github_webhook_ingress"), - claimGuard, - ), - ) - .returning(); + const [claimed] = await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "github_webhook_ingress"); + return tx + .update(chatActions) + .set({ + status: "processing", + result: { attempts }, + updatedAt: now, + }) + .where( + and( + eq(chatActions.id, action.id), + eq(chatActions.kind, "github_webhook_ingress"), + claimGuard, + ), + ) + .returning(); + }); if (!claimed) return null; action = claimed; @@ -26192,28 +26196,13 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { onlyActionId?: string, ) { const now = new Date(); - const staleBefore = new Date(now.getTime() - PROVIDER_EFFECT_STALE_MS); const actions = await db .select({ id: chatActions.id }) .from(chatActions) - .where( - and( - eq(chatActions.kind, "github_webhook_ingress"), - ...(onlyActionId ? [eq(chatActions.id, onlyActionId)] : []), - or( - eq(chatActions.status, "received"), - and( - eq(chatActions.status, "processing"), - lte(chatActions.updatedAt, staleBefore), - ), - and( - eq(chatActions.status, "failed"), - sql`coalesce(${chatActions.result}->>'retryable', 'false') = 'true'`, - sql`(${chatActions.result}->>'retryAt' is null or (${chatActions.result}->>'retryAt')::timestamptz <= ${now.toISOString()}::timestamptz)`, - ), - ), - ), - ) + .where(and( + onlyActionId ? eq(chatActions.id, onlyActionId) : undefined, + dueChatAction("github_webhook_ingress", now.getTime()), + )) .orderBy(asc(chatActions.createdAt)) .limit(limit); for (const action of actions) await processGitHubWebhookIngress(action.id); @@ -28092,8 +28081,6 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { async function processPendingChatMaintenance(limit = 25) { return Promise.allSettled([ - processPendingGitHubWebhookIngress(limit), - processPendingProviderEffects(limit), processPendingTelegramMaintenance(limit), processFailedChatRunRetries(limit), ]); @@ -28106,7 +28093,8 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { // so a crashed processing claim is quarantined before the delivery worker // could otherwise replay it. const actionRecovery = onlyDeliveryId || !maintenance ? null : Promise.allSettled([ - processPendingChatMaintenance(limit), processPendingSlackBoardMessages(limit), + processPendingChatMaintenance(limit), processPendingGitHubWebhookIngress(limit), + processPendingProviderEffects(limit), processPendingSlackBoardMessages(limit), processPendingSlackTaskStarts(limit), processPendingReceiptReactions(limit), processPendingSlackSessionStops(limit), slackRegistration.processPendingVerificationMessages(limit), ]); @@ -30095,6 +30083,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { ); } const now = new Date(); + await notifyChatActionWork(tx, "provider_effect"); const [claimed] = await tx .update(chatActions) .set({ @@ -30206,6 +30195,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { const now = new Date(); const nextStatus = resolution === "mark_delivered" ? "processed" : "cancelled"; + await notifyChatActionWork(tx, "provider_effect"); const [resolved] = await tx .update(chatActions) .set({ @@ -36931,43 +36921,36 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { ) { const now = new Date(); const actions = await db - .select() + .select({ action: chatActions }) .from(chatActions) - .where( - and( - eq(chatActions.kind, "slack_session_sync"), - ...(onlyActionId ? [eq(chatActions.id, onlyActionId)] : []), - or( - and( - eq(chatActions.status, "received"), - sql`(${chatActions.result}->>'retryAt' is null or (${chatActions.result}->>'retryAt')::timestamptz <= ${now.toISOString()}::timestamptz)`, - ), - and( - eq(chatActions.status, "processing"), - lte(chatActions.updatedAt, new Date(now.getTime() - 60_000)), - ), - ), - ), - ) + .leftJoin(chatEndpoints, and(eq(chatEndpoints.id, chatActions.endpointId), eq(chatEndpoints.companyId, chatActions.companyId))) + .where(and( + onlyActionId ? eq(chatActions.id, onlyActionId) : undefined, + dueChatAction("slack_session_sync", now.getTime()), + )) .orderBy(asc(chatActions.updatedAt)) - .limit(limit); + .limit(limit) + .then(rows => rows.map(row => row.action)); await options.slackSessionSyncSelectionBarrier?.(); for (const action of actions) { const payload = slackSessionSyncPayload(action.payload); if (!payload || !action.conversationId) { - await db - .update(chatActions) - .set({ - status: "failed", - result: { code: "slack_session_payload_invalid" }, - updatedAt: new Date(), - }) - .where( - and( - eq(chatActions.id, action.id), - eq(chatActions.updatedAt, action.updatedAt), - ), - ); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slack_session_sync"); + return tx + .update(chatActions) + .set({ + status: "failed", + result: { code: "slack_session_payload_invalid" }, + updatedAt: new Date(), + }) + .where( + and( + eq(chatActions.id, action.id), + eq(chatActions.updatedAt, action.updatedAt), + ), + ); + }); continue; } const revisionWhere = and( @@ -36980,26 +36963,19 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { eq(chatActions.updatedAt, action.updatedAt), ); const record = await endpointRecord(action.endpointId); - if (record && ["paused", "attention"].includes(record.endpoint.status)) { - await db - .update(chatActions) - .set({ - status: "received", - result: { retryAt: new Date(Date.now() + 30_000).toISOString() }, - updatedAt: new Date(), - }) - .where(selectedSnapshotWhere); - continue; - } + if (record && ["paused", "attention"].includes(record.endpoint.status)) continue; if (!record || record.endpoint.provider !== "slack") { - await db - .update(chatActions) - .set({ - status: "cancelled", - result: { code: "slack_session_endpoint_unavailable" }, - updatedAt: new Date(), - }) - .where(selectedSnapshotWhere); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slack_session_sync"); + return tx + .update(chatActions) + .set({ + status: "cancelled", + result: { code: "slack_session_endpoint_unavailable" }, + updatedAt: new Date(), + }) + .where(selectedSnapshotWhere); + }); continue; } const attempts = @@ -37084,6 +37060,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { ? endpoint.allowDirectMessages : nonDirectDestinationAllowed(endpoint, resource)) ) { + await notifyChatActionWork(tx, "slack_session_sync"); await tx .update(chatActions) .set({ @@ -37118,6 +37095,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { .limit(1) .then((rows) => rows[0] ?? null); if (!publication) { + await notifyChatActionWork(tx, "slack_session_sync"); await tx .update(chatActions) .set({ @@ -37150,6 +37128,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { if (!run || !["queued", "running"].includes(run.status)) status = "active"; } + await notifyChatActionWork(tx, "slack_session_sync"); const claimed = await tx .update(chatActions) .set({ @@ -37195,6 +37174,7 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { outcome === "updated" && claim.status === "processing" ? 30 * 60_000 : null; + await notifyChatActionWork(tx, "slack_session_sync"); await tx .update(chatActions) .set({ @@ -37232,35 +37212,38 @@ export function chatChannelService(db: Db, options: ChatChannelServiceOptions) { disposition.kind === "retry" ? disposition.retryAfterMs : Math.min(5 * 60_000, 1_000 * 2 ** Math.min(attempts, 8)); - await db - .update(chatActions) - .set({ - status: retryable ? "received" : "failed", - result: { - attempts, - code: retryable - ? "slack_session_sync_retry" - : "slack_session_sync_rejected", - ...(retryable - ? { retryAt: new Date(Date.now() + retryMs).toISOString() } - : {}), - }, - updatedAt: new Date(), - }) - .where( - and( - revisionWhere, - ownsAttempt - ? and( - eq(chatActions.status, "processing"), - sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`, - ) - : and( - eq(chatActions.status, "received"), - eq(chatActions.updatedAt, action.updatedAt), - ), - ), - ); + await db.transaction(async (tx) => { + await notifyChatActionWork(tx, "slack_session_sync"); + return tx + .update(chatActions) + .set({ + status: retryable ? "received" : "failed", + result: { + attempts, + code: retryable + ? "slack_session_sync_retry" + : "slack_session_sync_rejected", + ...(retryable + ? { retryAt: new Date(Date.now() + retryMs).toISOString() } + : {}), + }, + updatedAt: new Date(), + }) + .where( + and( + revisionWhere, + ownsAttempt + ? and( + eq(chatActions.status, "processing"), + sql`${chatActions.result}->>'ownerToken' = ${ownerToken}`, + ) + : and( + eq(chatActions.status, "received"), + eq(chatActions.updatedAt, action.updatedAt), + ), + ), + ); + }); logger.warn( { endpointId: action.endpointId, diff --git a/server/src/services/chat-slack-session-outbox.ts b/server/src/services/chat-slack-session-outbox.ts index 4be13a63e46..21b4a0d64a8 100644 --- a/server/src/services/chat-slack-session-outbox.ts +++ b/server/src/services/chat-slack-session-outbox.ts @@ -1,6 +1,8 @@ import { sql } from "drizzle-orm"; import { chatActions, type Db } from "@paperclipai/db"; +import { notifyChatActionWork } from "./chat-work-notifications.js"; + type DbOrTransaction = Db | Parameters[0]>[0]; export interface SlackSessionSyncPayload { @@ -47,6 +49,7 @@ export async function stageSlackSessionSync( runtimeGeneration: input.runtimeGeneration, credentialFingerprint: input.credentialFingerprint, }; + await notifyChatActionWork(database, "slack_session_sync"); await database .insert(chatActions) .values({ diff --git a/server/src/services/chat-work-notifications.ts b/server/src/services/chat-work-notifications.ts index 50da6074d55..f76ac13c72e 100644 --- a/server/src/services/chat-work-notifications.ts +++ b/server/src/services/chat-work-notifications.ts @@ -11,6 +11,9 @@ export async function notifyChatPublicationWork(tx: object): Promise { } export const CHAT_ACTION_QUEUES = { + provider_effect: DELIVERY_QUEUES.chatProviderEffects, + github_webhook_ingress: DELIVERY_QUEUES.chatGitHubIngress, + slack_session_sync: DELIVERY_QUEUES.chatSessionSyncs, slack_board_message: DELIVERY_QUEUES.chatBoardMessages, slash_task_start: DELIVERY_QUEUES.chatTaskStarts, receipt_reaction: DELIVERY_QUEUES.chatReceiptReactions, diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 686d860f02c..fdee30932f8 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -6,6 +6,9 @@ export const DELIVERY_QUEUES = { chatTaskStarts: "chat-task-starts", chatReceiptReactions: "chat-receipt-reactions", chatSessionStops: "chat-session-stops", + chatProviderEffects: "chat-provider-effects", + chatGitHubIngress: "chat-github-ingress", + chatSessionSyncs: "chat-session-syncs", chatVerification: "chat-verification", chatInbound: "chat-inbound", chatPublications: "chat-publications",