From 6d21cc118e1b22967032ac4c5a4e557196b4ef51 Mon Sep 17 00:00:00 2001 From: eleith Date: Tue, 20 Oct 2026 10:34:17 -0700 Subject: [PATCH 1/2] feat(worker): await native cron maintenance and stop resources in order --- .changeset/native-maintenance.md | 7 + apps/worker/src/calendar/refresh.test.ts | 346 +++++++++++++++- apps/worker/src/calendar/refresh.ts | 34 +- apps/worker/src/calendar/scheduler.test.ts | 25 -- apps/worker/src/calendar/scheduler.ts | 32 -- apps/worker/src/calendar/sync-scanner.test.ts | 66 --- apps/worker/src/calendar/sync-scanner.ts | 54 --- apps/worker/src/index.test.ts | 388 ++++++++++++++++++ apps/worker/src/index.ts | 129 ++---- .../services/lifecycle.integration.test.ts | 157 +++++++ apps/worker/src/services/lifecycle.test.ts | 124 ++++++ apps/worker/src/services/lifecycle.ts | 38 ++ apps/worker/src/services/startup.ts | 94 +++++ .../workflows/expire-pending-appointments.ts | 27 ++ apps/worker/src/workflows/index.ts | 6 + .../workflows/maintenance.integration.test.ts | 376 +++++++++++++++++ apps/worker/src/workflows/maintenance.test.ts | 183 +++++++++ .../worker/src/workflows/refresh-calendars.ts | 22 + apps/worker/src/workflows/sync-calendars.ts | 28 +- packages/jobs/src/index.ts | 6 + packages/jobs/src/specs.test.ts | 53 ++- packages/jobs/src/specs.ts | 25 +- 22 files changed, 1906 insertions(+), 314 deletions(-) create mode 100644 .changeset/native-maintenance.md delete mode 100644 apps/worker/src/calendar/scheduler.test.ts delete mode 100644 apps/worker/src/calendar/scheduler.ts delete mode 100644 apps/worker/src/calendar/sync-scanner.test.ts delete mode 100644 apps/worker/src/calendar/sync-scanner.ts create mode 100644 apps/worker/src/index.test.ts create mode 100644 apps/worker/src/services/lifecycle.integration.test.ts create mode 100644 apps/worker/src/services/lifecycle.test.ts create mode 100644 apps/worker/src/services/lifecycle.ts create mode 100644 apps/worker/src/services/startup.ts create mode 100644 apps/worker/src/workflows/expire-pending-appointments.ts create mode 100644 apps/worker/src/workflows/maintenance.integration.test.ts create mode 100644 apps/worker/src/workflows/maintenance.test.ts create mode 100644 apps/worker/src/workflows/refresh-calendars.ts diff --git a/.changeset/native-maintenance.md b/.changeset/native-maintenance.md new file mode 100644 index 00000000..6e9852be --- /dev/null +++ b/.changeset/native-maintenance.md @@ -0,0 +1,7 @@ +--- +'@when/web': patch +'@when/worker': patch +'@when/cli': patch +--- + +Run refresh, pending expiry and calendar repair as awaited native UTC cron workflows, with independent startup passes and pause-preserving schedule persistence. Remove application maintenance timers and coalescing; one execution slot serializes passes, so slow maintenance can delay other jobs. Stop the worker before closing queue and application databases. diff --git a/apps/worker/src/calendar/refresh.test.ts b/apps/worker/src/calendar/refresh.test.ts index 8ae2686a..e6cc8289 100644 --- a/apps/worker/src/calendar/refresh.test.ts +++ b/apps/worker/src/calendar/refresh.test.ts @@ -1,10 +1,18 @@ import { expect, test, vi, beforeEach } from 'vitest'; import pino from 'pino'; +import { sql } from 'kysely'; import type { ResolvedCalendar, WhenConfiguration } from '@when/config'; import { openDb, runMigrations, replaceCalendarBusy, recordServiceOutcome } from '@when/db'; import type { Logger } from '../services/logger.js'; import type { WorkerContext } from '../services/context.js'; -import { busyCalendarIds, refreshCalendar, refreshCalendars, refreshWindow } from './refresh.js'; +import { calendarRefreshTotal } from '../services/metrics.js'; +import { + busyCalendarIds, + refreshCalendar, + refreshCalendars, + refreshCycle, + refreshWindow +} from './refresh.js'; const inst = (s: string) => Temporal.Instant.from(s); const window = { start: inst('2026-04-01T00:00:00Z'), end: inst('2026-05-01T00:00:00Z') }; @@ -45,6 +53,26 @@ END:VCALENDAR `; +const workMeetings = { + chat: { + booking_calendar: 'work', + additional_busy_calendars: [], + booking_window_days: 60, + schedule: 'standard' + } +} as unknown as WhenConfiguration['meetings']; + +const appointment = { + id: '1', + event_type_id: 'chat', + start_time: '2026-04-15T14:00:00Z', + end_time: '2026-04-15T15:00:00Z', + guest_name: 'A', + guest_email: 'a@example.com', + status: 'confirmed' as const, + cancel_token: 't' +}; + const defaultTestConfig = { url: { app: 'https://when.example.com' }, user: { name: 'Jane', email: 'jane@example.com' }, @@ -78,7 +106,14 @@ test('refreshCalendar populates the mirror and records success', async () => { vi.spyOn(globalThis, 'fetch').mockResolvedValue( new Response(oneEvent('remote-1@x'), { status: 207 }) ); - await refreshCalendar(ctx, workCal, window, { now: window.start }); + const metric = vi.spyOn(calendarRefreshTotal, 'inc'); + const result = await refreshCalendar(ctx, workCal, window, { now: window.start }); + expect(result).toEqual({ ok: true, busyCount: 1 }); + expect(metric).toHaveBeenCalledExactlyOnceWith({ + calendar_id: 'work', + provider_type: 'caldav', + status: 'success' + }); const busy = await ctx.db.selectFrom('external_calendar_busy').selectAll().execute(); expect(busy).toHaveLength(1); expect(busy[0].start_time).toBe('2026-04-15T14:00:00Z'); @@ -101,10 +136,23 @@ test('refreshCalendar keeps stale data and records the error on provider failure await replaceCalendarBusy(ctx.db, 'work', [ { start: '2026-04-10T09:00:00Z', end: '2026-04-10T09:30:00Z' } ]); + const lastOk = window.start.subtract({ minutes: 30 }).toString(); + await recordServiceOutcome( + ctx.db, + { kind: 'calendar', name: 'work' }, + { at: lastOk, via: 'refresh' } + ); vi.spyOn(globalThis, 'fetch').mockResolvedValue( new Response('boom', { status: 500, statusText: 'err' }) ); - await refreshCalendar(ctx, workCal, window, { now: window.start }); + const metric = vi.spyOn(calendarRefreshTotal, 'inc'); + const result = await refreshCalendar(ctx, workCal, window, { now: window.start, via: 'test' }); + expect(result).toEqual({ ok: false, error: expect.stringContaining('500') }); + expect(metric).toHaveBeenCalledExactlyOnceWith({ + calendar_id: 'work', + provider_type: 'caldav', + status: 'failure' + }); const busy = await ctx.db.selectFrom('external_calendar_busy').selectAll().execute(); expect(busy).toHaveLength(1); expect(busy[0].start_time).toBe('2026-04-10T09:00:00Z'); @@ -115,7 +163,8 @@ test('refreshCalendar keeps stale data and records the error on provider failure .where('name', '=', 'work') .executeTakeFirstOrThrow(); expect(status.error).toContain('500'); - expect(status.last_ok_at).toBeNull(); + expect(status.last_ok_at).toBe(lastOk); + expect(status.via).toBe('test'); } finally { await ctx.db.destroy(); } @@ -170,15 +219,8 @@ test('refreshCalendar drops our own published event', async () => { await ctx.db .insertInto('appointments') .values({ - id: '1', - event_type_id: 'chat', - start_time: '2026-04-15T14:00:00Z', - end_time: '2026-04-15T15:00:00Z', - guest_name: 'A', - guest_email: 'a@example.com', + ...appointment, location: null, - status: 'confirmed', - cancel_token: 't', external_event_id: 'remote-1@x', external_calendar_id: 'work', calendar_revision: 0, @@ -381,10 +423,286 @@ test('refreshCalendars skips a calendar refreshed within its interval, refreshes await refreshCalendars(ctx, { now: window.start.add({ minutes: 10 }) }); expect(fetched).toBe(false); - // past the interval → due, refreshes - await refreshCalendars(ctx, { now: window.start.add({ minutes: 31 }) }); + await refreshCalendars(ctx, { now: window.start.add({ minutes: 30 }) }); expect(fetched).toBe(true); } finally { await db.destroy(); } }); + +test('refreshCalendar lets the own-event DB read fail before fetching or recording an outcome', async () => { + const ctx = await ctxWithDb(); + try { + await ctx.db.schema.dropTable('appointments').execute(); + const fetch = vi.spyOn(globalThis, 'fetch'); + const log = vi.spyOn(ctx.logger, 'error'); + const metric = vi.spyOn(calendarRefreshTotal, 'inc'); + + await expect(refreshCalendar(ctx, workCal, window)).rejects.toThrow('no such table'); + + expect(fetch).not.toHaveBeenCalled(); + expect(log).not.toHaveBeenCalled(); + expect(metric).not.toHaveBeenCalled(); + const statuses = await ctx.db.selectFrom('service_status').selectAll().execute(); + expect(statuses).toEqual([]); + } finally { + await ctx.db.destroy(); + } +}); + +test('refreshCalendar rejects a mirror replacement failure without recording provider failure', async () => { + const ctx = await ctxWithDb(); + try { + const stale = { start: '2026-04-10T09:00:00Z', end: '2026-04-10T09:30:00Z' }; + await replaceCalendarBusy(ctx.db, 'work', [stale]); + await sql`create trigger fail_mirror before insert on external_calendar_busy + begin select raise(abort, 'mirror unavailable'); end`.execute(ctx.db); + vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response(oneEvent('r1'), { status: 207 })); + const log = vi.spyOn(ctx.logger, 'error'); + const metric = vi.spyOn(calendarRefreshTotal, 'inc'); + + await expect(refreshCalendar(ctx, workCal, window)).rejects.toThrow('mirror unavailable'); + + expect(log).not.toHaveBeenCalled(); + expect(metric).not.toHaveBeenCalled(); + const statuses = await ctx.db.selectFrom('service_status').selectAll().execute(); + expect(statuses).toEqual([]); + const busy = await ctx.db + .selectFrom('external_calendar_busy') + .select(['start_time', 'end_time']) + .execute(); + expect(busy).toEqual([{ start_time: stale.start, end_time: stale.end }]); + } finally { + await ctx.db.destroy(); + } +}); + +test.each(['calendar', 'provider'] as const)( + 'refreshCalendar does not reclassify a successful fetch when the %s status write fails', + async (kind) => { + const ctx = await ctxWithDb(); + try { + await sql`create trigger fail_status before insert on service_status + when new.kind = ${sql.lit(kind)} + begin select raise(abort, 'status unavailable'); end`.execute(ctx.db); + vi.spyOn(globalThis, 'fetch').mockResolvedValue( + new Response(oneEvent('r1'), { status: 207 }) + ); + const log = vi.spyOn(ctx.logger, 'error'); + const metric = vi.spyOn(calendarRefreshTotal, 'inc'); + + await expect(refreshCalendar(ctx, workCal, window)).rejects.toThrow('status unavailable'); + + expect(log).not.toHaveBeenCalled(); + expect(metric).not.toHaveBeenCalled(); + const statuses = await ctx.db.selectFrom('service_status').selectAll().execute(); + expect(statuses).toHaveLength(kind === 'provider' ? 1 : 0); + for (const status of statuses) { + expect(status.error).toBeNull(); + expect(status.last_ok_at).toBe(window.start.toString()); + } + const busy = await ctx.db.selectFrom('external_calendar_busy').selectAll().execute(); + expect(busy).toHaveLength(1); + } finally { + await ctx.db.destroy(); + } + } +); + +test('refreshCalendar lets a success metric failure escape without recording provider failure', async () => { + const ctx = await ctxWithDb(); + try { + vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response(oneEvent('r1'), { status: 207 })); + const error = new Error('metric unavailable'); + const metric = vi.spyOn(calendarRefreshTotal, 'inc').mockImplementationOnce(() => { + throw error; + }); + const log = vi.spyOn(ctx.logger, 'error'); + + await expect(refreshCalendar(ctx, workCal, window)).rejects.toBe(error); + + expect(log).not.toHaveBeenCalled(); + expect(metric).toHaveBeenCalledTimes(1); + const statuses = await ctx.db.selectFrom('service_status').selectAll().execute(); + expect(statuses).toHaveLength(2); + for (const status of statuses) expect(status.error).toBeNull(); + } finally { + await ctx.db.destroy(); + } +}); + +test('refreshCalendar rejects if recording an expected provider failure fails locally', async () => { + const ctx = await ctxWithDb(); + try { + await sql`create trigger fail_status before insert on service_status + begin select raise(abort, 'status unavailable'); end`.execute(ctx.db); + vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('boom', { status: 500 })); + const metric = vi.spyOn(calendarRefreshTotal, 'inc'); + + await expect(refreshCalendar(ctx, workCal, window)).rejects.toThrow('status unavailable'); + + expect(metric).not.toHaveBeenCalled(); + } finally { + await ctx.db.destroy(); + } +}); + +test('refreshCycle flags conflicts using the newly refreshed mirror', async () => { + const ctx = await ctxWithDb({ meetings: workMeetings }); + try { + await ctx.db.insertInto('appointments').values(appointment).execute(); + await replaceCalendarBusy(ctx.db, 'work', [ + { start: '2026-04-10T09:00:00Z', end: '2026-04-10T09:30:00Z' } + ]); + vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response(oneEvent('r1'), { status: 207 })); + + await refreshCycle(ctx, { now: window.start }); + + const row = await ctx.db + .selectFrom('appointments') + .select('has_possible_conflict') + .executeTakeFirstOrThrow(); + expect(row.has_possible_conflict).toBe(1); + const statuses = await ctx.db.selectFrom('service_status').selectAll().execute(); + expect(statuses).toHaveLength(2); + for (const status of statuses) { + expect(status.last_attempt_at).toBe(window.start.toString()); + expect(status.error).toBeNull(); + } + } finally { + await ctx.db.destroy(); + } +}); + +test('refreshCycle flags conflicts using the stale mirror after a provider failure', async () => { + const ctx = await ctxWithDb({ meetings: workMeetings }); + try { + await ctx.db.insertInto('appointments').values(appointment).execute(); + await replaceCalendarBusy(ctx.db, 'work', [ + { start: appointment.start_time, end: appointment.end_time } + ]); + vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response('boom', { status: 500 })); + + await refreshCycle(ctx, { now: window.start }); + + const row = await ctx.db + .selectFrom('appointments') + .select('has_possible_conflict') + .executeTakeFirstOrThrow(); + expect(row.has_possible_conflict).toBe(1); + const statuses = await ctx.db.selectFrom('service_status').selectAll().execute(); + expect(statuses).toHaveLength(2); + for (const status of statuses) { + expect(status.last_attempt_at).toBe(window.start.toString()); + expect(status.error).toEqual(expect.stringContaining('500')); + } + } finally { + await ctx.db.destroy(); + } +}); + +test('refreshCycle lets conflict writes fail without marking a successful provider refresh as failed', async () => { + const ctx = await ctxWithDb({ meetings: workMeetings }); + try { + await ctx.db.insertInto('appointments').values(appointment).execute(); + await sql`create trigger fail_conflict before update of has_possible_conflict on appointments + begin select raise(abort, 'conflict unavailable'); end`.execute(ctx.db); + vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response(oneEvent('r1'), { status: 207 })); + const log = vi.spyOn(ctx.logger, 'error'); + const metric = vi.spyOn(calendarRefreshTotal, 'inc'); + + await expect(refreshCycle(ctx, { now: window.start })).rejects.toThrow('conflict unavailable'); + + expect(log).not.toHaveBeenCalled(); + expect(metric).toHaveBeenCalledExactlyOnceWith({ + calendar_id: 'work', + provider_type: 'caldav', + status: 'success' + }); + const statuses = await ctx.db.selectFrom('service_status').selectAll().execute(); + expect(statuses).toHaveLength(2); + for (const status of statuses) { + expect(status.error).toBeNull(); + expect(status.last_ok_at).toBe(window.start.toString()); + } + } finally { + await ctx.db.destroy(); + } +}); + +test('refreshCalendars keeps independent due intervals and retries a stale failure on later passes', async () => { + const ctx = await ctxWithDb({ + providers: { + 'work-dav': { + ...workDav, + calendars: { + ...workDav.calendars, + other: { href: 'https://cal.example.com/other/', sync: { refresh_every_minutes: 30 } } + } + } + }, + meetings: { + chat: { ...workMeetings.chat, additional_busy_calendars: ['other'] } + } + }); + try { + for (const name of ['work', 'other']) { + await recordServiceOutcome( + ctx.db, + { kind: 'calendar', name }, + { at: window.start.toString(), via: 'refresh' } + ); + } + const fetch = vi.spyOn(globalThis, 'fetch').mockImplementation(async () => { + return new Response('boom', { status: 500 }); + }); + + await refreshCalendars(ctx, { now: window.start.add({ minutes: 9 }) }); + expect(fetch).not.toHaveBeenCalled(); + await refreshCalendars(ctx, { now: window.start.add({ minutes: 10 }) }); + await refreshCalendars(ctx, { now: window.start.add({ minutes: 11 }) }); + expect(fetch.mock.calls.map(([url]) => String(url))).toEqual([ + 'https://cal.example.com/work/', + 'https://cal.example.com/work/' + ]); + + await refreshCalendars(ctx, { now: window.start.add({ minutes: 30 }) }); + expect(fetch.mock.calls.slice(2).map(([url]) => String(url))).toEqual([ + 'https://cal.example.com/work/', + 'https://cal.example.com/other/' + ]); + } finally { + await ctx.db.destroy(); + } +}); + +test('refreshCalendars uses a reloaded calendar interval on its next pass', async () => { + const ctx = await ctxWithDb({ meetings: workMeetings }); + try { + await recordServiceOutcome( + ctx.db, + { kind: 'calendar', name: 'work' }, + { at: window.start.toString(), via: 'refresh' } + ); + const fetch = vi + .spyOn(globalThis, 'fetch') + .mockResolvedValue(new Response(oneEvent('r1'), { status: 207 })); + const now = window.start.add({ minutes: 5 }); + await refreshCalendars(ctx, { now }); + expect(fetch).not.toHaveBeenCalled(); + + ctx.config = { + ...ctx.config, + providers: { + 'work-dav': { + ...workDav, + calendars: { work: { ...workDav.calendars.work, sync: { refresh_every_minutes: 5 } } } + } + } + }; + await refreshCalendars(ctx, { now }); + expect(fetch).toHaveBeenCalledTimes(1); + } finally { + await ctx.db.destroy(); + } +}); diff --git a/apps/worker/src/calendar/refresh.ts b/apps/worker/src/calendar/refresh.ts index 2320dbf8..1f2e1da5 100644 --- a/apps/worker/src/calendar/refresh.ts +++ b/apps/worker/src/calendar/refresh.ts @@ -1,5 +1,5 @@ import { findCalendar, type ResolvedCalendar, type WhenConfiguration } from '@when/config'; -import type { ExpandWindow } from '@when/calendar'; +import type { ExpandWindow, Interval } from '@when/calendar'; import { busyCalendarsFor, fetchBusyIntervals } from '@when/calendar'; import { listOwnEventIds, @@ -52,21 +52,11 @@ export async function refreshCalendar( ): Promise { const at = (opts.now ?? window.start).toString(); const via = opts.via ?? 'refresh'; + const ownEventIds = await listOwnEventIds(ctx.db, cal.name); + const excludeUids = new Set(ownEventIds); + let intervals: Interval[]; try { - const excludeUids = new Set(await listOwnEventIds(ctx.db, cal.name)); - const intervals = await fetchBusyIntervals(cal, window, { excludeUids }); - await replaceCalendarBusy( - ctx.db, - cal.name, - intervals.map((i) => ({ start: i.start.toString(), end: i.end.toString() })) - ); - await recordRefreshOutcome(ctx, cal, { at, via }); - calendarRefreshTotal.inc({ - calendar_id: cal.name, - provider_type: cal.provider.type, - status: 'success' - }); - return { ok: true, busyCount: intervals.length }; + intervals = await fetchBusyIntervals(cal, window, { excludeUids }); } catch (err) { const error = err instanceof Error ? err.message : String(err); ctx.logger.error( @@ -84,6 +74,18 @@ export async function refreshCalendar( }); return { ok: false, error }; } + await replaceCalendarBusy( + ctx.db, + cal.name, + intervals.map((i) => ({ start: i.start.toString(), end: i.end.toString() })) + ); + await recordRefreshOutcome(ctx, cal, { at, via }); + calendarRefreshTotal.inc({ + calendar_id: cal.name, + provider_type: cal.provider.type, + status: 'success' + }); + return { ok: true, busyCount: intervals.length }; } // A provider backing no calendars has no other source of status. @@ -109,8 +111,6 @@ export async function refreshCalendars( ctx.logger.warn({ calendarId: id }, 'busy calendar id not found in calendars; skipping'); continue; } - // Each calendar refreshes on its own interval; a never-synced or failing one - // (no recent success) is always due, so failures retry every tick. if (!isDue(lastSuccess.get(id) ?? null, cal, now)) continue; await refreshCalendar(ctx, cal, refreshWindow(ctx.config, id, now), { ...opts, now }); } diff --git a/apps/worker/src/calendar/scheduler.test.ts b/apps/worker/src/calendar/scheduler.test.ts deleted file mode 100644 index 6a35cf53..00000000 --- a/apps/worker/src/calendar/scheduler.test.ts +++ /dev/null @@ -1,25 +0,0 @@ -import { afterEach, expect, test, vi } from 'vitest'; -import { createRefreshScheduler } from './scheduler.js'; - -afterEach(() => vi.useRealTimers()); - -test('scheduler runs on start, reschedules each interval, and stops', async () => { - vi.useFakeTimers(); - let calls = 0; - const scheduler = createRefreshScheduler(async () => { - calls++; - }, 1000); - - scheduler.start(); - expect(calls).toBe(1); - - await vi.advanceTimersByTimeAsync(1000); - expect(calls).toBe(2); - - await vi.advanceTimersByTimeAsync(1000); - expect(calls).toBe(3); - - scheduler.stop(); - await vi.advanceTimersByTimeAsync(5000); - expect(calls).toBe(3); -}); diff --git a/apps/worker/src/calendar/scheduler.ts b/apps/worker/src/calendar/scheduler.ts deleted file mode 100644 index d8074bc6..00000000 --- a/apps/worker/src/calendar/scheduler.ts +++ /dev/null @@ -1,32 +0,0 @@ -export interface RefreshScheduler { - start(): void; - stop(): void; -} - -export function createRefreshScheduler( - run: () => Promise, - intervalMs: number -): RefreshScheduler { - let timer: ReturnType | null = null; - let stopped = false; - - async function tick(): Promise { - try { - await run(); - } finally { - if (!stopped) timer = setTimeout(() => void tick(), intervalMs); - } - } - - return { - start() { - stopped = false; - void tick(); - }, - stop() { - stopped = true; - if (timer) clearTimeout(timer); - timer = null; - } - }; -} diff --git a/apps/worker/src/calendar/sync-scanner.test.ts b/apps/worker/src/calendar/sync-scanner.test.ts deleted file mode 100644 index 8c3496c2..00000000 --- a/apps/worker/src/calendar/sync-scanner.test.ts +++ /dev/null @@ -1,66 +0,0 @@ -import { afterEach, expect, test, vi } from 'vitest'; -import { createCalendarSyncScanner } from './sync-scanner.js'; - -afterEach(() => vi.useRealTimers()); - -const flush = () => new Promise((r) => setTimeout(r, 0)); - -test('runs on request and reschedules on the floor', async () => { - vi.useFakeTimers(); - let calls = 0; - const scanner = createCalendarSyncScanner(async () => { - calls++; - }, 1000); - - scanner.requestScan(); - await vi.advanceTimersByTimeAsync(0); - expect(calls).toBe(1); - - await vi.advanceTimersByTimeAsync(1000); - expect(calls).toBe(2); - - scanner.stop(); - await vi.advanceTimersByTimeAsync(5000); - expect(calls).toBe(2); -}); - -test('coalesces requests that arrive during a scan into one rerun', async () => { - let calls = 0; - let release: () => void = () => {}; - const run = () => { - calls++; - return new Promise((r) => { - release = r; - }); - }; - const scanner = createCalendarSyncScanner(run, 10_000); - - scanner.requestScan(); - await flush(); - expect(calls).toBe(1); - - scanner.requestScan(); - scanner.requestScan(); - expect(calls).toBe(1); - - release(); - await flush(); - expect(calls).toBe(2); - - release(); - await flush(); - expect(calls).toBe(2); - - scanner.stop(); -}); - -test('stop prevents new requests from running', async () => { - let calls = 0; - const scanner = createCalendarSyncScanner(async () => { - calls++; - }, 10_000); - scanner.stop(); - scanner.requestScan(); - await flush(); - expect(calls).toBe(0); -}); diff --git a/apps/worker/src/calendar/sync-scanner.ts b/apps/worker/src/calendar/sync-scanner.ts deleted file mode 100644 index f4a92a97..00000000 --- a/apps/worker/src/calendar/sync-scanner.ts +++ /dev/null @@ -1,54 +0,0 @@ -export interface CalendarSyncScanner { - requestScan(): void; - stop(): void; -} - -// One scan at a time (overlap guard via `scanning`); a request that arrives -// mid-scan is remembered and runs once more after, so bursts collapse into a -// single extra pass. Each run reschedules a floor timer, so an idle worker still -// scans every `floorMs`. -export function createCalendarSyncScanner( - run: () => Promise, - floorMs: number -): CalendarSyncScanner { - let scanning = false; - let rescanRequested = false; - let floorTimer: ReturnType | null = null; - let stopped = false; - - function requestScan(): void { - if (stopped) return; - if (scanning) { - rescanRequested = true; - return; - } - scanning = true; - void loop(); - } - - async function loop(): Promise { - try { - do { - rescanRequested = false; - await run(); - } while (rescanRequested && !stopped); - } catch { - // A scan threw; the floor timer below retries. - } finally { - scanning = false; - if (!stopped) { - if (floorTimer) clearTimeout(floorTimer); - floorTimer = setTimeout(requestScan, floorMs); - } - } - } - - return { - requestScan, - stop() { - stopped = true; - if (floorTimer) clearTimeout(floorTimer); - floorTimer = null; - } - }; -} diff --git a/apps/worker/src/index.test.ts b/apps/worker/src/index.test.ts new file mode 100644 index 00000000..1ba6590a --- /dev/null +++ b/apps/worker/src/index.test.ts @@ -0,0 +1,388 @@ +import { EventEmitter } from 'node:events'; +import { afterEach, beforeEach, expect, test, vi } from 'vitest'; +import type { ReloadResult, WhenConfiguration } from '@when/config'; +import { expirePendingAppointments, refreshCalendars, syncCalendars } from '@when/jobs'; + +const mocks = vi.hoisted(() => ({ + loadConfig: vi.fn(), + watchConfig: vi.fn(), + stopWatch: vi.fn(), + openDb: vi.fn(), + runMigrations: vi.fn(), + destroy: vi.fn(), + initOpenWorkflow: vi.fn(), + closeOpenWorkflow: vi.fn(), + newWorker: vi.fn(), + start: vi.fn(), + stop: vi.fn(), + runWorkflow: vi.fn(), + result: vi.fn(), + setLogger: vi.fn(), + createMailer: vi.fn(), + setWorkerContext: vi.fn(), + createHealthServer: vi.fn(), + createEngineLogger: vi.fn(), + registerWorkflows: vi.fn(), + logger: { info: vi.fn(), warn: vi.fn(), error: vi.fn() } +})); + +vi.mock('@when/config', async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, loadConfig: mocks.loadConfig, watchConfig: mocks.watchConfig }; +}); +vi.mock('@when/db', () => ({ openDb: mocks.openDb, runMigrations: mocks.runMigrations })); +vi.mock('@when/jobs', async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + initOpenWorkflow: mocks.initOpenWorkflow, + closeOpenWorkflow: mocks.closeOpenWorkflow + }; +}); +vi.mock('@when/calendar', () => ({ setLogger: mocks.setLogger })); +vi.mock('./email/smtp.js', () => ({ createMailer: mocks.createMailer })); +vi.mock('./services/context.js', () => ({ setWorkerContext: mocks.setWorkerContext })); +vi.mock('./services/health.js', () => ({ createHealthServer: mocks.createHealthServer })); +vi.mock('./services/logger.js', () => ({ logger: mocks.logger })); +vi.mock('./services/engine-logger.js', () => ({ createEngineLogger: mocks.createEngineLogger })); +vi.mock('./workflows/index.js', () => ({ registerWorkflows: mocks.registerWorkflows })); + +const config = { + user: { name: 'Owner', email: 'owner@example.test', timezone: 'UTC' }, + database: { app: '/fixture/app.sqlite', queue: '/fixture/queue.sqlite' }, + smtp: { host: 'smtp.example.test', port: 587, username: 'owner', password: 'test' } +} as WhenConfiguration; +const db = { destroy: mocks.destroy }; +const mailer = {}; +const engineLogger = {}; +const worker = { start: mocks.start, stop: mocks.stop }; +const client = { newWorker: mocks.newWorker, runWorkflow: mocks.runWorkflow }; +let events: string[]; +let signals: Map void>; +let reload: (result: ReloadResult) => void; +let server: EventEmitter & { + listening: boolean; + listen: ReturnType; + close: ReturnType; +}; + +function finishListening() { + server.listening = true; + server.emit('listening'); +} + +async function boot() { + await import('./index.js'); + await vi.waitFor(() => { + expect(mocks.watchConfig).toHaveBeenCalledOnce(); + expect(signals.has('SIGTERM')).toBe(true); + expect(signals.has('SIGINT')).toBe(true); + }); +} + +beforeEach(() => { + vi.resetModules(); + vi.resetAllMocks(); + vi.stubEnv('PORT', '19099'); + events = []; + signals = new Map(); + vi.spyOn(process, 'on').mockImplementation((event, listener) => { + if (event === 'SIGTERM' || event === 'SIGINT') signals.set(event, listener); + return process; + }); + vi.spyOn(process, 'exit').mockImplementation(() => undefined as never); + mocks.loadConfig.mockResolvedValue(config); + mocks.openDb.mockReturnValue(db); + mocks.runMigrations.mockResolvedValue([]); + mocks.createMailer.mockReturnValue(mailer); + mocks.createEngineLogger.mockReturnValue(engineLogger); + mocks.initOpenWorkflow.mockReturnValue(client); + mocks.newWorker.mockReturnValue(worker); + mocks.start.mockImplementation(async () => { + events.push('worker:start'); + }); + mocks.runWorkflow.mockImplementation(async (spec: { name: string }) => { + events.push(`enqueue:${spec.name}`); + return { result: mocks.result }; + }); + mocks.stopWatch.mockImplementation(() => { + events.push('watch:stop'); + }); + mocks.watchConfig.mockImplementation((callback: typeof reload) => { + reload = callback; + return mocks.stopWatch; + }); + mocks.stop.mockImplementation(async () => { + events.push('worker:stop'); + }); + mocks.closeOpenWorkflow.mockImplementation(async () => { + events.push('queue:close'); + }); + mocks.destroy.mockImplementation(async () => { + events.push('db:close'); + }); + server = Object.assign(new EventEmitter(), { + listening: false, + listen: vi.fn(() => { + events.push('server:listen'); + queueMicrotask(finishListening); + return server; + }), + close: vi.fn((callback: (error?: Error) => void) => { + events.push('server:close'); + server.listening = false; + queueMicrotask(() => { + events.push('server:closed'); + server.emit('close'); + callback(); + }); + return server; + }) + }); + mocks.createHealthServer.mockReturnValue(server); +}); + +afterEach(() => { + vi.restoreAllMocks(); + vi.unstubAllEnvs(); +}); + +test('registers workflows before starting one default worker, then enqueues all maintenance', async () => { + const started = Promise.withResolvers(); + mocks.start.mockImplementation(async () => { + events.push('worker:start'); + await started.promise; + events.push('worker:started'); + }); + await import('./index.js'); + await vi.waitFor(() => expect(mocks.start).toHaveBeenCalledOnce()); + expect(mocks.runWorkflow).not.toHaveBeenCalled(); + expect(mocks.createHealthServer).not.toHaveBeenCalled(); + expect(mocks.loadConfig).toHaveBeenCalledWith(); + expect(mocks.openDb).toHaveBeenCalledWith(config.database.app); + expect(mocks.runMigrations).toHaveBeenCalledWith(db); + expect(mocks.createMailer).toHaveBeenCalledWith(config, mocks.logger); + expect(mocks.setWorkerContext).toHaveBeenCalledWith({ config, db, mailer, logger: mocks.logger }); + expect(mocks.setLogger).toHaveBeenCalledWith(mocks.logger); + expect(mocks.createEngineLogger).toHaveBeenCalledWith(mocks.logger); + expect(mocks.initOpenWorkflow).toHaveBeenCalledWith({ + dbPath: config.database.queue, + logger: engineLogger + }); + expect(mocks.registerWorkflows).toHaveBeenCalledOnce(); + expect(mocks.initOpenWorkflow).toHaveBeenCalledBefore(mocks.registerWorkflows); + expect(mocks.registerWorkflows).toHaveBeenCalledBefore(mocks.newWorker); + expect(mocks.newWorker).toHaveBeenCalledWith(); + + started.resolve(); + await vi.waitFor(() => expect(mocks.watchConfig).toHaveBeenCalledOnce()); + expect(mocks.runWorkflow.mock.calls).toEqual([ + [refreshCalendars, {}], + [expirePendingAppointments, {}], + [syncCalendars, {}] + ]); + expect(mocks.runWorkflow.mock.calls.map(([spec]) => spec.name)).toEqual([ + 'refresh-calendars', + 'expire-pending-appointments', + 'sync-calendars' + ]); + expect(mocks.result).not.toHaveBeenCalled(); + expect(events).toEqual([ + 'worker:start', + 'worker:started', + 'enqueue:refresh-calendars', + 'enqueue:expire-pending-appointments', + 'enqueue:sync-calendars', + 'server:listen' + ]); + expect(server.listen.mock.calls[0][0]).toBe(19099); + const getConfig = mocks.createHealthServer.mock.calls[0][0] as () => WhenConfiguration; + expect(getConfig()).toBe(config); + expect(process.exit).not.toHaveBeenCalled(); +}); + +test.each([0, 1, 2])( + 'awaits startup enqueue %i and HTTP listening before watching config or installing signals', + async (index) => { + const enqueued = Promise.withResolvers(); + for (let preceding = 0; preceding < index; preceding++) { + mocks.runWorkflow.mockResolvedValueOnce({ result: mocks.result }); + } + mocks.runWorkflow.mockImplementationOnce(() => enqueued.promise); + server.listen.mockImplementation(() => server); + await import('./index.js'); + await vi.waitFor(() => expect(mocks.runWorkflow).toHaveBeenCalledTimes(index + 1)); + expect(mocks.createHealthServer).not.toHaveBeenCalled(); + enqueued.resolve(); + await vi.waitFor(() => expect(server.listen).toHaveBeenCalledOnce()); + expect(mocks.runWorkflow).toHaveBeenCalledTimes(3); + expect(mocks.watchConfig).not.toHaveBeenCalled(); + expect(signals.size).toBe(0); + finishListening(); + await vi.waitFor(() => expect(mocks.watchConfig).toHaveBeenCalledOnce()); + expect(signals.size).toBe(2); + } +); + +test('signals and restart-required config changes share one safe resource shutdown', async () => { + await boot(); + const stopped = Promise.withResolvers(); + const httpClosed = Promise.withResolvers(); + const queueClosed = Promise.withResolvers(); + mocks.closeOpenWorkflow.mockImplementation(async () => { + events.push('queue:close'); + await queueClosed.promise; + }); + mocks.stop.mockImplementation(async () => { + events.push('worker:stop'); + await stopped.promise; + events.push('worker:stopped'); + }); + server.close.mockImplementation((callback: () => void) => { + events.push('server:close'); + void httpClosed.promise.then(() => { + events.push('server:closed'); + callback(); + }); + return server; + }); + events = []; + signals.get('SIGTERM')!(); + signals.get('SIGINT')!(); + reload({ ok: true, config: { ...config, database: { ...config.database, app: '/new.sqlite' } } }); + expect(events).toEqual(['watch:stop', 'server:close', 'worker:stop']); + expect(mocks.closeOpenWorkflow).not.toHaveBeenCalled(); + expect(mocks.destroy).not.toHaveBeenCalled(); + expect(process.exit).not.toHaveBeenCalled(); + stopped.resolve(); + await vi.waitFor(() => expect(events).toContain('worker:stopped')); + expect(mocks.closeOpenWorkflow).not.toHaveBeenCalled(); + httpClosed.resolve(); + await vi.waitFor(() => expect(mocks.closeOpenWorkflow).toHaveBeenCalledOnce()); + expect(mocks.destroy).not.toHaveBeenCalled(); + expect(process.exit).not.toHaveBeenCalled(); + queueClosed.resolve(); + await vi.waitFor(() => expect(process.exit).toHaveBeenCalledWith(0)); + expect(events).toEqual([ + 'watch:stop', + 'server:close', + 'worker:stop', + 'worker:stopped', + 'server:closed', + 'queue:close', + 'db:close' + ]); + for (const close of [ + mocks.stopWatch, + server.close, + mocks.stop, + mocks.closeOpenWorkflow, + mocks.destroy + ]) { + expect(close).toHaveBeenCalledOnce(); + } +}); + +test.each(['migration', 'start', 'enqueue', 'listen'] as const)( + '%s failure awaits cleanup of partially created resources before exiting', + async (stage) => { + const failure = new Error(`${stage} failed`); + if (stage === 'migration') mocks.runMigrations.mockRejectedValue(failure); + if (stage === 'start') mocks.start.mockRejectedValue(failure); + if (stage === 'enqueue') mocks.runWorkflow.mockRejectedValueOnce(failure); + if (stage === 'listen') { + server.listen.mockImplementation(() => { + queueMicrotask(() => server.emit('error', failure)); + return server; + }); + } + const databaseClosed = Promise.withResolvers(); + mocks.destroy.mockImplementation(async () => { + events.push('db:close'); + await databaseClosed.promise; + }); + await import('./index.js'); + await vi.waitFor(() => expect(mocks.destroy).toHaveBeenCalledOnce()); + expect(process.exit).not.toHaveBeenCalled(); + expect(mocks.watchConfig).not.toHaveBeenCalled(); + expect(signals.size).toBe(0); + if (stage === 'migration') { + expect(mocks.initOpenWorkflow).not.toHaveBeenCalled(); + expect(mocks.stop).not.toHaveBeenCalled(); + } else { + expect(mocks.stop).toHaveBeenCalledOnce(); + expect(mocks.stop).toHaveBeenCalledBefore(mocks.closeOpenWorkflow); + expect(mocks.closeOpenWorkflow).toHaveBeenCalledBefore(mocks.destroy); + } + if (stage !== 'listen') expect(mocks.createHealthServer).not.toHaveBeenCalled(); + databaseClosed.resolve(); + await vi.waitFor(() => expect(process.exit).toHaveBeenCalledWith(1)); + expect(mocks.logger.error).toHaveBeenCalledWith( + { error: failure.message }, + 'worker failed to start' + ); + } +); + +test.each(['SIGTERM', 'SIGINT', 'config-change'])( + '%s handles rejected worker stop without closing live connections', + async (trigger) => { + await boot(); + mocks.stop.mockRejectedValue(new Error('active work did not stop')); + if (trigger === 'config-change') { + reload({ + ok: true, + config: { ...config, smtp: { ...config.smtp, host: 'new.example.test' } } + }); + } else { + signals.get(trigger)!(); + } + await vi.waitFor(() => expect(process.exit).toHaveBeenCalledWith(1)); + expect(mocks.logger.error).toHaveBeenCalled(); + expect(process.exit).not.toHaveBeenCalledWith(0); + expect(mocks.stopWatch).toHaveBeenCalledOnce(); + expect(server.close).toHaveBeenCalledOnce(); + expect(mocks.closeOpenWorkflow).not.toHaveBeenCalled(); + expect(mocks.destroy).not.toHaveBeenCalled(); + } +); + +test('queue close rejection still closes the app database and exits unsuccessfully', async () => { + await boot(); + mocks.closeOpenWorkflow.mockRejectedValue(new Error('queue close failed')); + signals.get('SIGTERM')!(); + await vi.waitFor(() => expect(process.exit).toHaveBeenCalledWith(1)); + expect(mocks.stop).toHaveBeenCalledBefore(mocks.closeOpenWorkflow); + expect(mocks.closeOpenWorkflow).toHaveBeenCalledBefore(mocks.destroy); + expect(mocks.destroy).toHaveBeenCalledOnce(); + expect(mocks.logger.error).toHaveBeenCalled(); + expect(process.exit).not.toHaveBeenCalledWith(0); +}); + +test('startup cleanup handles a failed stop without closing connections or losing the boot error', async () => { + mocks.start.mockRejectedValue(new Error('start failed')); + mocks.stop.mockRejectedValue(new Error('stop failed')); + await import('./index.js'); + await vi.waitFor(() => expect(process.exit).toHaveBeenCalledWith(1)); + expect(mocks.stop).toHaveBeenCalledOnce(); + expect(mocks.closeOpenWorkflow).not.toHaveBeenCalled(); + expect(mocks.destroy).not.toHaveBeenCalled(); + expect(mocks.logger.error.mock.calls).toEqual([ + [{ error: 'stop failed' }, 'worker startup cleanup failed'], + [{ error: 'start failed' }, 'worker failed to start'] + ]); +}); + +test('startup cleanup closes the app database even if closing the queue fails', async () => { + mocks.runWorkflow.mockRejectedValueOnce(new Error('enqueue failed')); + mocks.closeOpenWorkflow.mockRejectedValue(new Error('queue close failed')); + await import('./index.js'); + await vi.waitFor(() => expect(process.exit).toHaveBeenCalledWith(1)); + expect(mocks.stop).toHaveBeenCalledBefore(mocks.closeOpenWorkflow); + expect(mocks.closeOpenWorkflow).toHaveBeenCalledBefore(mocks.destroy); + expect(mocks.destroy).toHaveBeenCalledOnce(); + expect(mocks.logger.error.mock.calls).toEqual([ + [{ error: 'queue close failed' }, 'worker startup cleanup failed'], + [{ error: 'enqueue failed' }, 'worker failed to start'] + ]); +}); diff --git a/apps/worker/src/index.ts b/apps/worker/src/index.ts index c4413e98..84015293 100644 --- a/apps/worker/src/index.ts +++ b/apps/worker/src/index.ts @@ -1,109 +1,44 @@ -import { loadConfig, watchConfig } from '@when/config'; -import { expireStalePending, openDb, runMigrations } from '@when/db'; -import { initOpenWorkflow } from '@when/jobs'; -import { setLogger } from '@when/calendar'; -import { createMailer } from './email/smtp.js'; -import { applyConfig } from './services/config-reload.js'; -import { setWorkerContext, type WorkerContext } from './services/context.js'; -import { createHealthServer } from './services/health.js'; +import { loadConfig } from '@when/config'; +import { openDb } from '@when/db'; +import { closeOpenWorkflow } from '@when/jobs'; import { logger } from './services/logger.js'; -import { createEngineLogger } from './services/engine-logger.js'; -import { registerWorkflows } from './workflows/index.js'; -import { registerSyncCalendarsWorkflow } from './workflows/sync-calendars.js'; -import { refreshCycle } from './calendar/refresh.js'; -import { createRefreshScheduler } from './calendar/scheduler.js'; -import { scanOnce } from './calendar/sync.js'; -import { createCalendarSyncScanner } from './calendar/sync-scanner.js'; -import { calendarSyncTotal } from './services/metrics.js'; - -const DEFAULT_PORT = 9000; -const CALENDAR_SYNC_FLOOR_MS = 10 * 60_000; -const REFRESH_TICK_MINUTES = 5; -const EXPIRE_TICK_MINUTES = 60; +import { createShutdown, type ShutdownResources } from './services/lifecycle.js'; +import { + initializeWorkerContext, + startWorkflowWorker, + enqueueStartupMaintenance, + startHealthServer, + watchConfigAndSignals +} from './services/startup.js'; async function main(): Promise { const config = await loadConfig(); logger.info({ user: config.user.email }, 'config loaded'); const db = openDb(config.database.app); - const applied = await runMigrations(db); - if (applied.length > 0) logger.info({ migrations: applied }, 'migrations applied'); - - const ctx: WorkerContext = { config, logger, db, mailer: createMailer(config, logger) }; - setWorkerContext(ctx); - - setLogger(logger); - - const calendarSync = createCalendarSyncScanner(async () => { - try { - await scanOnce(ctx); - calendarSyncTotal.inc({ status: 'success' }); - } catch (err) { - calendarSyncTotal.inc({ status: 'failure' }); - logger.error( - { - error: err instanceof Error ? err.message : String(err) - }, - 'calendar sync failed' - ); - } - }, CALENDAR_SYNC_FLOOR_MS); - - const client = initOpenWorkflow({ - dbPath: config.database.queue, - logger: createEngineLogger(logger) - }); - registerWorkflows(); - registerSyncCalendarsWorkflow(calendarSync); - const worker = client.newWorker(); - await worker.start(); - logger.info('worker started'); - - calendarSync.requestScan(); - - const refresh = createRefreshScheduler(() => refreshCycle(ctx), REFRESH_TICK_MINUTES * 60_000); - refresh.start(); - logger.info({ tickMinutes: REFRESH_TICK_MINUTES }, 'calendar refresh scheduled'); - - const expireSweep = createRefreshScheduler(async () => { - try { - const expired = await expireStalePending(ctx.db, Temporal.Now.instant().toString()); - if (expired > 0) logger.info({ count: expired }, 'expired stale pending requests'); - } catch (err) { - logger.error( - { - error: err instanceof Error ? err.message : String(err) - }, - 'expire sweep failed' - ); - } - }, EXPIRE_TICK_MINUTES * 60_000); - expireSweep.start(); - logger.info({ tickMinutes: EXPIRE_TICK_MINUTES }, 'expiry sweep scheduled'); - - const port = Number(process.env.PORT) || DEFAULT_PORT; - const server = createHealthServer(() => ctx.config); - server.listen(port, () => logger.info({ port }, 'health server listening')); - - let stopWatch: () => void = () => {}; - - const shutdown = async (signal: string): Promise => { - logger.info({ signal }, 'worker shutting down'); - stopWatch(); - refresh.stop(); - expireSweep.stop(); - calendarSync.stop(); - server.close(); - await worker.stop(); - process.exit(0); + const resources: ShutdownResources = { + closeQueue: closeOpenWorkflow, + closeDatabase: () => db.destroy() }; + const shutdown = createShutdown(resources); - stopWatch = watchConfig((result) => - applyConfig(result, ctx, () => void shutdown('config-change')) - ); - - process.on('SIGTERM', () => void shutdown('SIGTERM')); - process.on('SIGINT', () => void shutdown('SIGINT')); + try { + const ctx = await initializeWorkerContext(config, db); + const client = await startWorkflowWorker(ctx, resources); + await enqueueStartupMaintenance(client); + await startHealthServer(ctx, resources); + watchConfigAndSignals(ctx, resources, shutdown); + } catch (error) { + try { + await shutdown(); + } catch (shutdownError) { + logger.error( + { error: shutdownError instanceof Error ? shutdownError.message : String(shutdownError) }, + 'worker startup cleanup failed' + ); + } + throw error; + } } main().catch((err) => { diff --git a/apps/worker/src/services/lifecycle.integration.test.ts b/apps/worker/src/services/lifecycle.integration.test.ts new file mode 100644 index 00000000..047f67ec --- /dev/null +++ b/apps/worker/src/services/lifecycle.integration.test.ts @@ -0,0 +1,157 @@ +import { expect, test } from 'vitest'; +import { OpenWorkflow, defineWorkflowSpec } from 'openworkflow'; +import { BackendSqlite } from 'openworkflow/sqlite'; +import { openDb, runMigrations } from '@when/db'; +import { sampleAppointment } from '../email/__fixtures__/appointment.js'; +import { createShutdown } from './lifecycle.js'; + +test('shutdown waits for an active native step before closing either database', async () => { + const db = openDb(':memory:'); + await runMigrations(db); + await db.insertInto('appointments').values(sampleAppointment).execute(); + const backend = BackendSqlite.connect(':memory:'); + const client = new OpenWorkflow({ backend }); + const started = Promise.withResolvers(); + const release = Promise.withResolvers(); + const events: string[] = []; + const spec = defineWorkflowSpec, 'done'>({ name: 'active-maintenance' }); + client.implementWorkflow(spec, async ({ step }) => { + await step.run({ name: 'active-pass' }, async () => { + started.resolve(); + await release.promise; + await db.updateTable('appointments').set({ note: 'finished' }).execute(); + events.push('pass:finished'); + }); + return 'done'; + }); + const worker = client.newWorker({ concurrency: 1 }); + const job = await client.runWorkflow(spec, {}); + let status: string | undefined; + let note: string | null | undefined; + const shutdown = createShutdown({ + worker, + closeQueue: async () => { + status = (await backend.getWorkflowRun({ workflowRunId: job.workflowRun.id }))?.status; + events.push('queue:close'); + await backend.stop(); + }, + closeDatabase: async () => { + note = (await db.selectFrom('appointments').select('note').executeTakeFirst())?.note; + events.push('database:close'); + await db.destroy(); + } + }); + try { + const claimedCount = await worker.tick(); + expect(claimedCount).toBe(1); + await started.promise; + const stopping = shutdown(); + expect(shutdown()).toBe(stopping); + expect(events).toEqual([]); + release.resolve(); + await stopping; + expect(events).toEqual(['pass:finished', 'queue:close', 'database:close']); + expect(status).toBe('completed'); + expect(note).toBe('finished'); + } finally { + release.resolve(); + await shutdown(); + } +}); + +test('shutdown closes a native queue without waiting for a future retry', async () => { + const backend = BackendSqlite.connect(':memory:'); + const client = new OpenWorkflow({ backend }); + const spec = defineWorkflowSpec, 'done'>({ name: 'parked-retry' }); + let calls = 0; + client.implementWorkflow(spec, async ({ step }) => { + await step.run( + { + name: 'retrying-pass', + retryPolicy: { maximumAttempts: 5, initialInterval: '1h', maximumInterval: '1h' } + }, + () => { + calls += 1; + throw new Error('temporary provider failure'); + } + ); + return 'done'; + }); + const worker = client.newWorker({ concurrency: 1 }); + const job = await client.runWorkflow(spec, {}); + let status: string | undefined; + let availableAt: Date | null | undefined; + let databaseClosed = false; + const shutdown = createShutdown({ + worker, + closeQueue: async () => { + const run = await backend.getWorkflowRun({ workflowRunId: job.workflowRun.id }); + status = run?.status; + availableAt = run?.availableAt; + await backend.stop(); + }, + closeDatabase: async () => { + databaseClosed = true; + } + }); + try { + const claimedCount = await worker.tick(); + expect(claimedCount).toBe(1); + await worker.stop(); + await shutdown(); + expect(status).toBe('pending'); + expect(availableAt?.getTime()).toBeGreaterThan(Date.now()); + expect(calls).toBe(1); + expect(databaseClosed).toBe(true); + } finally { + await shutdown(); + } +}); + +test('shutdown does not drain pending children of a parked native parent', async () => { + const backend = BackendSqlite.connect(':memory:'); + const client = new OpenWorkflow({ backend }); + const child = defineWorkflowSpec, 'sent'>({ name: 'parked-child' }); + const parent = defineWorkflowSpec, 'done'>({ name: 'parked-parent' }); + let childCalls = 0; + client.implementWorkflow(child, () => { + childCalls += 1; + return 'sent'; + }); + client.implementWorkflow(parent, async ({ step }) => { + await step.runWorkflow(child, {}, { name: 'delivery' }); + return 'done'; + }); + const worker = client.newWorker({ concurrency: 1 }); + await client.runWorkflow(parent, {}); + let states: { workflowName: string; status: string; workerId: string | null }[] = []; + const shutdown = createShutdown({ + worker, + closeQueue: async () => { + const page = await backend.listWorkflowRuns({ limit: 10 }); + states = page.data.map(({ workflowName, status, workerId }) => ({ + workflowName, + status, + workerId + })); + await backend.stop(); + }, + closeDatabase: async () => {} + }); + try { + const claimedCount = await worker.tick(); + expect(claimedCount).toBe(1); + await worker.stop(); + await shutdown(); + expect(states).toEqual( + expect.arrayContaining([ + { workflowName: child.name, status: 'pending', workerId: null }, + { workflowName: parent.name, status: 'running', workerId: null } + ]) + ); + expect(states).toHaveLength(2); + expect(childCalls).toBe(0); + } finally { + await shutdown(); + } +}); diff --git a/apps/worker/src/services/lifecycle.test.ts b/apps/worker/src/services/lifecycle.test.ts new file mode 100644 index 00000000..e9bfc73d --- /dev/null +++ b/apps/worker/src/services/lifecycle.test.ts @@ -0,0 +1,124 @@ +import { createServer } from 'node:http'; +import { afterEach, expect, test, vi } from 'vitest'; +import { createShutdown } from './lifecycle.js'; + +afterEach(() => vi.restoreAllMocks()); + +test('repeated shutdown requests share a promise and close resources in order', async () => { + const events: string[] = []; + const activePass = Promise.withResolvers(); + const server = createServer(); + vi.spyOn(server, 'listening', 'get').mockReturnValue(true); + let finishHttp = () => {}; + vi.spyOn(server, 'close').mockImplementation((callback) => { + events.push('http:stop'); + finishHttp = () => { + events.push('http:closed'); + callback?.(); + }; + return server; + }); + const stopWorker = vi.fn(async () => { + events.push('worker:stop'); + await activePass.promise; + events.push('worker:stopped'); + }); + const shutdown = createShutdown({ + server, + worker: { stop: stopWorker }, + stopWatch: () => events.push('watch:stop'), + closeQueue: async () => { + events.push('queue:close'); + }, + closeDatabase: async () => { + events.push('database:close'); + } + }); + + const first = shutdown(); + expect(shutdown()).toBe(first); + expect(events).toEqual(['watch:stop', 'http:stop', 'worker:stop']); + activePass.resolve(); + await vi.waitFor(() => expect(events).toContain('worker:stopped')); + expect(events).not.toContain('queue:close'); + finishHttp(); + await first; + expect(events).toEqual([ + 'watch:stop', + 'http:stop', + 'worker:stop', + 'worker:stopped', + 'http:closed', + 'queue:close', + 'database:close' + ]); + expect(shutdown()).toBe(first); + expect(stopWorker).toHaveBeenCalledOnce(); +}); + +test('partial startup closes the queue and database without a worker or server', async () => { + const closeQueue = vi.fn(async () => {}); + const closeDatabase = vi.fn(async () => {}); + const shutdown = createShutdown({ closeQueue, closeDatabase }); + await shutdown(); + await shutdown(); + expect(closeQueue).toHaveBeenCalledOnce(); + expect(closeDatabase).toHaveBeenCalledOnce(); + expect(closeQueue.mock.invocationCallOrder[0]).toBeLessThan( + closeDatabase.mock.invocationCallOrder[0] + ); +}); + +test('a failed worker stop leaves its database connections open', async () => { + const error = new Error('worker stop failed'); + const closeQueue = vi.fn(async () => {}); + const closeDatabase = vi.fn(async () => {}); + const stop = vi.fn().mockRejectedValue(error); + const shutdown = createShutdown({ worker: { stop }, closeQueue, closeDatabase }); + const pending = shutdown(); + await expect(pending).rejects.toBe(error); + expect(shutdown()).toBe(pending); + expect(stop).toHaveBeenCalledOnce(); + expect(closeQueue).not.toHaveBeenCalled(); + expect(closeDatabase).not.toHaveBeenCalled(); +}); + +test('a queue close failure still closes the application database', async () => { + const error = new Error('queue close failed'); + const closeDatabase = vi.fn(async () => {}); + const shutdown = createShutdown({ + closeQueue: vi.fn().mockRejectedValue(error), + closeDatabase + }); + await expect(shutdown()).rejects.toBe(error); + expect(closeDatabase).toHaveBeenCalledOnce(); +}); + +test.each(['callback', 'throw'])( + 'an HTTP close %s error is reported after database cleanup', + async (mode) => { + const error = new Error('HTTP close failed'); + const server = createServer(); + vi.spyOn(server, 'listening', 'get').mockReturnValue(true); + vi.spyOn(server, 'close').mockImplementation((callback) => { + if (mode === 'throw') throw error; + callback?.(error); + return server; + }); + const closeQueue = vi.fn(async () => {}); + const closeDatabase = vi.fn(async () => {}); + const shutdown = createShutdown({ server, closeQueue, closeDatabase }); + await expect(shutdown()).rejects.toBe(error); + expect(closeQueue).toHaveBeenCalledOnce(); + expect(closeDatabase).toHaveBeenCalledOnce(); + } +); + +test('a database close failure is propagated and not retried by another signal', async () => { + const error = new Error('database close failed'); + const closeDatabase = vi.fn().mockRejectedValue(error); + const shutdown = createShutdown({ closeQueue: async () => {}, closeDatabase }); + await expect(shutdown()).rejects.toBe(error); + await expect(shutdown()).rejects.toBe(error); + expect(closeDatabase).toHaveBeenCalledOnce(); +}); diff --git a/apps/worker/src/services/lifecycle.ts b/apps/worker/src/services/lifecycle.ts new file mode 100644 index 00000000..dd7d694c --- /dev/null +++ b/apps/worker/src/services/lifecycle.ts @@ -0,0 +1,38 @@ +import type { Server } from 'node:http'; +import type { Worker } from 'openworkflow'; + +export interface ShutdownResources { + worker?: Pick; + server?: Server; + stopWatch?: () => void; + closeQueue(): Promise; + closeDatabase(): Promise; +} + +export function createShutdown(resources: ShutdownResources): () => Promise { + let shutdownPromise: Promise | undefined; + return () => { + shutdownPromise ??= stopResources(resources); + return shutdownPromise; + }; +} + +async function stopResources(resources: ShutdownResources): Promise { + resources.stopWatch?.(); + const serverClosed = new Promise((resolve) => { + try { + if (resources.server?.listening) resources.server.close(resolve); + else resolve(undefined); + } catch (error) { + resolve(error); + } + }); + await resources.worker?.stop(); + const serverError = await serverClosed; + try { + await resources.closeQueue(); + } finally { + await resources.closeDatabase(); + } + if (serverError) throw serverError; +} diff --git a/apps/worker/src/services/startup.ts b/apps/worker/src/services/startup.ts new file mode 100644 index 00000000..312770c3 --- /dev/null +++ b/apps/worker/src/services/startup.ts @@ -0,0 +1,94 @@ +import { once } from 'node:events'; +import { watchConfig, type WhenConfiguration } from '@when/config'; +import { runMigrations } from '@when/db'; +import { + expirePendingAppointments, + initOpenWorkflow, + refreshCalendars, + syncCalendars +} from '@when/jobs'; +import { setLogger } from '@when/calendar'; +import type { OpenWorkflow } from 'openworkflow'; +import { createMailer } from '../email/smtp.js'; +import { registerWorkflows } from '../workflows/index.js'; +import { applyConfig } from './config-reload.js'; +import { setWorkerContext, type WorkerContext } from './context.js'; +import { createEngineLogger } from './engine-logger.js'; +import { createHealthServer } from './health.js'; +import type { ShutdownResources } from './lifecycle.js'; +import { logger } from './logger.js'; + +const DEFAULT_PORT = 9000; + +export async function initializeWorkerContext( + config: WhenConfiguration, + db: WorkerContext['db'] +): Promise { + const applied = await runMigrations(db); + if (applied.length > 0) logger.info({ migrations: applied }, 'migrations applied'); + + const ctx: WorkerContext = { config, logger, db, mailer: createMailer(config, logger) }; + setWorkerContext(ctx); + setLogger(logger); + return ctx; +} + +export async function startWorkflowWorker( + ctx: WorkerContext, + resources: ShutdownResources +): Promise { + const client = initOpenWorkflow({ + dbPath: ctx.config.database.queue, + logger: createEngineLogger(ctx.logger) + }); + registerWorkflows(); + const worker = client.newWorker(); + resources.worker = worker; + await worker.start(); + ctx.logger.info('worker started'); + return client; +} + +export async function enqueueStartupMaintenance(client: OpenWorkflow): Promise { + await client.runWorkflow(refreshCalendars, {}); + await client.runWorkflow(expirePendingAppointments, {}); + await client.runWorkflow(syncCalendars, {}); +} + +export async function startHealthServer( + ctx: WorkerContext, + resources: ShutdownResources +): Promise { + const port = Number(process.env.PORT) || DEFAULT_PORT; + const server = createHealthServer(() => ctx.config); + resources.server = server; + server.listen(port); + await once(server, 'listening'); + ctx.logger.info({ port }, 'health server listening'); +} + +export function watchConfigAndSignals( + ctx: WorkerContext, + resources: ShutdownResources, + shutdown: () => Promise +): void { + const requestShutdown = (signal: string): void => { + ctx.logger.info({ signal }, 'worker shutting down'); + void shutdown().then( + () => process.exit(0), + (error) => { + ctx.logger.error( + { error: error instanceof Error ? error.message : String(error) }, + 'worker shutdown failed' + ); + process.exit(1); + } + ); + }; + + resources.stopWatch = watchConfig((result) => + applyConfig(result, ctx, () => requestShutdown('config-change')) + ); + process.on('SIGTERM', () => requestShutdown('SIGTERM')); + process.on('SIGINT', () => requestShutdown('SIGINT')); +} diff --git a/apps/worker/src/workflows/expire-pending-appointments.ts b/apps/worker/src/workflows/expire-pending-appointments.ts new file mode 100644 index 00000000..addd7570 --- /dev/null +++ b/apps/worker/src/workflows/expire-pending-appointments.ts @@ -0,0 +1,27 @@ +import { expireStalePending } from '@when/db'; +import { expirePendingAppointments, type ExpirePendingAppointmentsResult } from '@when/jobs'; +import { systemClock, type Clock } from '../email/clock.js'; +import { getWorkerContext } from '../services/context.js'; +import { implementObservedWorkflow } from '../services/metrics.js'; + +interface ExpiryStep { + run(config: { name: string }, fn: () => Promise): Promise; +} + +export async function runExpirePendingAppointments( + step: ExpiryStep, + clock: Clock = systemClock +): Promise { + await step.run({ name: 'expire-pending-appointments' }, async () => { + const ctx = getWorkerContext(); + const count = await expireStalePending(ctx.db, clock.now().toISOString()); + if (count > 0) ctx.logger.info({ count }, 'expired stale pending requests'); + }); + return 'expired'; +} + +export function registerExpirePendingAppointmentsWorkflow(): void { + implementObservedWorkflow(expirePendingAppointments, ({ step }) => + runExpirePendingAppointments(step) + ); +} diff --git a/apps/worker/src/workflows/index.ts b/apps/worker/src/workflows/index.ts index ae096b4c..d9bb4414 100644 --- a/apps/worker/src/workflows/index.ts +++ b/apps/worker/src/workflows/index.ts @@ -5,6 +5,9 @@ import { registerTestEmailWorkflow } from './test-email.js'; import { registerProbeProviderWorkflows } from './probe-provider.js'; import { registerProbeCalendarWorkflow } from './probe-calendar.js'; import { registerGenerateVideoChatWorkflow } from './generate-video-chat.js'; +import { registerSyncCalendarsWorkflow } from './sync-calendars.js'; +import { registerRefreshCalendarsWorkflow } from './refresh-calendars.js'; +import { registerExpirePendingAppointmentsWorkflow } from './expire-pending-appointments.js'; /** * Register every workflow implementation on the openworkflow singleton. Called @@ -18,4 +21,7 @@ export function registerWorkflows(): void { registerProbeProviderWorkflows(); registerProbeCalendarWorkflow(); registerGenerateVideoChatWorkflow(); + registerSyncCalendarsWorkflow(); + registerRefreshCalendarsWorkflow(); + registerExpirePendingAppointmentsWorkflow(); } diff --git a/apps/worker/src/workflows/maintenance.integration.test.ts b/apps/worker/src/workflows/maintenance.integration.test.ts new file mode 100644 index 00000000..633b02dc --- /dev/null +++ b/apps/worker/src/workflows/maintenance.integration.test.ts @@ -0,0 +1,376 @@ +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { afterEach, beforeEach, expect, test, vi } from 'vitest'; +import { OpenWorkflow, type Worker } from 'openworkflow'; +import { BackendSqlite } from 'openworkflow/sqlite'; +import { openDb, runMigrations } from '@when/db'; +import { + refreshCalendars, + expirePendingAppointments, + syncCalendars, + testCalendar +} from '@when/jobs'; +import { fetchBusyIntervals, pushAppointment } from '@when/calendar'; +import { sampleAppointment, sampleConfig } from '../email/__fixtures__/appointment.js'; +import { setWorkerContext, type WorkerContext } from '../services/context.js'; +import { createLogger } from '../services/logger.js'; +import { registerWorkflows } from './index.js'; + +const engine = vi.hoisted(() => ({ client: null as OpenWorkflow | null })); +vi.mock('@when/jobs', async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, getOpenWorkflow: () => engine.client! }; +}); +vi.mock('@when/calendar', async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, fetchBusyIntervals: vi.fn(), pushAppointment: vi.fn() }; +}); + +const specs = [refreshCalendars, expirePendingAppointments, syncCalendars]; +const at = '2026-04-01T07:03:19.000Z'; +let directory: string; +let backend: BackendSqlite; +let client: OpenWorkflow; +let worker: Worker; +let ctx: WorkerContext; + +beforeEach(async () => { + vi.useFakeTimers({ toFake: ['Date'] }); + vi.setSystemTime(new Date(at)); + vi.mocked(fetchBusyIntervals).mockReset().mockResolvedValue([]); + vi.mocked(pushAppointment).mockReset().mockResolvedValue({ ok: false, reason: 'offline' }); + directory = await mkdtemp(join(tmpdir(), 'when-maintenance-')); + const db = openDb(join(directory, 'app.sqlite')); + await runMigrations(db); + ctx = { + db, + config: { ...sampleConfig, providers: {} }, + logger: createLogger(), + mailer: { send: async () => ({ ok: true }) } + }; + setWorkerContext(ctx); + backend = BackendSqlite.connect(join(directory, 'queue.sqlite')); + client = new OpenWorkflow({ backend }); + engine.client = client; + registerWorkflows(); + worker = client.newWorker({ concurrency: 1 }); +}); + +afterEach(async () => { + try { + await worker.stop(); + await backend.stop(); + await ctx.db.destroy(); + } finally { + vi.useRealTimers(); + await rm(directory, { recursive: true, force: true }); + } +}); + +async function schedules() { + const page = await client.schedules.list({ limit: 100 }); + expect(page.pagination.next).toBeNull(); + return page.data; +} + +async function runs() { + const page = await backend.listWorkflowRuns({ limit: 100 }); + expect(page.pagination.next).toBeNull(); + return page.data; +} + +async function drain(count: number) { + for (let i = 0; i < count; i++) { + const claimedCount = await worker.tick(); + expect(claimedCount).toBe(1); + await worker.stop(); + } +} + +function useCalendar() { + ctx.config = { + ...ctx.config, + providers: { + dav: { + type: 'caldav', + url: 'https://calendar.example.com', + username: 'owner', + password: 'secret', + calendars: { + work: { href: '/work/', sync: { refresh_every_minutes: 10 } } + } + } + }, + meetings: { + chat: { + title: 'Chat', + booking_calendar: 'work', + additional_busy_calendars: [], + booking_window_days: 60 + } + } as unknown as WorkerContext['config']['meetings'] + }; +} + +test('producer-only enqueue and implementation registration persist no schedules', async () => { + const run = await client.runWorkflow(syncCalendars, {}); + const schedulesBeforeDrain = await schedules(); + expect(schedulesBeforeDrain).toEqual([]); + const pendingRuns = await runs(); + expect(pendingRuns).toMatchObject([ + { id: run.workflowRun.id, workflowName: 'sync-calendars', input: {}, status: 'pending' } + ]); + await drain(1); + const syncResult = await run.result({ timeoutMs: 500 }); + expect(syncResult).toBe('requested'); + const schedulesAfterDrain = await schedules(); + expect(schedulesAfterDrain).toEqual([]); +}); + +test('native start upserts declarations and preserves pause across a file-backed restart', async () => { + await worker.start(); + await worker.stop(); + const initialSchedules = await schedules(); + expect(initialSchedules).toHaveLength(3); + await client.scheduleWorkflow( + refreshCalendars, + {}, + { + id: 'when:refresh-calendars', + cron: '*/7 * * * *', + timezone: 'UTC' + } + ); + await client.schedules.pause('when:sync-calendars'); + await backend.stop(); + backend = BackendSqlite.connect(join(directory, 'queue.sqlite')); + client = new OpenWorkflow({ backend }); + engine.client = client; + registerWorkflows(); + worker = client.newWorker({ concurrency: 1 }); + await worker.start(); + await worker.stop(); + const persisted = await schedules(); + expect(persisted).toHaveLength(3); + for (const spec of specs) { + const declaration = spec.schedules![0]!; + expect(persisted.find((schedule) => schedule.id === declaration.id)).toMatchObject({ + ...declaration, + cron: `0 ${declaration.cron}`, + workflowName: spec.name + }); + } + const pausedSchedule = await client.schedules.get('when:sync-calendars'); + expect(pausedSchedule).toMatchObject({ nextRunAt: null }); + const resumedSchedule = await client.schedules.resume('when:sync-calendars'); + expect(resumedSchedule).toMatchObject({ + nextRunAt: new Date('2026-04-01T07:10:00Z') + }); +}); + +test('UTC cron aligns to wall-clock boundaries rather than startup time', async () => { + await worker.start(); + await worker.stop(); + const persisted = await schedules(); + const expected = new Map([ + ['when:refresh-calendars', '2026-04-01T07:05:00.000Z'], + ['when:expire-pending-appointments', '2026-04-01T08:00:00.000Z'], + ['when:sync-calendars', '2026-04-01T07:10:00.000Z'] + ]); + for (const schedule of persisted) { + expect(schedule.timezone).toBe('UTC'); + expect(schedule.nextRunAt?.toISOString()).toBe(expected.get(schedule.id)); + } + const enqueuedBeforeFirstBoundary = await backend.enqueueScheduledRuns({ + before: new Date('2026-04-01T07:04:59Z'), + limit: 100 + }); + expect(enqueuedBeforeFirstBoundary).toBe(0); + const enqueuedAtFirstBoundary = await backend.enqueueScheduledRuns({ + before: new Date('2026-04-01T07:05:00Z'), + limit: 100 + }); + expect(enqueuedAtFirstBoundary).toBe(1); + const enqueuedBeforeSecondBoundary = await backend.enqueueScheduledRuns({ + before: new Date('2026-04-01T07:09:59Z'), + limit: 100 + }); + expect(enqueuedBeforeSecondBoundary).toBe(0); + const enqueuedAtSecondBoundary = await backend.enqueueScheduledRuns({ + before: new Date('2026-04-01T07:10:00Z'), + limit: 100 + }); + expect(enqueuedAtSecondBoundary).toBe(2); + vi.setSystemTime(new Date('2026-04-01T07:10:00Z')); + await drain(3); + const completedRuns = await runs(); + expect(completedRuns.every((run) => run.status === 'completed')).toBe(true); +}); + +test('downtime catches up once per schedule and explicit startup passes coexist with cron', async () => { + await worker.start(); + await worker.stop(); + vi.setSystemTime(new Date('2026-04-03T15:37:00Z')); + const before = new Date(); + const catchupCount = await backend.enqueueScheduledRuns({ before, limit: 100 }); + expect(catchupCount).toBe(3); + const repeatedCatchupCount = await backend.enqueueScheduledRuns({ before, limit: 100 }); + expect(repeatedCatchupCount).toBe(0); + for (const spec of specs) await client.runWorkflow(spec, {}); + await drain(6); + const completed = await runs(); + expect(completed).toHaveLength(6); + for (const spec of specs) { + const matching = completed.filter((run) => run.workflowName === spec.name); + expect(matching).toHaveLength(2); + expect(matching.map((run) => run.workflowScheduleId)).toEqual( + expect.arrayContaining([null, spec.schedules![0]!.id]) + ); + for (const run of matching) { + expect(run).toMatchObject({ status: 'completed', input: {}, idempotencyKey: null }); + const attempts = await backend.listStepAttempts({ workflowRunId: run.id, limit: 100 }); + expect(attempts.data).toMatchObject([{ status: 'completed', kind: 'function' }]); + } + } + for (const schedule of await schedules()) { + expect(schedule.nextRunAt!.getTime()).toBeGreaterThan(before.getTime()); + } +}); + +test('paused cron survives startup but does not suppress explicit startup work', async () => { + await worker.start(); + await worker.stop(); + for (const schedule of await schedules()) await client.schedules.pause(schedule.id); + vi.setSystemTime(new Date('2026-04-03T15:37:00Z')); + await worker.start(); + await worker.stop(); + const pausedSchedules = await schedules(); + expect(pausedSchedules.every((schedule) => schedule.nextRunAt === null)).toBe(true); + const enqueuedCount = await backend.enqueueScheduledRuns({ before: new Date(), limit: 100 }); + expect(enqueuedCount).toBe(0); + for (const spec of specs) await client.runWorkflow(spec, {}); + await drain(3); + const completedRuns = await runs(); + expect(completedRuns).toMatchObject([ + { status: 'completed', workflowScheduleId: null }, + { status: 'completed', workflowScheduleId: null }, + { status: 'completed', workflowScheduleId: null } + ]); +}); + +test('repeated sync requests serialize without coalescing and stop awaits the active pass', async () => { + useCalendar(); + await ctx.db + .insertInto('appointments') + .values({ ...sampleAppointment, event_type_id: 'chat' }) + .execute(); + const release = Promise.withResolvers(); + let entered = false; + let active = 0; + let maximumActive = 0; + vi.mocked(pushAppointment).mockImplementation(async () => { + active += 1; + maximumActive = Math.max(active, maximumActive); + entered = true; + await release.promise; + active -= 1; + return { ok: false, reason: 'offline' }; + }); + const first = await client.runWorkflow(syncCalendars, {}); + try { + const claimedCount = await worker.tick(); + expect(claimedCount).toBe(1); + await vi.waitUntil(() => entered, { timeout: 1000, interval: 5 }); + await client.runWorkflow(syncCalendars, {}); + await client.runWorkflow(syncCalendars, {}); + const claimedWhileActive = await worker.tick(); + expect(claimedWhileActive).toBe(0); + let stopped = false; + const stopping = worker.stop().then(() => { + stopped = true; + }); + await Promise.resolve(); + expect(stopped).toBe(false); + release.resolve(); + await stopping; + const syncResult = await first.result({ timeoutMs: 500 }); + expect(syncResult).toBe('requested'); + const runsAfterStop = await runs(); + expect(runsAfterStop.filter((run) => run.status === 'pending')).toHaveLength(2); + await drain(2); + const completedRuns = await runs(); + expect(completedRuns.every((run) => run.status === 'completed')).toBe(true); + expect(pushAppointment).toHaveBeenCalledTimes(3); + expect(maximumActive).toBe(1); + const runsForUniqueIds = await runs(); + expect(new Set(runsForUniqueIds.map((run) => run.id)).size).toBe(3); + } finally { + release.resolve(); + } +}); + +test('slow refresh blocks a queued probe at concurrency one until the real provider call finishes', async () => { + useCalendar(); + const release = Promise.withResolvers(); + let entered = false; + vi.mocked(fetchBusyIntervals).mockImplementationOnce(async () => { + entered = true; + await release.promise; + return []; + }); + const refresh = await client.runWorkflow(refreshCalendars, {}); + try { + const claimedCount = await worker.tick(); + expect(claimedCount).toBe(1); + await vi.waitUntil(() => entered, { timeout: 1000, interval: 5 }); + const probe = await client.runWorkflow(testCalendar, { name: 'work' }); + const claimedWhileActive = await worker.tick(); + expect(claimedWhileActive).toBe(0); + release.resolve(); + await worker.stop(); + const refreshResult = await refresh.result({ timeoutMs: 500 }); + expect(refreshResult).toBe('refreshed'); + const pendingProbe = await backend.getWorkflowRun({ workflowRunId: probe.workflowRun.id }); + expect(pendingProbe).toMatchObject({ + status: 'pending' + }); + expect(fetchBusyIntervals).toHaveBeenCalledTimes(1); + await drain(1); + const probeResult = await probe.result({ timeoutMs: 500 }); + expect(probeResult).toEqual({ busyCount: 0, days: 60 }); + expect(fetchBusyIntervals).toHaveBeenCalledTimes(2); + } finally { + release.resolve(); + } +}); + +test.each(specs)( + '$name local failure uses ten native attempts with 1s ×2 backoff capped at 100s', + async (spec) => { + await ctx.db.schema.dropTable('appointments').execute(); + const handle = await client.runWorkflow(spec, {}); + const delays = [1000, 2000, 4000, 8000, 16000, 32000, 64000, 100000, 100000]; + for (let i = 0; i < 10; i++) { + const started = Date.now(); + const claimedCount = await worker.tick(); + expect(claimedCount).toBe(1); + await worker.stop(); + const run = await backend.getWorkflowRun({ workflowRunId: handle.workflowRun.id }); + const attempts = await backend.listStepAttempts({ + workflowRunId: handle.workflowRun.id, + limit: 100 + }); + expect(attempts.data).toHaveLength(i + 1); + expect(attempts.data.every((attempt) => attempt.status === 'failed')).toBe(true); + if (i < 9) { + expect(run).toMatchObject({ status: 'pending', workerId: null }); + expect(run!.availableAt!.getTime() - started).toBe(delays[i]); + vi.setSystemTime(run!.availableAt!); + } else { + expect(run).toMatchObject({ status: 'failed', output: null, workerId: null }); + } + } + await expect(handle.result({ timeoutMs: 500 })).rejects.toThrow(); + } +); diff --git a/apps/worker/src/workflows/maintenance.test.ts b/apps/worker/src/workflows/maintenance.test.ts new file mode 100644 index 00000000..f3e1e0ae --- /dev/null +++ b/apps/worker/src/workflows/maintenance.test.ts @@ -0,0 +1,183 @@ +import { afterEach, beforeEach, expect, test, vi } from 'vitest'; +import { openDb, parseActionLog, runMigrations } from '@when/db'; +import { scanOnce } from '../calendar/sync.js'; +import { refreshCycle } from '../calendar/refresh.js'; +import { sampleAppointment, sampleConfig } from '../email/__fixtures__/appointment.js'; +import { setWorkerContext, type WorkerContext } from '../services/context.js'; +import { createLogger } from '../services/logger.js'; +import { calendarSyncTotal } from '../services/metrics.js'; +import { runSyncCalendars } from './sync-calendars.js'; +import { runRefreshCalendars } from './refresh-calendars.js'; +import { runExpirePendingAppointments } from './expire-pending-appointments.js'; + +vi.mock('../calendar/sync.js', () => ({ scanOnce: vi.fn() })); +vi.mock('../calendar/refresh.js', () => ({ refreshCycle: vi.fn() })); + +const at = '2026-04-01T12:00:00.000Z'; +const clock = { now: () => new Date(at), nowMs: () => Date.parse(at) }; +let ctx: WorkerContext; + +function makeStep() { + const configs: { name: string }[] = []; + return { + configs, + async run(config: { name: string }, fn: () => Promise): Promise { + configs.push(config); + const result = await fn(); + return result; + } + }; +} + +beforeEach(async () => { + vi.resetAllMocks(); + calendarSyncTotal.reset(); + const db = openDb(':memory:'); + await runMigrations(db); + ctx = { + db, + config: sampleConfig, + logger: createLogger(), + mailer: { send: async () => ({ ok: true }) } + }; + vi.spyOn(ctx.logger, 'info'); + setWorkerContext(ctx); +}); + +afterEach(async () => { + vi.restoreAllMocks(); + await ctx.db.destroy(); +}); + +test('sync awaits the real callback and counts only completed physical scans', async () => { + const scan = Promise.withResolvers(); + vi.mocked(scanOnce).mockReturnValue(scan.promise); + const step = makeStep(); + let completed = false; + const running = runSyncCalendars(step).then((result) => { + completed = true; + return result; + }); + await Promise.resolve(); + expect(completed).toBe(false); + const metricsBeforeCompletion = await calendarSyncTotal.get(); + expect(metricsBeforeCompletion.values).toEqual([]); + expect(scanOnce).toHaveBeenCalledWith(ctx); + scan.resolve(); + const scanResult = await running; + expect(scanResult).toBe('requested'); + expect(step.configs).toEqual([{ name: 'scan-calendars' }]); + const metricsAfterCompletion = await calendarSyncTotal.get(); + expect(metricsAfterCompletion.values).toEqual([ + expect.objectContaining({ labels: { status: 'success' }, value: 1 }) + ]); +}); + +test('sync callback errors escape with failure metrics', async () => { + const error = new Error('local scan read failed'); + vi.mocked(scanOnce).mockRejectedValue(error); + await expect(runSyncCalendars(makeStep())).rejects.toBe(error); + const failureMetrics = await calendarSyncTotal.get(); + expect(failureMetrics.values).toEqual([ + expect.objectContaining({ labels: { status: 'failure' }, value: 1 }) + ]); +}); + +test('an opaque engine escape is not counted as a failed scan', async () => { + const error = new Error('opaque engine escape'); + const step = { run: vi.fn().mockRejectedValue(error) }; + await expect(runSyncCalendars(step)).rejects.toBe(error); + expect(scanOnce).not.toHaveBeenCalled(); + const metrics = await calendarSyncTotal.get(); + expect(metrics.values).toEqual([]); +}); + +test('refresh awaits the cycle with injectable options and native step defaults', async () => { + const cycle = Promise.withResolvers(); + vi.mocked(refreshCycle).mockReturnValue(cycle.promise); + const opts = { now: Temporal.Instant.from(at), via: 'refresh' as const }; + const step = makeStep(); + let completed = false; + const running = runRefreshCalendars(step, opts).then((result) => { + completed = true; + return result; + }); + await Promise.resolve(); + expect(completed).toBe(false); + expect(refreshCycle).toHaveBeenCalledWith(ctx, opts); + cycle.resolve(); + const refreshResult = await running; + expect(refreshResult).toBe('refreshed'); + expect(step.configs).toEqual([{ name: 'refresh-calendars' }]); +}); + +test('refresh local errors and engine escapes propagate unchanged', async () => { + const error = new Error('mirror write failed'); + vi.mocked(refreshCycle).mockRejectedValue(error); + await expect(runRefreshCalendars(makeStep())).rejects.toBe(error); + const engineError = new Error('opaque engine escape'); + await expect(runRefreshCalendars({ run: vi.fn().mockRejectedValue(engineError) })).rejects.toBe( + engineError + ); +}); + +test('expiry uses the injected clock, is idempotent, and logs the changed count', async () => { + const cases = [ + { id: 'stale', status: 'pending', start: '2026-04-01T11:00:00Z', expectedStatus: 'expired' }, + { id: 'boundary', status: 'pending', start: at, expectedStatus: 'expired' }, + { id: 'future', status: 'pending', start: '2026-04-01T13:00:00Z', expectedStatus: 'pending' }, + { + id: 'confirmed', + status: 'confirmed', + start: '2026-04-01T11:00:00Z', + expectedStatus: 'confirmed' + }, + { id: 'purged', status: 'purged', start: '2026-04-01T11:00:00Z', expectedStatus: 'purged' } + ] as const; + for (const { id, status, start } of cases) { + await ctx.db + .insertInto('appointments') + .values({ + ...sampleAppointment, + id, + event_type_id: id, + cancel_token: `token-${id}`, + status, + start_time: start, + action_log: '[]' + }) + .execute(); + } + const step = makeStep(); + const firstExpiryResult = await runExpirePendingAppointments(step, clock); + expect(firstExpiryResult).toBe('expired'); + const repeatedExpiryResult = await runExpirePendingAppointments(step, clock); + expect(repeatedExpiryResult).toBe('expired'); + for (const { id, expectedStatus } of cases) { + const row = await ctx.db + .selectFrom('appointments') + .selectAll() + .where('id', '=', id) + .executeTakeFirstOrThrow(); + expect(row.status).toBe(expectedStatus); + if (expectedStatus === 'expired') { + expect(parseActionLog(row.action_log)).toEqual([{ action: 'expire', actor: 'system', at }]); + } else { + expect(parseActionLog(row.action_log)).toEqual([]); + } + } + expect(ctx.logger.info).toHaveBeenCalledExactlyOnceWith( + { count: 2 }, + 'expired stale pending requests' + ); + expect(step.configs).toEqual([ + { name: 'expire-pending-appointments' }, + { name: 'expire-pending-appointments' } + ]); +}); + +test('expiry local SQL errors escape rather than completing the pass', async () => { + await ctx.db.schema.dropTable('appointments').execute(); + await expect(runExpirePendingAppointments(makeStep(), clock)).rejects.toThrow(); + expect(ctx.logger.info).not.toHaveBeenCalled(); +}); diff --git a/apps/worker/src/workflows/refresh-calendars.ts b/apps/worker/src/workflows/refresh-calendars.ts new file mode 100644 index 00000000..84820c67 --- /dev/null +++ b/apps/worker/src/workflows/refresh-calendars.ts @@ -0,0 +1,22 @@ +import { refreshCalendars, type RefreshCalendarsResult } from '@when/jobs'; +import { refreshCycle, type RefreshOptions } from '../calendar/refresh.js'; +import { getWorkerContext } from '../services/context.js'; +import { implementObservedWorkflow } from '../services/metrics.js'; + +interface RefreshStep { + run(config: { name: string }, fn: () => Promise): Promise; +} + +export async function runRefreshCalendars( + step: RefreshStep, + opts: RefreshOptions = {} +): Promise { + await step.run({ name: 'refresh-calendars' }, async () => { + await refreshCycle(getWorkerContext(), opts); + }); + return 'refreshed'; +} + +export function registerRefreshCalendarsWorkflow(): void { + implementObservedWorkflow(refreshCalendars, ({ step }) => runRefreshCalendars(step)); +} diff --git a/apps/worker/src/workflows/sync-calendars.ts b/apps/worker/src/workflows/sync-calendars.ts index b13a6790..f04e03f6 100644 --- a/apps/worker/src/workflows/sync-calendars.ts +++ b/apps/worker/src/workflows/sync-calendars.ts @@ -1,10 +1,26 @@ import { syncCalendars, type SyncCalendarsResult } from '@when/jobs'; -import type { CalendarSyncScanner } from '../calendar/sync-scanner.js'; -import { implementObservedWorkflow } from '../services/metrics.js'; +import { scanOnce } from '../calendar/sync.js'; +import { getWorkerContext } from '../services/context.js'; +import { calendarSyncTotal, implementObservedWorkflow } from '../services/metrics.js'; -export function registerSyncCalendarsWorkflow(scanner: CalendarSyncScanner): void { - implementObservedWorkflow(syncCalendars, (): SyncCalendarsResult => { - scanner.requestScan(); - return 'requested'; +interface SyncStep { + run(config: { name: string }, fn: () => Promise): Promise; +} + +export async function runSyncCalendars(step: SyncStep): Promise { + await step.run({ name: 'scan-calendars' }, async () => { + const ctx = getWorkerContext(); + try { + await scanOnce(ctx); + calendarSyncTotal.inc({ status: 'success' }); + } catch (error) { + calendarSyncTotal.inc({ status: 'failure' }); + throw error; + } }); + return 'requested'; +} + +export function registerSyncCalendarsWorkflow(): void { + implementObservedWorkflow(syncCalendars, ({ step }) => runSyncCalendars(step)); } diff --git a/packages/jobs/src/index.ts b/packages/jobs/src/index.ts index 4d7b04f6..c1bbf1cb 100644 --- a/packages/jobs/src/index.ts +++ b/packages/jobs/src/index.ts @@ -1,6 +1,8 @@ export { reconcileAppointment, syncCalendars, + refreshCalendars, + expirePendingAppointments, purgeAppointment, testEmail, testProvider, @@ -14,6 +16,10 @@ export type { ReconcileAppointmentResult, SyncCalendarsInput, SyncCalendarsResult, + RefreshCalendarsInput, + RefreshCalendarsResult, + ExpirePendingAppointmentsInput, + ExpirePendingAppointmentsResult, PurgeAppointmentRow, PurgeAppointmentInput, PurgeAppointmentResult, diff --git a/packages/jobs/src/specs.test.ts b/packages/jobs/src/specs.test.ts index 79c244cf..cabb580e 100644 --- a/packages/jobs/src/specs.test.ts +++ b/packages/jobs/src/specs.test.ts @@ -1,7 +1,13 @@ +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; import { expect, test } from 'vitest'; +import { initOpenWorkflow, closeOpenWorkflow } from './openworkflow.js'; import { reconcileAppointment, syncCalendars, + refreshCalendars, + expirePendingAppointments, testEmail, testProvider, listProviderCalendars, @@ -15,8 +21,51 @@ test('reconcileAppointment carries the shared workflow name', () => { expect(reconcileAppointment.name).toBe('reconcile-appointment'); }); -test('syncCalendars carries the shared workflow name', () => { - expect(syncCalendars.name).toBe('sync-calendars'); +test.each([ + [refreshCalendars, 'refresh-calendars', 'when:refresh-calendars', '*/5 * * * *'], + [ + expirePendingAppointments, + 'expire-pending-appointments', + 'when:expire-pending-appointments', + '0 * * * *' + ], + [syncCalendars, 'sync-calendars', 'when:sync-calendars', '*/10 * * * *'] +])('%s declares its stable payload-free UTC schedule', (spec, name, id, cron) => { + expect(spec.name).toBe(name); + expect(spec.version).toBeUndefined(); + expect(spec.schedules).toEqual([{ id, cron, timezone: 'UTC', input: {} }]); +}); + +test('producer initialization and maintenance enqueues do not persist schedules', async () => { + const directory = await mkdtemp(join(tmpdir(), 'when-producer-schedules-')); + try { + const client = initOpenWorkflow({ dbPath: join(directory, 'queue.sqlite') }); + const schedulesBeforeEnqueue = await client.schedules.list(); + expect(schedulesBeforeEnqueue.data).toEqual([]); + for (const spec of [refreshCalendars, expirePendingAppointments, syncCalendars]) { + const handle = await client.runWorkflow(spec, {}); + expect(handle.workflowRun).toMatchObject({ + workflowName: spec.name, + status: 'pending', + input: {}, + workflowScheduleId: null + }); + } + const schedulesAfterEnqueue = await client.schedules.list(); + expect(schedulesAfterEnqueue.data).toEqual([]); + } finally { + await closeOpenWorkflow(); + await rm(directory, { recursive: true, force: true }); + } +}); + +test('sync retains its workflow-level backstop', () => { + expect(syncCalendars.retryPolicy).toEqual({ + maximumAttempts: 1, + initialInterval: '1s', + backoffCoefficient: 2, + maximumInterval: '1s' + }); }); test('testEmail carries the shared workflow name', () => { diff --git a/packages/jobs/src/specs.ts b/packages/jobs/src/specs.ts index 5ff4ccbb..60855cfb 100644 --- a/packages/jobs/src/specs.ts +++ b/packages/jobs/src/specs.ts @@ -152,13 +152,36 @@ export const testCalendar: WorkflowSpec = } }); +export type RefreshCalendarsInput = Record; +export type RefreshCalendarsResult = 'refreshed'; + +export const refreshCalendars: WorkflowSpec = + defineWorkflowSpec({ + name: 'refresh-calendars', + schedules: [{ id: 'when:refresh-calendars', cron: '*/5 * * * *', timezone: 'UTC', input: {} }] + }); + +export type ExpirePendingAppointmentsInput = Record; +export type ExpirePendingAppointmentsResult = 'expired'; + +export const expirePendingAppointments: WorkflowSpec< + ExpirePendingAppointmentsInput, + ExpirePendingAppointmentsResult +> = defineWorkflowSpec({ + name: 'expire-pending-appointments', + schedules: [ + { id: 'when:expire-pending-appointments', cron: '0 * * * *', timezone: 'UTC', input: {} } + ] +}); + export type SyncCalendarsInput = Record; export type SyncCalendarsResult = 'requested'; -// Payload-free wake-up the worker handles by running its calendar sync. +// A requested result means the pass finished, not that every provider succeeded. export const syncCalendars: WorkflowSpec = defineWorkflowSpec({ name: 'sync-calendars', + schedules: [{ id: 'when:sync-calendars', cron: '*/10 * * * *', timezone: 'UTC', input: {} }], retryPolicy: { maximumAttempts: 1, initialInterval: '1s', -- 2.54.0 From b7dfc5aca2b6888187e5158fca8e04beab42c531 Mon Sep 17 00:00:00 2001 From: eleith Date: Tue, 20 Oct 2026 10:34:25 -0700 Subject: [PATCH 2/2] docs(worker): explain native maintenance and the drained grouped rollout --- .changeset/native-maintenance.md | 2 + apps/cli/src/utils/worker.test.ts | 77 +++++++++ .../web/src/lib/server/workflow/index.test.ts | 162 ++++++++++++++++-- apps/worker/README.md | 73 +++++--- docs/architecture.md | 59 +++++-- docs/deployment.md | 123 ++++++++++++- docs/development.md | 2 + packages/jobs/README.md | 66 ++++++- 8 files changed, 504 insertions(+), 60 deletions(-) create mode 100644 apps/cli/src/utils/worker.test.ts diff --git a/.changeset/native-maintenance.md b/.changeset/native-maintenance.md index 6e9852be..43e7f4e6 100644 --- a/.changeset/native-maintenance.md +++ b/.changeset/native-maintenance.md @@ -5,3 +5,5 @@ --- Run refresh, pending expiry and calendar repair as awaited native UTC cron workflows, with independent startup passes and pause-preserving schedule persistence. Remove application maintenance timers and coalescing; one execution slot serializes passes, so slow maintenance can delay other jobs. Stop the worker before closing queue and application databases. + +Document attempt-level email outcomes, retained personal data after purge failure, sensitive queue backups, and the drained grouped rollout with paired SQLite backup/restore checks. Deploy web, worker and CLI together only after all old work drains; rollback requires proven compatibility or explicit approval for paired restore and loss of later writes. diff --git a/apps/cli/src/utils/worker.test.ts b/apps/cli/src/utils/worker.test.ts new file mode 100644 index 00000000..60b66ede --- /dev/null +++ b/apps/cli/src/utils/worker.test.ts @@ -0,0 +1,77 @@ +import { mkdtemp, rm } from 'node:fs/promises'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; +import { DatabaseSync } from 'node:sqlite'; +import { afterEach, expect, test, vi } from 'vitest'; +import type { WhenConfiguration } from '@when/config'; +import * as jobs from '@when/jobs'; +import { runInWorker } from './worker.ts'; + +afterEach(() => vi.restoreAllMocks()); + +test('CLI initializes only a producer and gives each probe its own random enqueue key', async () => { + const directory = await mkdtemp(join(tmpdir(), 'when-cli-producer-')); + const dbPath = join(directory, 'queue.sqlite'); + const config = { + database: { queue: dbPath }, + url: { worker: 'http://worker.example.com:9000/' } + } as WhenConfiguration; + const client = jobs.initOpenWorkflow({ dbPath }); + const init = vi.spyOn(jobs, 'initOpenWorkflow'); + const newWorker = vi.spyOn(client, 'newWorker'); + const implement = vi.spyOn(client, 'implementWorkflow'); + const schedule = vi.spyOn(client, 'scheduleWorkflow'); + const result = vi.fn().mockResolvedValue('authenticated'); + const enqueue = client.runWorkflow.bind(client); + const run = vi.spyOn(client, 'runWorkflow').mockImplementation(async (...args) => { + const handle = await enqueue(...args); + vi.spyOn(handle, 'result').mockImplementation(result); + return handle; + }); + const fetch = vi.spyOn(globalThis, 'fetch').mockResolvedValue(new Response(null)); + const uuid = vi.spyOn(crypto, 'randomUUID'); + uuid.mockReturnValueOnce('00000000-0000-4000-8000-000000000001'); + uuid.mockReturnValueOnce('00000000-0000-4000-8000-000000000002'); + try { + for (let i = 1; i <= 2; i++) { + const probeResult = await runInWorker(config, jobs.testProvider, { name: 'work' }); + expect(probeResult).toEqual({ + ok: true, + value: 'authenticated' + }); + expect(run).toHaveBeenNthCalledWith( + i, + jobs.testProvider, + { name: 'work' }, + { idempotencyKey: `00000000-0000-4000-8000-00000000000${i}` } + ); + } + expect(jobs.testProvider.name).toBe('test-provider'); + expect(init).toHaveBeenCalledTimes(2); + expect(init).toHaveBeenCalledWith({ dbPath }); + expect(fetch).toHaveBeenCalledWith('http://worker.example.com:9000/healthz'); + expect(result).toHaveBeenCalledTimes(2); + expect(result).toHaveBeenCalledWith({ timeoutMs: 30_000 }); + expect(newWorker).not.toHaveBeenCalled(); + expect(implement).not.toHaveBeenCalled(); + expect(schedule).not.toHaveBeenCalled(); + const raw = new DatabaseSync(dbPath, { readOnly: true }); + try { + expect(raw.prepare('SELECT COUNT(*) AS count FROM workflow_schedules').get()).toEqual({ + count: 0 + }); + expect(raw.prepare('SELECT workflow_name, status, input FROM workflow_runs').all()).toEqual( + Array.from({ length: 2 }, () => ({ + workflow_name: 'test-provider', + status: 'pending', + input: '{"name":"work"}' + })) + ); + } finally { + raw.close(); + } + } finally { + await jobs.closeOpenWorkflow(); + await rm(directory, { recursive: true, force: true }); + } +}); diff --git a/apps/web/src/lib/server/workflow/index.test.ts b/apps/web/src/lib/server/workflow/index.test.ts index 7b2b6ed6..1e75d248 100644 --- a/apps/web/src/lib/server/workflow/index.test.ts +++ b/apps/web/src/lib/server/workflow/index.test.ts @@ -1,15 +1,29 @@ import { beforeEach, describe, expect, test, vi } from 'vitest'; import { openDb, runMigrations } from '@when/db'; +import { OpenWorkflow } from 'openworkflow'; +import { BackendSqlite } from 'openworkflow/sqlite'; -const runWorkflow = vi.fn(); - -vi.mock('@when/jobs', () => ({ - getOpenWorkflow: () => ({ runWorkflow }), - reconcileAppointment: { spec: { name: 'reconcile-appointment' } } +const { runWorkflow, getOpenWorkflow } = vi.hoisted(() => ({ + runWorkflow: vi.fn(), + getOpenWorkflow: vi.fn() })); -import { enqueueAppointmentReconciliation } from './index'; -import { reconcileAppointment } from '@when/jobs'; +vi.mock('@when/jobs', async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, getOpenWorkflow }; +}); + +import { + enqueueAppointmentReconciliation, + enqueueCalendarSync, + enqueuePurgeAppointment +} from './index'; +import { reconcileAppointment, syncCalendars, purgeAppointment } from '@when/jobs'; + +beforeEach(() => { + runWorkflow.mockReset(); + getOpenWorkflow.mockReturnValue({ runWorkflow }); +}); async function seed() { const db = openDb(':memory:'); @@ -33,9 +47,7 @@ async function seed() { } describe('enqueueAppointmentReconciliation', () => { - beforeEach(() => runWorkflow.mockReset()); - - test('snapshots the appointment and runs the workflow', async () => { + test('returns the appointment and enqueues the public workflow contract', async () => { const db = await seed(); const result = await enqueueAppointmentReconciliation(db, 'appt-1', 'confirmed'); @@ -48,6 +60,7 @@ describe('enqueueAppointmentReconciliation', () => { }, { idempotencyKey: 'appt-1:confirmed:0' } ); + expect(reconcileAppointment.name).toBe('reconcile-appointment'); expect(result.id).toBe('appt-1'); await db.destroy(); @@ -63,10 +76,133 @@ describe('enqueueAppointmentReconciliation', () => { await enqueueAppointmentReconciliation(db, 'appt-1', 'rescheduled-by-guest'); - expect(runWorkflow).toHaveBeenCalledWith(reconcileAppointment, expect.anything(), { - idempotencyKey: 'appt-1:rescheduled-by-guest:2' - }); + expect(runWorkflow).toHaveBeenCalledWith( + reconcileAppointment, + { appointmentId: 'appt-1', emailKind: 'rescheduled-by-guest' }, + { idempotencyKey: 'appt-1:rescheduled-by-guest:2' } + ); await db.destroy(); }); }); + +describe('unchanged producer inputs and keys', () => { + test('forwards the cleanup URL and keys email-free reconciliation by sequence', async () => { + const db = await seed(); + await db.updateTable('appointments').set({ ics_sequence: 3 }).execute(); + + await enqueueAppointmentReconciliation( + db, + 'appt-1', + undefined, + 'https://video.example.com/obsolete-room' + ); + + expect(runWorkflow).toHaveBeenCalledWith( + reconcileAppointment, + { + appointmentId: 'appt-1', + emailKind: undefined, + cleanupVideoChatUrl: 'https://video.example.com/obsolete-room' + }, + { idempotencyKey: 'appt-1:sync:3' } + ); + await db.destroy(); + }); + + test('sorts purge IDs for the key without changing row order or remote identifiers', async () => { + const rows = [ + { id: 'z', externalEventId: 'remote-z', externalCalendarId: 'work' }, + { id: 'a', externalEventId: null, externalCalendarId: null } + ]; + + await enqueuePurgeAppointment(rows); + await enqueuePurgeAppointment([...rows].reverse()); + + expect(purgeAppointment.name).toBe('purge-appointment'); + expect(runWorkflow).toHaveBeenNthCalledWith( + 1, + purgeAppointment, + { rows }, + { idempotencyKey: 'purge:a,z' } + ); + expect(runWorkflow).toHaveBeenNthCalledWith( + 2, + purgeAppointment, + { rows: [...rows].reverse() }, + { idempotencyKey: 'purge:a,z' } + ); + expect(rows.map((row) => row.id)).toEqual(['z', 'a']); + }); + + test('sync enqueues an empty input with a fresh random key and does not wait for the scan', async () => { + const result = vi.fn(); + runWorkflow.mockResolvedValue({ result }); + const uuid = vi.spyOn(crypto, 'randomUUID'); + uuid.mockReturnValueOnce('00000000-0000-4000-8000-000000000001'); + uuid.mockReturnValueOnce('00000000-0000-4000-8000-000000000002'); + try { + const firstEnqueueResult = await enqueueCalendarSync(); + expect(firstEnqueueResult).toBeUndefined(); + const secondEnqueueResult = await enqueueCalendarSync(); + expect(secondEnqueueResult).toBeUndefined(); + expect(syncCalendars.name).toBe('sync-calendars'); + for (let i = 1; i <= 2; i++) { + expect(runWorkflow).toHaveBeenNthCalledWith( + i, + syncCalendars, + {}, + { + idempotencyKey: `sync-calendars:00000000-0000-4000-8000-00000000000${i}` + } + ); + } + expect(result).not.toHaveBeenCalled(); + } finally { + uuid.mockRestore(); + } + }); + + test('an identical enqueue reuses terminal history within the engine dedup window', async () => { + const db = await seed(); + const backend = BackendSqlite.connect(':memory:'); + const client = new OpenWorkflow({ backend }); + const previousHandler = vi.fn(() => 'reconciled' as const); + const changedHandler = vi.fn(() => 'reconciled' as const); + client.implementWorkflow(reconcileAppointment, previousHandler); + getOpenWorkflow.mockReturnValue(client); + const enqueue = vi.spyOn(client, 'runWorkflow'); + const worker = client.newWorker({ concurrency: 1 }); + try { + await enqueueAppointmentReconciliation(db, 'appt-1', 'confirmed'); + const first = await enqueue.mock.results[0].value; + const processedRuns = await worker.tick(); + expect(processedRuns).toBe(1); + await worker.stop(); + const reconciliationResult = await first.result(); + expect(reconciliationResult).toBe('reconciled'); + + const restartedClient = new OpenWorkflow({ backend }); + restartedClient.implementWorkflow(reconcileAppointment, changedHandler); + getOpenWorkflow.mockReturnValue(restartedClient); + const repeatedEnqueue = vi.spyOn(restartedClient, 'runWorkflow'); + await enqueueAppointmentReconciliation(db, 'appt-1', 'confirmed'); + const repeated = await repeatedEnqueue.mock.results[0].value; + expect(repeated.workflowRun.id).toBe(first.workflowRun.id); + expect(repeated.workflowRun.status).toBe('completed'); + const restartedWorker = restartedClient.newWorker({ concurrency: 1 }); + try { + const restartedProcessedRuns = await restartedWorker.tick(); + expect(restartedProcessedRuns).toBe(0); + } finally { + await restartedWorker.stop(); + } + expect(previousHandler).toHaveBeenCalledTimes(1); + expect(changedHandler).not.toHaveBeenCalled(); + } finally { + await worker.stop(); + await backend.stop(); + await db.destroy(); + } + }); +}); diff --git a/apps/worker/README.md b/apps/worker/README.md index ad634bd0..307ee459 100644 --- a/apps/worker/README.md +++ b/apps/worker/README.md @@ -10,7 +10,9 @@ coordinates calendar/video updates, renders emails, sends them over SMTP, and records outcomes in the appointment action log and service status. It also handles purge, calendar-repair requests, video-link generation, and service probes. On boot it loads config, migrates the app DB, connects the queue, registers workflows, -starts polling with one execution slot, and serves `/healthz`. +starts polling with one execution slot and a 30-second lease, and serves `/healthz`. +Native `worker.start()` persists declared schedules before startup enqueues all three +maintenance passes with `{}` and no run options. ## Run it @@ -36,14 +38,14 @@ database: queue: ./data/openworkflow.sqlite # default ``` -Relative paths resolve against the config file's directory, so web and worker +Relative paths resolve against the config directory's parent (the deployment root), so web and worker (sharing one `when.yaml`) open the same files. ## Environment - `CONFIG_PATH` — path to `when.yaml` (default `/app/config/when.yaml` in prod, `/config/when.yaml` otherwise). -- `DATABASE_PATH` / `QUEUE_DB_PATH` — optional escape hatches that override the +- `WHEN_DATABASE_PATH` / `WHEN_QUEUE_DB_PATH` — optional escape hatches that override the config's `database.app` / `database.queue`. - `PORT` — health server port (default `9000`). @@ -63,25 +65,24 @@ attempts (30-second initial delay, doubling up to ten minutes). A successful sen is checkpointed separately from its status write. Host and guest keep independent retry budgets, even when they share an email address. -An email `failed` action means a provider attempt failed and retries may remain. -The parent records `done` only when every intended email was sent and the appointment -still exists and is not purged. Ineligible deliveries are skipped. Exhaustion fails -the engine run rather than returning a successful summary of failed delivery. - Purge deletes each remote event with up to thirty native attempts (one-minute -initial delay, doubling up to one hour). Success or an already-missing event permits -video cleanup and local deletion. Exhaustion fails the job, retaining that row and -all later rows in the chain. These rows are already marked `purged` and hidden from -normal scheduling; another purge click does not automatically recover them. Review -failed jobs before deciding how to finish cleanup. +initial delay, doubling up to one hour). Exhaustion fails the job and retains the +current and later hidden `purged` rows, including local personal data. Clicking +purge again is not promised recovery; cleanup requires explicit operator review. + +Email action logs describe attempts, not terminal job outcomes. See +[interpreting outcomes](../../docs/deployment.md#interpreting-outcomes) for email +log/service-status meanings and purge recovery guidance. Prepared envelopes include personal data and guest capability links: treat queue -backups like application backups. Checkpoints do not guarantee exactly-once SMTP; -a crash after remote acceptance but before saving success can still cause a duplicate. +backups like application backups. Effects are at-least-once: a crash after remote +acceptance but before saving success can still duplicate effects or action-log entries. +Metrics and logs are not exactly-once ledgers. -Before deploying these changed workflows, stop producers, let the old worker drain -unfinished jobs, then stop it and back up both databases before a grouped restart. -Do not replay unfinished old workflow histories through the changed step graph. +Deploy web, worker and CLI together only after draining old work. Follow the +[grouped rollout runbook](../../docs/deployment.md#grouped-workflow-rollout) for read-only +inspection, paired SQLite backups, restore verification and rollback restrictions. +Do not replay unfinished old workflow histories through changed step graphs. ## Engine integration @@ -94,6 +95,36 @@ escaped handler execution may mean durable suspension rather than a terminal job failure. Email metrics count physical SMTP attempt outcomes, not an exactly-once delivery ledger. -OpenWorkflow `0.11.0` supports persisted cron schedules. Refresh, expiry, and repair -still use the existing process timers; migrating those tasks to awaited cron -workflows is a separate application change. +## Maintenance and shutdown + +| Workflow | UTC cron | Stable schedule ID | Result | +| ----------------------------- | -------------- | ---------------------------------- | ----------- | +| `refresh-calendars` | `*/5 * * * *` | `when:refresh-calendars` | `refreshed` | +| `expire-pending-appointments` | `0 * * * *` | `when:expire-pending-appointments` | `expired` | +| `sync-calendars` | `*/10 * * * *` | `when:sync-calendars` | `requested` | + +All inputs are `{}`. Each handler awaits the actual pass in durable steps; `requested` +means the scan finished, not that an in-memory request was accepted. Refresh checks +per-calendar due intervals, keeps stale mirrors on provider fetch failure, and runs +before conflict checks. Expiry SQL is idempotent. Subsequent callbacks read live config. +Expected provider failures record status for later passes; local errors escape to native +retries. Completion is not a promise that every provider succeeded. + +Cron follows UTC wall-clock boundaries, not time since the last pass finished. After +downtime it dispatches one overdue run per schedule. Native startup upserts preserve +administrative pause state, but independent startup runs still execute while paused; +startup and overdue runs can coexist. See [schedule management](../../packages/jobs/README.md#schedule-management) +for the public list/pause/resume/delete API; removing declarations does not retire stored schedules. + +No application maintenance timers or coalescers remain. Every cron/manual request gets +its own pass. One slot serializes active work: slow calls block appointment jobs and +probes, and repeated scans can build a backlog. Native retry delays park and release +the slot. Ordinary function steps use the engine default of ten attempts, one-second +initial delay, doubling to a 100-second cap. Workflow backstops are not extra provider +budgets; claim counts include resumes, and native exhaustion is terminal. Bookkeeping +failures can consume step attempts too: these limits are not exact physical-call counts. + +Signals and config restarts share shutdown: stop config watching and HTTP admission, +await `worker.stop()`, finish HTTP closure, close the queue, then close the app DB. +Stop waits for active passes to finish or park, not for pending work/future retries; +it is not queue draining. Force-kill recovery relies on saved history and lease expiry. diff --git a/docs/architecture.md b/docs/architecture.md index 320cf944..2c10b8b2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -51,16 +51,37 @@ appointment request ─► web: write appointment row ─► enqueue job ─► Jobs run on openworkflow over `node:sqlite`. `packages/jobs` is the single source of truth for **what** jobs exist and their input/output shapes -(`packages/jobs/src/specs.ts`): the web app triggers a run from a spec -(\`runWorkflow(sendAppointmentEmail, …)\`), and the worker provides the implementation. Both -sides share types and resolve a workflow by name, so web never imports worker code. +(`packages/jobs/src/specs.ts`): the web app enqueues `reconcile-appointment` from its +spec, and the worker provides the implementation. Both sides share types and resolve a +workflow by name, so web never imports worker code. Web and CLI initialize producer-only +clients: they neither start a worker nor persist schedules. -- **Durable steps.** Each side-effect (one SMTP send) is a memoized `step.run(...)` with - its own retry policy, so a replay never re-sends something already sent. -- **Idempotency keys** dedupe enqueues within openworkflow's dedup window. Appointment emails - key on `appointmentId:kind:ics_sequence` — `ics_sequence` bumps on every reschedule, so - a repeat same-kind email isn't swallowed as a duplicate. Calendar sync uses a random - key per call (each appointment change should trigger a scan). +- **Durable steps.** Saved successes are reused on replay. Reconciliation checkpoints + prepared role-tagged email envelopes, then starts independent guest/host delivery + children with native retries. Native retry waits park and release the execution slot. + A crash after a remote effect but before its checkpoint can still duplicate effects + or logs: execution is at-least-once, not exactly-once. +- **Idempotency keys** dedupe enqueues for 24 hours, including matching terminal history; + they do not automatically rerun old jobs. Appointment emails key on + `appointmentId:kind:ics_sequence`; email-free reconciliation uses + `appointmentId:sync:ics_sequence`. Purge keys use sorted chain IDs. Calendar sync + uses a random key per call, and the producer awaits enqueueing only. +- **Native maintenance.** Worker `start()` persists declared UTC schedules, preserving + pause state: refresh every five minutes, pending expiry hourly, repair every ten + minutes. Startup then independently enqueues all three `{}` passes without options. + Cron is wall-clock aligned and catches up once per overdue schedule, not once per + missed tick. Pausing cron does not suppress startup passes or already queued work. +- **One execution slot.** Maintenance handlers await their actual pass; sync's + `requested` result means its scan finished. There are no application maintenance + timers or coalescers. Repeated requests each get a pass and can build a backlog; + slow active maintenance blocks appointment jobs and probes. Expected provider + failures are recorded for later passes; unexpected local errors escape to bounded + native retries. A completed pass does not mean every provider succeeded. + +Email action logs describe attempts; engine history determines terminal outcomes. +Failed purges retain hidden local personal data and require explicit operator review. +See [interpreting outcomes](deployment.md#interpreting-outcomes) for detailed semantics +and the [rollout runbook](deployment.md#grouped-workflow-rollout) for backup requirements. ## Calendar I/O (off the request path) @@ -92,8 +113,10 @@ delivery: `packages/db` wraps `node:sqlite` behind a small Kysely dialect (no native addons) and owns the schema types, migrations, and runner. Migrations run on boot and are idempotent. -Both web and worker open the same `when.sqlite`; the worker additionally owns the -openworkflow queue database. +Both web and worker open the same `when.sqlite`; web/CLI producers and the worker +also open the shared openworkflow queue database. Worker shutdown stops config watching and +HTTP admission, awaits worker stop and HTTP closure, then closes queue and app DBs. +Worker stop waits for active work to finish or park, not for pending jobs or future retries. ## Configuration system @@ -145,14 +168,14 @@ moment, not a zone. the `auth` section of `docs/config.md` for when it fits. - Secrets are injected via `${ENV_VAR}` interpolation in `when.yaml`, never committed. - Nothing in SQLite is encrypted at the column level. The database holds guest names, - emails and booking history in plain text, so the data directory deserves the same - handling as `.env` — and a backup that leaves the host should be encrypted as a whole, - which protects that data rather than one field of it. + emails and booking history in plain text. The queue also holds prepared email + envelopes, personal data, and guest capability links. Treat both databases and their + backups like `.env`; encrypt backups that leave the host. - Never log raw request bodies, session tokens, `cancel_token` values, or secrets. #### The guest capability model -Guests have no account, so their link *is* their credential — a `cancel_token` in the query +Guests have no account, so their link _is_ their credential — a `cancel_token` in the query string. A phone-only booking may not even have an email address to send a second factor to, so the goal is not to make a leak impossible. It is to keep the blast radius small, bounded in time, and reversible by the host. @@ -166,13 +189,13 @@ time, and reversible by the host. token for it. A stale link therefore shows that the meeting moved and stops there; the guest was sent the new link when it moved. - **Everything expires.** `APPOINTMENT_VIEW_GRACE_DAYS` (7) past the appointment's end, and the - same window gates the `.ics` download. A superseded row counts those 7 days from the *move*, + same window gates the `.ics` download. A superseded row counts those 7 days from the _move_, not from its old slot, which would otherwise still be months away. - **The host can revoke.** Rotating from the guest-link dialog mints a fresh token, so every link issued so far stops resolving. Offered only on the chain tip, enforced in SQL. No email is sent — re-delivery is the host's call, since the address may itself be the leak path. -What a leaked *live* link still permits: cancelling, or moving the appointment — which on a +What a leaked _live_ link still permits: cancelling, or moving the appointment — which on a `require_approval: true` meeting drops it back to `pending` and re-arms host approval. Both are written to the action log, both are visible to the host, and both are recoverable. That is accepted as proportionate. View-only tokens and emailed confirmation before destructive actions @@ -185,7 +208,7 @@ These are understood and deliberately not fixed. Recorded so a future reader doe them for oversights. - **`cancel_token` and bearer tokens compare with `!==`, not in constant time.** They are - full-entropy random values, which makes remote timing extraction impractical. The *password* + full-entropy random values, which makes remote timing extraction impractical. The _password_ comparison is constant-time, because a human-chosen secret is a different problem. - **`/public/[...file]` follows symlinks.** `resolve()` blocks `../`, but a symlink planted inside the public directory serves its target. That directory is operator-mounted, and an diff --git a/docs/deployment.md b/docs/deployment.md index 868e32b5..68b777ae 100644 --- a/docs/deployment.md +++ b/docs/deployment.md @@ -146,15 +146,132 @@ State is two SQLite files under the data directory the compose file mounts (`./data` → `/app/data`): - `when.sqlite` — appointments, mirrored busy times, and service status (shared by web + worker). -- `openworkflow.sqlite` — the background job queue (owned by the worker). +- `openworkflow.sqlite` — the background job queue (shared by web/CLI producers and worker). Branding overrides live in a separate `public/` directory mounted alongside (`./public` → `/app/public`, overridable with `WHEN_PUBLIC_DIR`); the web app serves them at `/public/...`. The bundled defaults under `/assets/*` are baked into the image and need no mount. -Back up the data directory (and `public/`, if you use it) to preserve appointments and -settings across restarts. Database paths are configurable under `database` in `when.yaml`. +Back up both databases consistently using SQLite-aware backup handling, not live file +copies that omit committed WAL data. Include `when.yaml`, its environment secrets, and +`public/` overrides separately. Database paths are configurable under `database`. +Both databases are sensitive: the queue includes prepared email envelopes, personal data +and guest capability links. Encrypt off-host backups and restrict local access. + +## Grouped workflow rollout + +Deploy web, worker and CLI as one stopped, grouped release. Do not deploy intermediate +PR images automatically; release/version/tag operations require separate authorization. +Application tests and the read-only drain/backup procedure are the gates, not an +assumption of old/new queue compatibility. + +1. **Verify the release.** Run affected jobs/worker/web/CLI checks, lint, coverage and + builds, workspace checks and browser regressions. Confirm both CI paths execute the + co-located integration tests. Keep the currently deployed artifacts available. +2. **Stop producers**, including web, CLI commands and any other enqueuers. Let the + **old worker** drain every unfinished old run. Worker shutdown alone is not draining: + future pending jobs, parked retries/parents, running jobs and legacy `sleeping` work + all count. Inspect the old queue read-only; do not initialize the new jobs client, + since connection runs migrations. If recurring dispatch exists, pause it with the + matching deployed API so the queue can drain. +3. **Halt if it cannot drain.** Require an explicit operator retry/cancel/reset decision + after reviewing local and remote state. Never automatically wipe/reset a queue, even + an apparently empty one, edit checkpoints, or resume changed old application graphs. + Inspect again after any authorized action; cancelling a parent need not cancel its + already-started children. +4. **Stop the old worker**, recheck that no unfinished runs remain, then take a consistent + paired app/queue backup while all writers stay stopped. Include committed WAL data + with SQLite-aware handling and verify restoration on copies (examples below). +5. **Deploy the grouped artifacts once**, against the same config and database paths, + with no old worker or competing maintenance timers. Native `worker.start()` persists + schedules and preserves pause; startup then enqueues all three `{}` passes without + options, independently of pause. Inspect startup passes, schedules, service status + and queue progress. Cron catches up once per overdue schedule and can coexist with + startup work. Repeated scans are not coalesced; slow active maintenance blocks other + jobs at concurrency one, while native retry waits park. Do not replay old graphs or + automatically rerun terminal history; dedup may return an existing terminal run. +6. **Rollback only with evidence.** Use a proven compatible binary, or obtain explicit + approval for restoring the paired backups and losing all later writes. Old binaries + are not assumed compatible with migrated queues. Stop every writer before a paired + restore; restore into clean destinations without stale WAL/SHM sidecars. Remote + calendar/email effects are not undone by restoring SQLite. + +### Read-only drain inspection + +Set `QUEUE` to the actual old queue path. With the SQLite CLI, inspect the schema first: + +```sh +sqlite3 -readonly "$QUEUE" '.schema workflow_runs' +``` + +The installed `0.11.0` schema uses `workflow_runs`, with `namespace_id`, `id`, +`workflow_name`, `status`, and `available_at`. Verify these columns against the deployed +schema; a mismatch requires investigation, not opening it with a newer client. Installed +public status types/terminal logic recognize `completed`, legacy `succeeded`, `failed`, +and **`canceled`** (one l) as terminal. `pending`, `running`, and legacy `sleeping` are +unfinished. Unknown statuses halt the rollout, even if they look terminal. + +```sh +sqlite3 -readonly -header -column "$QUEUE" <<'SQL' +PRAGMA query_only = ON; +SELECT status, COUNT(*) AS runs FROM workflow_runs GROUP BY status; +SELECT namespace_id, id, workflow_name, status, available_at +FROM workflow_runs +WHERE status NOT IN ('completed', 'succeeded', 'failed', 'canceled') +ORDER BY namespace_id, created_at, id; +SQL +``` + +Proceed only when the second query returns no rows and every observed status is known. +There is deliberately **no due-time filter**: future pending work is unfinished. Inspect +all namespaces and child runs, not just top-level appointment runs. Do not dump inputs, +envelopes, errors or capability links into shared logs. + +### Paired backup and restore check + +With producers and worker stopped, set `APP`, `QUEUE`, and `BACKUP` to the real paths +and a fresh restricted backup directory. For ordinary paths without single quotes: + +```sh +mkdir -m 700 "$BACKUP" +sqlite3 -readonly "$APP" ".backup '$BACKUP/when.sqlite'" +sqlite3 -readonly "$QUEUE" ".backup '$BACKUP/openworkflow.sqlite'" +mkdir -m 700 "$BACKUP/restore-check" +cp "$BACKUP/when.sqlite" "$BACKUP/restore-check/when.sqlite" +cp "$BACKUP/openworkflow.sqlite" "$BACKUP/restore-check/openworkflow.sqlite" +sqlite3 -readonly "$BACKUP/restore-check/when.sqlite" 'PRAGMA integrity_check;' +sqlite3 -readonly "$BACKUP/restore-check/openworkflow.sqlite" 'PRAGMA integrity_check;' +``` + +Both integrity checks must report `ok`. Compare restored appointment counts and queue +status counts to the stopped sources, and rerun drain inspection on the restored queue. +SQLite `.backup` includes committed WAL state; do not copy live main files alone or +remove source WAL files to make backups look clean. Because every writer is stopped, +the sequential backups form a consistent pair. Keep that pair together; test copies +must never reach real providers or become the running deployment. + +### Interpreting outcomes + +An email action `failed` is an **observed provider attempt**, possibly followed by retries +and recovery; local errors are not provider-failure observations. `done` requires a +nonempty intended set where every delivery reported `sent` and the row is still eligible +at the final write. Mixed/all skipped sets and missing/purged rows get no summary. +Engine history is authoritative for terminal failure; no aggregate terminal email-failure +summary is synthesized. An already-started sibling can continue independently. SMTP +service status is the **last recorded physical attempt outcome**, not an aggregate +recipient/job result. Handler metrics include replays/suspensions; email metrics count +physical attempt outcomes. Crashes before checkpoints can duplicate effects or log +entries: this is at-least-once execution, not an exactly-once ledger. + +A failed purge retains the current and later hidden `purged` rows, including local +personal data. Earlier successful deletions remain deleted. Web selection skips +already-purged rows, so clicking purge again is not promised recovery. Review retained +rows, engine history and remote state before choosing explicit operator cleanup; there +is no automatic terminal rerun or remote-orphan fallback. A completed maintenance pass +likewise does not guarantee every provider succeeded. See the +[worker README](../apps/worker/README.md#maintenance-and-shutdown) for cadence and the +[jobs README](../packages/jobs/README.md#schedule-management) for public schedule management. ## Config reloads diff --git a/docs/development.md b/docs/development.md index 659ebfe7..ec4f3a31 100644 --- a/docs/development.md +++ b/docs/development.md @@ -105,6 +105,8 @@ suite in both pipelines. - **Type safety at boundaries:** validate every external input (YAML, forms, API) against the config JSON Schema (ajv) — the repo does not use Zod. Database access goes through Kysely. +- **Async readability:** assign awaited results to named variables before returning + them or passing them as arguments. - **Time:** never call `new Date()` / `Date.now()` in domain logic; inject a `Clock`. - **Comments:** comment non-obvious rationale only. Don't restate what a name already says. diff --git a/packages/jobs/README.md b/packages/jobs/README.md index de1750fd..6176a00b 100644 --- a/packages/jobs/README.md +++ b/packages/jobs/README.md @@ -12,7 +12,10 @@ recurring schedules. - `reconcileAppointment` — coordinate appointment calendar/video updates and email. - `purgeAppointment` — clean up purged appointment chains. -- `syncCalendars` — request the worker's calendar repair scan. +- `syncCalendars` — `{}` → `requested`, after the actual calendar repair scan finishes. +- `refreshCalendars` — `{}` → `refreshed`, after refresh and conflict checks. +- `expirePendingAppointments` — `{}` → `expired`, after idempotent expiry SQL. + Completed maintenance passes can include recorded provider failures. - `testEmail`, `testProvider`, `listProviderCalendars`, `testCalendar`, `generateVideoChat` — on-demand worker operations. - `initOpenWorkflow({ dbPath, logger? })` — open and migrate the queue database and @@ -28,10 +31,63 @@ recurring schedules. Specs and their input/output types live in `src/specs.ts`. The `./specs` entrypoint exports contracts without the process-wide client helpers. -Queue migrations run on connection, including producer connections. Existing step -results survive the engine upgrade, but that does not make a changed application -workflow graph safe to replay: drain old appointment work before deploying changed -handlers. Back up both SQLite databases consistently before an engine upgrade. +Queue migrations run on connection, including producer connections. Do not start a +new producer against the old deployment during cutover. Saved results do not make +changed application graphs safe to replay. Follow the +[grouped rollout runbook](../../docs/deployment.md#grouped-workflow-rollout): drain all +unfinished old work before deployment, then stop and back up both SQLite databases +consistently. Never automatically rerun terminal history. Reusing an enqueue key +can return an existing terminal run within the engine's 24-hour dedup window. + +Prepared envelopes contain personal data and guest capability links; queue files and +backups are sensitive. Remote effects and logs can duplicate after a crash before +checkpointing, so neither durable steps nor enqueue keys provide exactly-once effects. + +## Schedule management + +Only native worker `start()` persists the declarations from registered implementations. +Startup awaits it, then enqueues refresh, expiry and sync independently with `{}` and +no run options. Their UTC schedules are `*/5 * * * *`, `0 * * * *`, and `*/10 * * * *`, +with IDs `when:refresh-calendars`, `when:expire-pending-appointments`, and +`when:sync-calendars`. Upserts preserve pause; startup runs do not depend on cron's pause +state. Cron is wall-clock aligned and catches up once per overdue schedule, not per +missed tick. No application timers or coalescers supplement it. + +These are the installed `0.11.0` public APIs, not `when-cli` commands. Given an initialized +client connected to the intended queue, list every page: + +```ts +const client = getOpenWorkflow(); +let after: string | undefined; +do { + const page = await client.schedules.list({ limit: 100, after }); + console.table( + page.data.map(({ id, cron, timezone, nextRunAt }) => ({ + id, + cron, + timezone, + nextRunAt + })) + ); + after = page.pagination.next ?? undefined; +} while (after); +``` + +Manage one schedule explicitly (the following are separate operator actions): + +```ts +await client.schedules.pause('when:sync-calendars'); +await client.schedules.resume('when:sync-calendars'); +await client.schedules.delete('when:sync-calendars'); +``` + +`nextRunAt === null` means paused; resume computes the next future occurrence. Pause +and delete do not cancel already queued runs or suppress independent startup runs. +Removing a declaration alone leaves its persisted schedule active. To retire a target, +stop the worker, delete its schedule through the public API, and remove the declaration +before restarting; otherwise native startup recreates it. Schedule-management connections +also migrate queues, so use the matching deployed version, not a new client for old-queue +read-only inspection. ## Scripts -- 2.54.0