From 2b113045f73c043ad616380f82e6c47e7a5bc66f Mon Sep 17 00:00:00 2001 From: Lucas Berger Date: Fri, 12 Jun 2026 16:52:36 -0400 Subject: [PATCH] feat(09-01): add scheduleOutboxDrain, drainRequested, initOutboxTrigger; route setInterval through wrapper MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Add import { onOutboxDrain } from outboxTrigger.js - Add let drainRequested = false (D-05 trailing-re-drain flag) - Export scheduleOutboxDrain(): void — isDraining guard + drainRequested loop (D-05/T-09-01) drainRequested=false reset precedes recursive call (Pitfall 3) errors caught via .catch to prevent crash (D-02/T-09-03) - Export initOutboxTrigger(): void — registers onOutboxDrain(() => scheduleOutboxDrain()) - startOutboxWorker setInterval body: scheduleOutboxDrain() replaces runOutboxDrain().catch() 15 * 1000 interval unchanged (D-08) - runOutboxDrain body/isDraining guard/finally unchanged (D-02/D-07) - Fix trigger-wiring tests: add beforeAll(initOutboxTrigger) to wire EventEmitter listener; fix Test C mock to return empty rows on trailing drain (correct D-07 behaviour) - 30/30 outboxWorker tests GREEN; tsc --noEmit clean --- apps/api/src/broker/outboxWorker.ts | 68 ++++++++++++++++++++-- apps/api/tests/broker/outboxWorker.test.ts | 23 ++++++-- 2 files changed, 82 insertions(+), 9 deletions(-) diff --git a/apps/api/src/broker/outboxWorker.ts b/apps/api/src/broker/outboxWorker.ts index 962aa56..6b2e7f5 100644 --- a/apps/api/src/broker/outboxWorker.ts +++ b/apps/api/src/broker/outboxWorker.ts @@ -17,7 +17,9 @@ * T-03-14: create-before-delete ordering; create-fail aborts delete. * * runOutboxDrain is exported for unit testing. - * startOutboxWorker wraps it in a 15-second setInterval. + * scheduleOutboxDrain is exported for unit testing; it wraps runOutboxDrain with the isDraining + * guard + drainRequested trailing-re-drain loop (D-05). + * startOutboxWorker wraps scheduleOutboxDrain in a 15-second setInterval. * (node-cron 4.2.1 silently skipped scheduled executions in the long-running server process; * setInterval fires reliably in the same process — replaced to fix the silent skip.) * @@ -34,6 +36,7 @@ import { createCalendarEvent, updateCalendarEvent, deleteCalendarEvent } from '. import { buildVeventString, extractRruleString, RRULE_PRESETS } from './vevent.js'; import type { FastmailClient } from './client.js'; import { dispatchEventChange } from '../lib/eventChangeDispatcher.js'; +import { onOutboxDrain } from '../lib/outboxTrigger.js'; // ── Constants (D-07) ──────────────────────────────────────────────────────── @@ -158,6 +161,47 @@ export function assembleRruleString( */ let isDraining = false; +/** + * D-05 / D-06: trailing-re-drain flag. + * Set to true when signalOutboxDrain() fires while a drain is already in flight. + * Collapses any number of mid-drain signals into exactly one trailing drain — never + * rate-limited, never dropped, never more than one extra pass (D-06). + * Reset to false BEFORE the recursive scheduleOutboxDrain() call (Pitfall 3 — resetting + * after would allow an unbounded re-drain chain against Fastmail rate limits / T-09-01). + */ +let drainRequested = false; + +/** + * Schedule an outbox drain pass. + * + * If no drain is currently running, kicks off runOutboxDrain() immediately. + * If a drain IS running (isDraining=true), records drainRequested=true so the + * currently-running drain triggers exactly one trailing re-drain on completion (D-05). + * + * Errors from runOutboxDrain are caught and logged (D-02 / T-09-03). + * + * Called by: + * - initOutboxTrigger's onOutboxDrain listener (event-driven path, CAL-15) + * - startOutboxWorker's 15-second setInterval (polling fallback, D-08) + */ +export function scheduleOutboxDrain(): void { + if (isDraining) { + drainRequested = true; + return; + } + runOutboxDrain() + .catch((err: unknown) => { + console.error('[outboxWorker] Unhandled runOutboxDrain error:', err); + }) + .finally(() => { + if (drainRequested) { + // Reset BEFORE recursive call (Pitfall 3) — prevents unbounded re-drain chain + drainRequested = false; + scheduleOutboxDrain(); + } + }); +} + /** * WR-06: max time to wait on the post-write targeted re-sync before marking the * outbox row 'done'. A stalled Fastmail connection cannot wedge the single-process @@ -788,15 +832,29 @@ export async function runOutboxDrain(): Promise { // ── Scheduler ──────────────────────────────────────────────────────────────── /** - * Starts the 15-second background outbox drain schedule. + * Register the in-process EventEmitter drain signal listener (CAL-15 / D-01). + * Call once at API startup (in index.ts, after startOutboxWorker). + * After registration, any signalOutboxDrain() call (fired post-enqueue) immediately + * routes through scheduleOutboxDrain — eliminating the up-to-15s polling delay. + * + * T-09-02: initOutboxTrigger is called only under isMainModule() in index.ts; + * tests import runOutboxDrain/scheduleOutboxDrain directly and never register + * a listener — no open-handle leak preventing process exit. + */ +export function initOutboxTrigger(): void { + onOutboxDrain(() => scheduleOutboxDrain()); +} + +/** + * Starts the 15-second background outbox drain schedule (polling fallback, D-08). * Call once at API startup (wired in index.ts beside startBrokerPoller). * Uses setInterval instead of node-cron: node-cron 4.2.1 silently skipped executions * in the long-running server process; setInterval fires reliably. + * The interval body calls scheduleOutboxDrain() so errors are absorbed by its .catch + * and mid-drain signals collapse correctly via drainRequested (D-05). */ export function startOutboxWorker(): void { setInterval(() => { - runOutboxDrain().catch((err: unknown) => { - console.error('[outboxWorker] Unhandled runOutboxDrain error:', err); - }); + scheduleOutboxDrain(); }, 15 * 1000); } diff --git a/apps/api/tests/broker/outboxWorker.test.ts b/apps/api/tests/broker/outboxWorker.test.ts index b4fab7d..75dbef5 100644 --- a/apps/api/tests/broker/outboxWorker.test.ts +++ b/apps/api/tests/broker/outboxWorker.test.ts @@ -13,12 +13,12 @@ * They will turn GREEN in Plan 03-03 when the implementation is added. */ -import { describe, it, expect, vi, beforeEach } from 'vitest'; +import { describe, it, expect, vi, beforeEach, beforeAll } from 'vitest'; // This import fails (RED) — broker/outboxWorker.ts does not exist yet. // eslint-disable-next-line @typescript-eslint/ban-ts-comment // @ts-ignore intentional RED import -import { runOutboxDrain, assembleRruleString, scheduleOutboxDrain } from '../../src/broker/outboxWorker.js'; +import { runOutboxDrain, assembleRruleString, scheduleOutboxDrain, initOutboxTrigger } from '../../src/broker/outboxWorker.js'; import { signalOutboxDrain } from '../../src/lib/outboxTrigger.js'; // ── Drizzle DB mock ──────────────────────────────────────────────────────── @@ -745,6 +745,13 @@ describe('FREQ persistence (D-07 regression)', () => { // RED: these tests fail because scheduleOutboxDrain is not yet exported. describe('scheduleOutboxDrain — trigger wiring (D-09)', () => { + // Wire the EventEmitter signal → scheduleOutboxDrain once for this describe block. + // initOutboxTrigger registers the 'drain' listener that connects signalOutboxDrain() + // to scheduleOutboxDrain(). Called once (not per-test) to avoid listener accumulation. + beforeAll(() => { + initOutboxTrigger(); + }); + beforeEach(() => { vi.resetAllMocks(); mockPendingRows = []; @@ -818,6 +825,8 @@ describe('scheduleOutboxDrain — trigger wiring (D-09)', () => { // Test C (SC-4 / D-07): two concurrent scheduleOutboxDrain() calls dispatch each row // exactly once — the second call is a no-op via the isDraining guard. + // The trailing re-drain (triggered by drainRequested) finds 0 pending rows so + // createCalendarEvent is called exactly once total. it('D-07: two concurrent scheduleOutboxDrain() calls invoke createCalendarEvent exactly once', async () => { vi.useFakeTimers(); try { @@ -829,14 +838,20 @@ describe('scheduleOutboxDrain — trigger wiring (D-09)', () => { setTimeout(() => resolve(makeResponse(201)), 20), ), ); - mockPendingRows = [makeRow({ id: 1 })]; + const row = makeRow({ id: 1 }); + // First pending-rows query returns the row; subsequent queries return empty + // (simulates: drain 1 processes and removes the row; trailing drain finds nothing) + mockWherePending + .mockImplementationOnce(() => Promise.resolve([row])) + .mockImplementation(() => Promise.resolve([])); - // Both calls are synchronous — second must no-op via isDraining guard + // Both calls are synchronous — second sees isDraining=true and sets drainRequested=true scheduleOutboxDrain(); scheduleOutboxDrain(); await vi.runAllTimersAsync(); + // Drain 1 dispatched the row once; trailing drain found 0 rows → createCalendarEvent once total expect(createCalendarEvent).toHaveBeenCalledTimes(1); } finally { vi.useRealTimers();