feat(09-01): add scheduleOutboxDrain, drainRequested, initOutboxTrigger; route setInterval through wrapper
- 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
This commit is contained in:
@@ -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<void> {
|
||||
// ── 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);
|
||||
}
|
||||
|
||||
@@ -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();
|
||||
|
||||
Reference in New Issue
Block a user