From ce024457f3d747657501fbcfe8eac71f6772150f Mon Sep 17 00:00:00 2001 From: Dotta <34892728+cryppadotta@users.noreply.github.com> Date: Fri, 9 Oct 2026 17:19:41 -0500 Subject: [PATCH] perf(server): wake email work from commits and deadlines (#15750) ## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work. > - Email connects agent tasks to durable incoming and outgoing messages. > - The email worker scans the database every second, even with no inboxes. > - These queries prevent idle database connections from closing. > - This pull request wakes email work from committed writes and known deadlines. > - Empty email installations stop polling. Existing delivery and catch-up guarantees remain. ## Linked Issues or Issue Description Refs #15625. Related standby work: #15314. **What existing behavior does this improve?** The AgentMail background worker and its database traffic when idle. **Current behavior** Every instance queries active email endpoints each second. Active inboxes also scan delivery queues each second even when no work is due. **Proposed behavior** Scan once at startup. Wake after endpoint and queue transactions settle. Restore the next retry or catch-up deadline from existing rows. An instance with no active inboxes has no recurring email queries or timer. **Reason and benefit** Thousands of empty instances should not keep their databases awake to discover that email is still unconfigured. ## What Changed - Use the shared transaction-aware coordinator for email work, including changes made by separate service objects on the same database owner. - Signal endpoint activation and control changes, incoming events, and outgoing sends inside their transactions. - Add an optional deadline to the coordinator. Email uses persisted retry times, mailbox catch-up times, and socket reconnect times. Only the first pending send in each conversation determines its send deadline. - Preserve startup recovery, lease fences, idempotency, standby suppression, idle drain, and shutdown. - Update the sleep inventory and add quiet-idle, commit, activation, retry, and lifecycle tests. ## Verification - Focused coordinator, email idle, and warm standby tests: 20 passed locally. - Database-backed email integration suite: 41 passed locally with real PostgreSQL and a simulated AgentMail provider. - After review, the updated delayed-reply regression passed separately: 1 selected test passed, 40 intentionally filtered out. It verifies no scans before the first reply deadline and delivery of both replies in order. - `pnpm -r typecheck` and `pnpm build` passed locally. - The local `pnpm test:run` was interrupted after its checkout was removed externally. Completed local checks above are preserved. The [final-commit hosted CI matrix](https://github.com/paperclipai/paperclip/actions/runs/37997072212) passed: 54 successful checks, 2 expected Storybook skips, no failures. It includes the full general and serialized test matrix, runner checks, browser tests, build, typecheck, and canary packaging. - Greptile: 5/5 on `ec98f3b86ee476d75e0edf3366b357377ea09d69`, with no actionable findings. ## Risks - New email producers must register work inside their write transaction. No new tables or migrations are required. - Signals are local to one database owner and its dedicated pools. Independent processes and direct SQL do not emit them. Startup recovers existing work; configured inboxes also retain their recovery deadline. - Active inboxes still catch up each minute. WebSocket mode still renews its lease every 20 seconds. This change does not provide external waking or make active sockets sleep. - Provider calls were simulated in tests. No live email was sent. ## Model Used OpenAI Codex, GPT-6, with reasoning, repository inspection, code execution, and test tools. The exact deployment model ID and context-window size are not exposed in this session. ## 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 | 18 ++- .../src/__tests__/cloud-warm-standby.test.ts | 3 +- .../delivery-work-coordinator.test.ts | 17 ++- .../email-channels.integration.test.ts | 72 ++++++++++- server/src/__tests__/email-idle-work.test.ts | 81 ++++++++++++ server/src/app.ts | 1 + .../src/services/delivery-work-coordinator.ts | 4 + .../services/delivery-work-notifications.ts | 1 + server/src/services/email-channels.ts | 118 +++++++++++++----- 9 files changed, 278 insertions(+), 37 deletions(-) create mode 100644 server/src/__tests__/email-idle-work.test.ts 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 cc7df120edd..0646390908a 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 @@ -423,11 +423,19 @@ Longer polling intervals alone do not complete an item. [app.ts](../../server/src/app.ts), [chat-channels.ts](../../server/src/services/chat-channels.ts). -- [ ] **JOB-02 — Email reconciliation.** Current: every **1s**, querying active - endpoints even when none exist. Change: initialize resource/pending state and - update on endpoint changes, webhook receipt, sends, and retries. **Complete - when:** empty mail state produces no polling SQL and new work activates the - worker without restart. Source: +- [x] **JOB-02 — Email reconciliation.** Replaced the unconditional **1s** + scan with startup recovery and transaction-aware wakes for endpoint activation, + pause/resume/removal, admitted events, and queued sends. Separate service + objects sharing the database owner wake the app worker after commit. The + worker restores its next deadline from persisted inbound/outbound retries and + catch-up state. No active inboxes means no recurring email SQL or timer. + Configured inboxes retain one-minute catch-up and live socket lease renewal + (NET-01/NET-02); this does not yet make an active inbox sleep indefinitely. + Intent reconciliation survives ambiguous transaction outcomes. Warm standby + suppresses SQL, and idle drain defers dispatch without losing its wake. + No new tables or provider infrastructure. Independent database roots and + direct SQL do not emit these process-local hints; startup and configured-inbox + catch-up remain the recovery boundaries. Source: [email-channels.ts](../../server/src/services/email-channels.ts). - [ ] **JOB-03 — Browser-use cleanup.** Current: every **3s** with due-session diff --git a/server/src/__tests__/cloud-warm-standby.test.ts b/server/src/__tests__/cloud-warm-standby.test.ts index 76247a35f3c..60cdc331114 100644 --- a/server/src/__tests__/cloud-warm-standby.test.ts +++ b/server/src/__tests__/cloud-warm-standby.test.ts @@ -3,6 +3,7 @@ import express from "express"; import request from "supertest"; import { afterEach, describe, expect, it, vi } from "vitest"; import type { Db } from "@paperclipai/db"; +import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; import { cloudWarmStandbyMiddleware, cloudWarmStandbyServerOptions } from "../middleware/cloud-warm-standby.js"; import { healthRoutes } from "../routes/health.js"; import { emailChannelService } from "../services/email-channels.js"; @@ -124,7 +125,7 @@ describe("unclaimed Cloud background work", () => { vi.useFakeTimers(); let enabled = false; const select = vi.fn(() => { throw new Error("SQL probe"); }); - const db = { select } as unknown as Db; + const db = installDatabaseWorkSignals({ select, execute: vi.fn(), transaction: vi.fn() }) as unknown as Db; const email = emailChannelService(db, { heartbeat: { wakeup: vi.fn() }, isBackgroundWorkEnabled: () => enabled }); const scheduler = createPluginJobScheduler({ db, diff --git a/server/src/__tests__/delivery-work-coordinator.test.ts b/server/src/__tests__/delivery-work-coordinator.test.ts index 1be5395f9df..c964714698c 100644 --- a/server/src/__tests__/delivery-work-coordinator.test.ts +++ b/server/src/__tests__/delivery-work-coordinator.test.ts @@ -27,7 +27,7 @@ function setup() { function task() { return { retryMs: 5000, run: vi.fn(async () => {}), hasPending: vi.fn(async () => false) }; } describe("delivery work coordinator", () => { - it("leaves five empty queues with no deadlines, timers, or periodic queries", async () => { + it("leaves empty queues with no deadlines, timers, or periodic queries", async () => { const s = setup(); const tasks = Object.values(DELIVERY_QUEUES).map(queue => { const t = task(); return { queue, t, worker: s.coordinator.register(queue, t) }; @@ -45,6 +45,21 @@ describe("delivery work coordinator", () => { expect(vi.getTimerCount()).toBe(0); expect(idleWorkSnapshot().active).toBe(0); }); + it("schedules maintenance without holding idle intent, and lets new work preempt it", async () => { + const s = setup(), t = task(); + let next: number | null = Date.now() + 60_000; + await s.coordinator.register(DELIVERY_QUEUES.email, { ...t, nextRunAt: () => next }).ready; + await vi.advanceTimersByTimeAsync(59_000); + expect(t.run).toHaveBeenCalledTimes(1); + await s.enqueue(DELIVERY_QUEUES.email); + await vi.advanceTimersByTimeAsync(1); + expect(t.run).toHaveBeenCalledTimes(2); + expect(idleWorkSnapshot().active).toBe(0); + next = null; + await vi.advanceTimersByTimeAsync(1000); + expect(t.run).toHaveBeenCalledTimes(3); + expect(s.coordinator.nextWakeAt()).toBeNull(); + }); it("does not postpone an already committed wake when another producer starts writing", async () => { const s = setup(), t = task(); await s.coordinator.register(DELIVERY_QUEUES.feedback, t).ready; diff --git a/server/src/__tests__/email-channels.integration.test.ts b/server/src/__tests__/email-channels.integration.test.ts index adcdda01844..c15499e947b 100644 --- a/server/src/__tests__/email-channels.integration.test.ts +++ b/server/src/__tests__/email-channels.integration.test.ts @@ -199,7 +199,9 @@ describe("AgentMail durable email pipeline", () => { expect(await db.select().from(authUsers).where(eq(authUsers.id, "email-board"))).toHaveLength(1); }); - async function fixture(mode: "websocket" | "webhook" = "webhook", storage?: StorageService) { + async function fixture(mode: "websocket" | "webhook" = "webhook", storage?: StorageService, + beforeSetup?: (input: { fetcher: typeof fetch; wakeup: ReturnType }) => Promise, + ) { const companyId = randomUUID(), agentId = randomUUID(), endpointId = randomUUID(); @@ -312,6 +314,7 @@ describe("AgentMail durable email pipeline", () => { storage, }); services.push(service); + await beforeSetup?.({ fetcher, wakeup }); await service.setup( companyId, { @@ -374,6 +377,73 @@ describe("AgentMail durable email pipeline", () => { setWebhookError: (status: number) => { webhookError = status; }, }; } + it("activates an empty running worker from another service and drains committed inbound and outbound work", async () => { + // Connector setup and tool calls construct short-lived service objects. + // Their committed writes must wake the app-owned worker sharing this DB. + let running!: EmailChannelService; + const f = await fixture("webhook", undefined, async ({ fetcher, wakeup }) => { + running = emailChannelService(db, { heartbeat: { wakeup }, fetch: fetcher }); + services.push(running); + await running.start(); + const select = vi.spyOn(db, "select"); + await new Promise(resolve => setTimeout(resolve, 1200)); + expect(select).not.toHaveBeenCalled(); + select.mockRestore(); + }); + await vi.waitFor(async () => { + const [config] = await db.select().from(emailEndpoints).where(eq(emailEndpoints.endpointId, f.endpointId)); + expect(config.lastSyncAt).not.toBeNull(); + }, { timeout: 5000 }); + await running.tick(); + await f.service.control(f.endpointId, "pause", { userId: "email-board" }); + await f.service.control(f.endpointId, "resume", { userId: "email-board" }); + const message = f.message(); + f.messages.set(message.message_id, message); + await f.service.admit(await f.service.getEndpoint(f.endpointId), { + event_type: "message.received", message, + }); + await vi.waitFor(() => expect(f.wakeup).toHaveBeenCalledTimes(1), { timeout: 5000 }); + const [parent] = await db.select().from(issues).where(eq(issues.companyId, f.companyId)); + const input = emailSendSchema.parse({ endpointId: f.endpointId, parentIssueId: parent.id, + to: ["recipient@example.test"], subject: "Wake on send", text: "Hello", idempotencyKey: randomUUID() }); + await f.service.queueSend(f.companyId, input, { userId: "email-board" }); + await vi.waitFor(async () => expect((await f.service.publication(input.idempotencyKey, f.companyId)).outcome).toBe("sent"), { timeout: 5000 }); + expect(f.sends).toHaveLength(1); + await running.tick(); + await f.service.control(f.endpointId, "pause", { userId: "email-board" }); + await new Promise(resolve => setTimeout(resolve, 100)); + const select = vi.spyOn(db, "select"); + await new Promise(resolve => setTimeout(resolve, 1200)); + expect(select).not.toHaveBeenCalled(); + }); + + it("restores a retry deadline without polling replies blocked behind that send", async () => { + const f = await fixture(); + const message = f.message("received", "outbound-thread"); + await f.receive(message); + const [conversation] = await db.select().from(chatConversations).where(eq(chatConversations.endpointId, f.endpointId)); + const input = emailSendSchema.parse({ endpointId: f.endpointId, conversationId: conversation.id, + replyToMessageId: message.message_id, text: "First reply", idempotencyKey: randomUUID() }); + await f.service.queueSend(f.companyId, input, { userId: "email-board" }); + const second = { ...input, text: "Second reply", idempotencyKey: randomUUID() }; + await f.service.queueSend(f.companyId, second, { userId: "email-board" }); + // Simulate a previous process persisting a provider retry deadline. The + // second reply has no deadline but cannot overtake this first reply. + await db.update(chatPublications).set({ state: "retry", nextAttemptAt: new Date(Date.now() + 5000) }) + .where(eq(chatPublications.id, input.idempotencyKey)); + await f.service.start(); + expect(f.sends).toHaveLength(0); + const select = vi.spyOn(db, "select"); + await new Promise(resolve => setTimeout(resolve, 1100)); + expect(select).not.toHaveBeenCalled(); + expect(f.sends).toHaveLength(0); + await vi.waitFor(() => expect(f.sends).toHaveLength(2), { timeout: 8000 }); + await f.service.tick(); + expect(f.sends.map(send => send.key)).toEqual([input.idempotencyKey, second.idempotencyKey]); + for (const id of [input.idempotencyKey, second.idempotencyKey]) + expect((await f.service.publication(id, f.companyId)).outcome).toBe("sent"); + }); + it("admits signed webhooks through the durable queue and rejects a valid signature for another inbox", async () => { const f = await fixture(); const endpoint = await f.service.getEndpoint(f.endpointId); diff --git a/server/src/__tests__/email-idle-work.test.ts b/server/src/__tests__/email-idle-work.test.ts new file mode 100644 index 00000000000..cd871b8d487 --- /dev/null +++ b/server/src/__tests__/email-idle-work.test.ts @@ -0,0 +1,81 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { Db } from "@paperclipai/db"; +import { installDatabaseWorkSignals } from "../../../packages/db/src/work-signals.js"; +import { emailChannelService } from "../services/email-channels.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "../services/delivery-work-notifications.js"; +import { idleWorkSnapshot } from "../services/task-admission.js"; + +const services: ReturnType[] = []; +afterEach(async () => { + for (const service of services.splice(0)) await service.shutdown(); + vi.useRealTimers(); +}); +function fixture() { + vi.useFakeTimers(); + const select = vi.fn(() => ({ from: () => ({ where: async () => [] }) })); + const execute = vi.fn(async () => [{ xid: "42", status: "aborted" }]); + const db = installDatabaseWorkSignals({ select, execute, + async transaction(callback: (tx: any) => Promise) { + return callback({ execute, transaction: db.transaction }); + }, + }); + let enabled = true; + const service = emailChannelService(db as unknown as Db, { + heartbeat: { wakeup: vi.fn() }, + isBackgroundWorkEnabled: () => enabled, + isReconciliationEnabled: () => true, + }); + services.push(service); + return { db, select, service, enable: (value: boolean) => { enabled = value; } }; +} + +describe("email idle work", () => { + it("scans at startup, then leaves an unconfigured stack with zero timers and queries", async () => { + const f = fixture(); + await f.service.start(); + expect(f.select).toHaveBeenCalledTimes(1); + await vi.advanceTimersByTimeAsync(3_600_000); + expect(f.select).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + }); + it("wakes on an outer transaction commit from a separate producer, never an inner savepoint", async () => { + const f = fixture(); + await f.service.start(); + await f.db.transaction(async tx => { + await tx.transaction(async (inner: object) => notifyDeliveryWork(inner, DELIVERY_QUEUES.email)); + expect(f.select).toHaveBeenCalledTimes(1); + }); + await vi.advanceTimersByTimeAsync(1); + expect(f.select).toHaveBeenCalledTimes(2); + expect(vi.getTimerCount()).toBe(0); + expect(idleWorkSnapshot().active).toBe(0); + }); + it("retains startup and committed work across idle drain without SQL or dispatch", async () => { + const f = fixture(); + f.enable(false); + await f.service.start(); + await f.db.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.email)); + await vi.advanceTimersByTimeAsync(60_000); + expect(f.select).not.toHaveBeenCalled(); + expect(idleWorkSnapshot().active).toBe(0); + f.enable(true); + await vi.advanceTimersByTimeAsync(1001); + expect(f.select).toHaveBeenCalledTimes(1); + expect(vi.getTimerCount()).toBe(0); + }); + it("retries a failed startup scan and unsubscribes on shutdown", async () => { + const f = fixture(); + const warning = vi.spyOn(process, "emitWarning").mockImplementation(() => {}); + f.select.mockImplementationOnce(() => { throw new Error("database unavailable"); }); + await f.service.start(); + await vi.advanceTimersByTimeAsync(1001); + expect(f.select).toHaveBeenCalledTimes(2); + expect(vi.getTimerCount()).toBe(0); + await f.service.shutdown(); + await f.db.transaction(tx => notifyDeliveryWork(tx, DELIVERY_QUEUES.email)); + await vi.advanceTimersByTimeAsync(60_000); + expect(f.select).toHaveBeenCalledTimes(2); + expect(vi.getTimerCount()).toBe(0); + warning.mockRestore(); + }); +}); diff --git a/server/src/app.ts b/server/src/app.ts index f588ff635a0..ed52a4c112a 100644 --- a/server/src/app.ts +++ b/server/src/app.ts @@ -646,6 +646,7 @@ export async function createApp( // mutation guard. The Chat SDK adapter verifies the provider signature // before Paperclip persists or acts on any event. const emailChannels = emailChannelService(db, { + isReconciliationEnabled: () => !isWarmStandby(), isBackgroundWorkEnabled: () => !isWarmStandby() && !isIdleTaskDrainActive(), heartbeat: connectionIntentHeartbeat, storage: opts.storageService, diff --git a/server/src/services/delivery-work-coordinator.ts b/server/src/services/delivery-work-coordinator.ts index 34a198671ba..0272fc2ed72 100644 --- a/server/src/services/delivery-work-coordinator.ts +++ b/server/src/services/delivery-work-coordinator.ts @@ -22,6 +22,8 @@ export function createDeliveryWorkCoordinator(input: { retryMs: number; run: (signal: AbortSignal) => Promise; hasPending: () => Promise; + // Optional maintenance/retry deadline restored from durable state. + nextRunAt?: () => number | null; }) { if (stopped) throw new Error("Delivery coordinator is stopped"); if (workers.has(queue)) throw new Error(`Delivery worker already registered: ${queue}`); @@ -51,6 +53,8 @@ export function createDeliveryWorkCoordinator(input: { .then(() => workerStopped || !input.canRun() ? undefined : task.run(attempt.signal)).then(() => workerStopped ? false : task.hasPending()).then(pending => { if (pending || databaseWorkPending(input.owner, queue)) schedule(task.retryMs); else { finishIntent?.(); finishIntent = undefined; } + const next = task.nextRunAt?.(); + if (next != null) schedule(Math.max(0, next - Date.now())); }).catch(error => { schedule(task.retryMs); // Logging must never turn a recoverable sweep failure into an diff --git a/server/src/services/delivery-work-notifications.ts b/server/src/services/delivery-work-notifications.ts index 14b67085d1d..d9696da5cf2 100644 --- a/server/src/services/delivery-work-notifications.ts +++ b/server/src/services/delivery-work-notifications.ts @@ -2,6 +2,7 @@ import { signalDatabaseWork, subscribeDatabaseWork } from "@paperclipai/db"; /** Topics for existing durable queues; register intent before their writes. */ export const DELIVERY_QUEUES = { + email: "email-channels", feedback: "feedback-exports", chatCompletion: "chat-completions", connection: "connection-continuations", diff --git a/server/src/services/email-channels.ts b/server/src/services/email-channels.ts index 62db025fd6c..830d5f921f9 100644 --- a/server/src/services/email-channels.ts +++ b/server/src/services/email-channels.ts @@ -1,3 +1,5 @@ +import { createDeliveryWorkCoordinator } from "./delivery-work-coordinator.js"; +import { DELIVERY_QUEUES, notifyDeliveryWork } from "./delivery-work-notifications.js"; import { HttpError } from "../errors.js"; import { createHash, randomUUID } from "node:crypto"; import WebSocket from "ws"; @@ -77,6 +79,8 @@ type Tx = Parameters[0]>[0]; export interface EmailChannelOptions { /** Suppress periodic database work while an unclaimed Cloud app stands by. */ isBackgroundWorkEnabled?: () => boolean; + /** Allow transaction outcome checks during idle drain, but not warm standby. */ + isReconciliationEnabled?: () => boolean; heartbeat: Pick, "wakeup">; storage?: StorageService; publicBaseUrl?: string; @@ -130,7 +134,12 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { let stopped = false; let ticking = false; let activeTick: Promise | null = null; - let timer: ReturnType | undefined; + let coordinator: ReturnType | undefined; + let worker: ReturnType["register"]> | undefined; + let nextWorkAt: number | null = null; + function needWorkAt(at: number) { + nextWorkAt = Math.min(nextWorkAt ?? Infinity, Math.max(Date.now() + 1000, at)); + } async function getEndpoint(id: string) { const [row] = await db @@ -881,6 +890,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { } const now = new Date(); await db.transaction(async (tx) => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.email); await tx .update(emailEndpoints) .set({ @@ -909,7 +919,6 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { return summary(await getEndpoint(endpoint.id)); }); if (!result) throw conflict("Inbox setup is already running"); - if (timer) void tick().catch(() => {}); return result; } @@ -919,22 +928,24 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { if (!event) return; if (event.inbox_id !== endpoint.botExternalId) throw forbidden("Email event belongs to a different inbox"); - await db - .insert(chatDeliveries) - .values({ - companyId: endpoint.companyId, - endpointId: endpoint.id, - providerEventId: event.eventId, - deduplicationKey: `${event.kind}:${event.message_id}`, - eventKind: "message", - normalizedEvent: event, - }) - .onConflictDoNothing(); - await db - .update(chatEndpoints) - .set({ lastEventAt: new Date() }) - .where(eq(chatEndpoints.id, endpoint.id)); - if (timer) void tick().catch(() => {}); + await db.transaction(async (tx) => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.email); + await tx + .insert(chatDeliveries) + .values({ + companyId: endpoint.companyId, + endpointId: endpoint.id, + providerEventId: event.eventId, + deduplicationKey: `${event.kind}:${event.message_id}`, + eventKind: "message", + normalizedEvent: event, + }) + .onConflictDoNothing(); + await tx + .update(chatEndpoints) + .set({ lastEventAt: new Date() }) + .where(eq(chatEndpoints.id, endpoint.id)); + }); } async function webhook( publicId: string, @@ -1550,6 +1561,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { if (!reply) throw badRequest("Reply target is not in this inbox conversation"); } + await notifyDeliveryWork(tx, DELIVERY_QUEUES.email); await tx.insert(chatPublications).values({ id: input.idempotencyKey, companyId, @@ -1571,7 +1583,6 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { await audit(endpoint, "email.queued", actor, { publicationId: input.idempotencyKey, }); - if (timer) void tick().catch(() => {}); return publication(input.idempotencyKey, companyId); } async function publication( @@ -1968,6 +1979,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { .update(emailEndpoints) .set({ lastSyncAt: null }) .where(eq(emailEndpoints.endpointId, endpoint.id)); + worker?.wake(); } else await admit(endpoint, value); })().catch(async () => { await markError( @@ -1988,6 +2000,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { const delay = Math.min(60_000, (backoff.get(endpoint.id) ?? 1000) * 2); backoff.set(endpoint.id, delay); reconnectAt.set(endpoint.id, Date.now() + delay); + worker?.wake(); void db .delete(chatEndpointLeases) .where( @@ -2056,6 +2069,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { async function runTick() { if (ticking || stopped) return; ticking = true; + nextWorkAt = null; try { const endpoints = await db .select() @@ -2096,7 +2110,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { ne(chatPublications.state, "delivery_unknown"), ), ) - .orderBy(asc(chatPublications.createdAt)) + .orderBy(asc(chatPublications.createdAt), asc(chatPublications.id)) .limit(25); // Each conversation is serial, while independent inbox threads can make progress. const firstSends = [ @@ -2199,7 +2213,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { ); if ( !config.lastSyncAt || - Date.now() - config.lastSyncAt.getTime() > 60_000 + Date.now() - config.lastSyncAt.getTime() >= 60_000 ) await withLease( endpoint, @@ -2208,7 +2222,40 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { }, "email-catchup", ); + // Catch-up is a real scheduled obligation; an empty queue is not. + const refreshed = await getConfig(endpoint.id); + needWorkAt((refreshed.lastSyncAt?.getTime() ?? 0) + 60_000); + if (refreshed.receiveMode === "websocket" && !sockets.has(endpoint.id)) + needWorkAt(reconnectAt.get(endpoint.id) ?? Date.now() + 1000); + const [inbound] = await db + .select({ at: sql`min(coalesce(${chatDeliveries.nextAttemptAt}, now()))` }) + .from(chatDeliveries) + .where(and( + eq(chatDeliveries.endpointId, endpoint.id), + inArray(chatDeliveries.state, ["received", "processing", "retry"]), + )); + // A later reply cannot run ahead of its conversation's first send. + // Its null deadline must not turn a delayed retry into 1s polling. + const sendHeads = db + .selectDistinctOn([chatPublications.conversationId], { + nextAttemptAt: chatPublications.nextAttemptAt, + }) + .from(emailSends) + .innerJoin(chatPublications, eq(chatPublications.id, emailSends.publicationId)) + .where(and( + eq(emailSends.endpointId, endpoint.id), + inArray(emailSends.outcome, ["queued", "uncertain"]), + ne(chatPublications.state, "delivery_unknown"), + )) + .orderBy(asc(chatPublications.conversationId), asc(chatPublications.createdAt), asc(chatPublications.id)) + .as("email_send_heads"); + const [outbound] = await db + .select({ at: sql`min(coalesce(${sendHeads.nextAttemptAt}, now()))` }) + .from(sendHeads); + for (const pending of [inbound, outbound]) + if (pending?.at) needWorkAt(new Date(pending.at).getTime()); } catch (e) { + needWorkAt(Date.now() + 1000); await markError(endpoint, diagnostic(e)); } }), @@ -2219,6 +2266,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { } async function stopEndpoint(endpoint: Endpoint) { await db.transaction(async (tx) => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.email); await tx .select() .from(chatEndpoints) @@ -2319,6 +2367,7 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { await removeUnusedSecret(secretId); } await db.transaction(async (tx) => { + await notifyDeliveryWork(tx, DELIVERY_QUEUES.email); await tx .update(chatEndpoints) .set({ @@ -2642,17 +2691,28 @@ export function emailChannelService(db: Db, options: EmailChannelOptions) { }; }, start: () => { - if (!timer) { - timer = setInterval(() => { - void tick().catch(() => {}); - }, 1000); - timer.unref(); - void tick().catch(() => {}); - } + if (coordinator || stopped) return; + coordinator = createDeliveryWorkCoordinator({ + owner: db, + canRun: () => options.isBackgroundWorkEnabled?.() !== false, + canReconcile: () => options.isReconciliationEnabled?.() ?? (options.isBackgroundWorkEnabled?.() !== false), + onError: (error) => process.emitWarning(`Email background work failed: ${diagnostic(error)}`), + }); + worker = coordinator.register(DELIVERY_QUEUES.email, { + retryMs: 1000, + run: tick, + // Retries and maintenance have deadlines; only unresolved writes need + // the coordinator's pending-intent hold, not a configured idle inbox. + hasPending: async () => false, + nextRunAt: () => options.isBackgroundWorkEnabled?.() === false + ? Date.now() + 1000 + : nextWorkAt, + }); + return worker.ready; }, shutdown: async () => { stopped = true; - clearInterval(timer); + await coordinator?.stop(); await activeTick; const heldSockets = [...sockets.entries()]; sockets.clear();