diff --git a/.changeset/time-relative-dispatch-ledger.md b/.changeset/time-relative-dispatch-ledger.md new file mode 100644 index 0000000000..b8e6008c89 --- /dev/null +++ b/.changeset/time-relative-dispatch-ledger.md @@ -0,0 +1,29 @@ +--- +'@objectstack/service-automation': minor +'@objectstack/trigger-schedule': minor +'@objectstack/spec': patch +--- + +Time-relative sweeps are now idempotent per matched window (#10220). Previously the sweep +held no cross-tick memory, so every re-scan of the same window re-dispatched the same +records — a 5s-interval flow minted 15 duplicate reminders in ~70s, and even under a daily +cron a kernel rebuild re-dispatched the day's window. + +- `@objectstack/service-automation` — new platform object `sys_flow_dispatch`: a persisted + dispatch-claim ledger (ADR-0057 telemetry retention, 30 days), registered alongside + `sys_automation_run` and exposed as `AutomationEngine.claim(key): Promise` on + the automation service surface (check-and-record; a concurrent duplicate insert re-reads + and reports the key as already claimed). When no ObjectQL engine / registration is + available the engine degrades to in-process dedup and logs the weakened guarantee once; + when the ledger errors, the claim falls back to the in-process check for that key so a + store outage never blocks a dispatch (availability over strict-once). +- `@objectstack/trigger-schedule` — the time-relative sweep computes a dispatch key from + the MATCHED WINDOW's identity and claims it before launching: offset mode keys on + `(flowName, recordId, windowDay, offset)` — so a dateField edit that moves the window + legitimately re-fires — and range mode keys on `(flowName, recordId, sweepDay, + rangeSpec)`, preserving the documented `withinDays` semantic ("fires every day the + record stays in range") while never firing twice in one day. The trigger resolves the + claim surface structurally from the automation service; without one it dedups + in-process and warns once. +- `@objectstack/spec` — `sys_flow_dispatch` added to `PLATFORM_OBJECTS_BY_PACKAGE` under + `service-automation` (registry conformance). diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index 35d3e3f51e..7dfb60c501 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -1133,6 +1133,27 @@ export interface SuspendedRunStore { loadTerminal?(runId: string): Promise; } +/** + * Persisted claim ledger for trigger dispatch idempotency (#10220). + * + * `claim(key)` is check-and-record: `true` means the caller now owns this + * dispatch key and should launch the flow; `false` means some earlier sweep — + * possibly in a previous process lifetime — already dispatched it. Backed by + * `sys_flow_dispatch` in production (see `ObjectStoreFlowDispatchStore`), so + * dedup survives kernel rebuild. + */ +export interface FlowDispatchStore { + claim(key: string): Promise; +} + +/** + * TTL for the engine's IN-PROCESS dispatch-claim fallback (#10220). Every + * dispatch key embeds a calendar day, so a key stops being producible once its + * sweep day has passed; 48h comfortably outlives any key's claimable lifetime + * while keeping the fallback map bounded. + */ +export const IN_PROCESS_DISPATCH_CLAIM_TTL_MS = 48 * 60 * 60 * 1000; + /** * Lift the `{ dialect, source }` envelopes the flow schema derives for edge * `condition`s back onto the conversion output — and take nothing else with @@ -1304,6 +1325,25 @@ export class AutomationEngine implements IAutomationService { * duplicate `resume(runId)` can't re-enter and double-run side effects. */ private resuming = new Set(); + /** + * Optional persisted dispatch-claim ledger (#10220). When set, `claim()` + * checks-and-records against `sys_flow_dispatch` so trigger dispatch dedup + * survives kernel rebuild; when absent, `claim()` degrades to the + * in-process map below — honestly, with a one-time warning. + */ + private flowDispatchStore: FlowDispatchStore | null = null; + /** + * In-process dispatch-claim fallback: key → claim time (epoch ms). Used + * when no persisted ledger is attached, and per-key when the ledger + * errors. Entries expire after {@link IN_PROCESS_DISPATCH_CLAIM_TTL_MS}. + */ + private readonly inProcessDispatchClaims = new Map(); + /** + * Whether this engine has already said its dispatch dedup is in-process + * only (#10220). Once per instance: a silent fallback hides a permanently + * weakened guarantee, but repeating it every sweep tick is log spam. + */ + private dispatchClaimDegradationWarned = false; constructor(logger: Logger, store?: SuspendedRunStore, options?: AutomationEngineOptions) { this.logger = logger; @@ -1322,6 +1362,71 @@ export class AutomationEngine implements IAutomationService { this.store = store; } + /** + * Attach (or replace) the persisted {@link FlowDispatchStore} (#10220). + * Used by the service plugin once the ObjectQL engine is available and + * `sys_flow_dispatch` is registered. + */ + setFlowDispatchStore(store: FlowDispatchStore): void { + this.flowDispatchStore = store; + } + + /** + * Claim a trigger dispatch key (#10220): `true` = the caller owns this + * dispatch and should launch the flow; `false` = it was already dispatched + * (this sweep, an earlier sweep, or a previous process lifetime). + * + * Exposed on the automation service surface so triggers — which resolve + * `automation` structurally and never learn the table name — can dedup + * their dispatches against the persisted `sys_flow_dispatch` ledger. + * + * Degradation contract (both halves deliberate): + * - No persisted ledger attached → in-process dedup, with a ONE-TIME + * warning that the guarantee is weakened (a rebuild can re-dispatch). + * - Persisted ledger ERRORS → availability over strict-once: the failure + * is logged and the claim falls back to the in-process check for that + * key, which returns `false` only when THIS process already dispatched + * it — so a store outage never blocks a dispatch, and never double-fires + * within one process lifetime either. + */ + async claim(key: string): Promise { + if (this.flowDispatchStore) { + try { + return await this.flowDispatchStore.claim(key); + } catch (err) { + this.logger.warn( + `[automation] flow-dispatch claim '${key}' failed against the persisted ledger — ` + + `falling back to in-process dedup for this key (availability over strict-once: ` + + `the dispatch proceeds unless this process already made it; a kernel rebuild may re-dispatch it). ` + + `The store failure is in this record's meta.`, + describeThrownForLog(err), + ); + return this.claimInProcess(key); + } + } + if (!this.dispatchClaimDegradationWarned) { + this.dispatchClaimDegradationWarned = true; + this.logger.warn( + '[automation] no persisted flow-dispatch ledger (no ObjectQL engine, or sys_flow_dispatch not registered) — ' + + 'trigger dispatch dedup is IN-PROCESS ONLY and will NOT survive a kernel rebuild: ' + + 'the same record/window can be re-dispatched after a restart.', + ); + } + return this.claimInProcess(key); + } + + /** In-process half of {@link claim}: TTL-pruned check-and-record. */ + private claimInProcess(key: string): boolean { + const now = Date.now(); + const cutoff = now - IN_PROCESS_DISPATCH_CLAIM_TTL_MS; + for (const [k, t] of this.inProcessDispatchClaims) { + if (t < cutoff) this.inProcessDispatchClaims.delete(k); + } + if (this.inProcessDispatchClaims.has(key)) return false; + this.inProcessDispatchClaims.set(key, now); + return true; + } + /** * Generate a process-unique run id. Includes a random component so ids do * not collide with runs persisted by a previous process lifetime (a plain diff --git a/packages/services/service-automation/src/flow-dispatch-store.ts b/packages/services/service-automation/src/flow-dispatch-store.ts new file mode 100644 index 0000000000..6f0873701f --- /dev/null +++ b/packages/services/service-automation/src/flow-dispatch-store.ts @@ -0,0 +1,97 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +import type { FlowDispatchStore } from './engine.js'; + +/** + * Durable claim ledger for trigger dispatch idempotency (#10220). + * + * A {@link FlowDispatchStore} answers exactly one question, atomically enough + * for a sweep: "has this dispatch key been claimed before?" — recording the + * claim in the same call. The time-relative trigger computes a key from the + * matched window's identity and calls `claim()` before launching the flow; a + * `false` means some earlier sweep (possibly in a previous process lifetime) + * already dispatched this exact (flow, record, window). + * + * Two implementations: + * - {@link InMemoryFlowDispatchStore} — a Set (tests / explicit + * `suspendedRunStore: 'memory'` hosts). Sharable across two engine + * instances to simulate a kernel rebuild against one surviving ledger. + * - {@link ObjectStoreFlowDispatchStore} — persists to `sys_flow_dispatch` + * via the ObjectQL engine, so dedup survives kernel rebuild (the #10220 + * fix requirement the in-process Set cannot meet). + */ + +const TABLE = 'sys_flow_dispatch'; +const SYSTEM_CTX = { isSystem: true, positions: [], permissions: [] } as const; + +/** + * The exact ObjectQL slice `claim()` needs: a keyed read and an insert. + * Narrower than `SuspendedRunStoreEngine` on purpose — the ledger never + * updates or deletes (rows are immutable claims; the platform Reaper owns + * deletion via the object's declared retention), and demanding only what is + * used keeps every test double honest about that. + */ +export interface FlowDispatchStoreEngine { + find(object: string, options?: any): Promise; + insert(object: string, data: any, options?: any): Promise; +} + +/** In-memory {@link FlowDispatchStore} — process-lifetime dedup only. */ +export class InMemoryFlowDispatchStore implements FlowDispatchStore { + private readonly keys = new Set(); + + async claim(key: string): Promise { + if (this.keys.has(key)) return false; + this.keys.add(key); + return true; + } +} + +/** + * Durable {@link FlowDispatchStore} backed by the `sys_flow_dispatch` object. + * + * `claim()` is check-and-record: read the key's row, insert it when absent. + * The key is the row's primary `id`, so a concurrent duplicate insert (two + * sweeps racing the same key) fails on the id — the loser re-reads and reports + * the key as already claimed instead of surfacing a store error. All access + * uses a system context: these are infrastructure rows, not tenant data. + */ +export class ObjectStoreFlowDispatchStore implements FlowDispatchStore { + constructor(private readonly engine: FlowDispatchStoreEngine) {} + + async claim(key: string): Promise { + const existing = await this.engine.find(TABLE, { + where: { id: key }, limit: 1, context: SYSTEM_CTX, + }); + if (Array.isArray(existing) && existing[0]) return false; + const now = new Date().toISOString(); + try { + await this.engine.insert( + TABLE, + { id: key, dispatched_at: now, created_at: now }, + { context: SYSTEM_CTX }, + ); + return true; + } catch (err) { + // The insert may have lost a race with a concurrent claimer (duplicate + // primary key). Re-read before treating this as a store failure: a row + // present now means the key IS claimed — by someone else — which is a + // correct `false`, not an error. + const again = await this.engine.find(TABLE, { + where: { id: key }, limit: 1, context: SYSTEM_CTX, + }); + if (Array.isArray(again) && again[0]) return false; + throw err; + } + } + + /** + * Read the backing table once so a misconfiguration surfaces at BOOT rather + * than as a per-claim failure at sweep time. Throws the driver error + * verbatim — `no such table: sys_flow_dispatch` means the object was never + * registered (or its schema never synced). + */ + async probe(): Promise { + await this.engine.find(TABLE, { where: {}, limit: 1, context: SYSTEM_CTX }); + } +} diff --git a/packages/services/service-automation/src/flow-dispatch.test.ts b/packages/services/service-automation/src/flow-dispatch.test.ts new file mode 100644 index 0000000000..23b48af6b6 --- /dev/null +++ b/packages/services/service-automation/src/flow-dispatch.test.ts @@ -0,0 +1,151 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. +// +// Trigger dispatch idempotency (#10220): the persisted `sys_flow_dispatch` +// claim ledger and the `AutomationEngine.claim()` surface triggers consume. + +import { describe, it, expect, vi } from 'vitest'; +import { AutomationEngine } from './engine.js'; +import { InMemoryFlowDispatchStore, ObjectStoreFlowDispatchStore } from './flow-dispatch-store.js'; +import type { FlowDispatchStoreEngine } from './flow-dispatch-store.js'; + +function testLogger() { + const warn = vi.fn(); + const logger = { + info: () => {}, + warn, + error: () => {}, + debug: () => {}, + child: () => logger, + } as any; + return { logger, warn }; +} + +/** + * Fake ObjectQL slice with a real primary-key uniqueness on `id`. The `where` + * handling REFUSES any shape other than the keyed-by-id read the store makes — + * an unsupported filter must fail the test, never silently match everything. + */ +function fakeQl() { + const rows = new Map>(); + const engine: FlowDispatchStoreEngine = { + async find(_table, options) { + const where = (options?.where ?? {}) as Record; + const keys = Object.keys(where); + if (keys.length === 0) return [...rows.values()]; + if (keys.length !== 1 || keys[0] !== 'id' || typeof where.id !== 'string') { + throw new Error(`fake driver: unsupported where shape ${JSON.stringify(where)}`); + } + const row = rows.get(where.id); + return row ? [row] : []; + }, + async insert(_table, data) { + const id = (data as { id: string }).id; + if (rows.has(id)) throw new Error('UNIQUE constraint failed: sys_flow_dispatch.id'); + rows.set(id, data as Record); + return data; + }, + }; + return { engine, rows }; +} + +describe('ObjectStoreFlowDispatchStore', () => { + it('claim() is check-and-record: first true (row written), repeat false', async () => { + const { engine, rows } = fakeQl(); + const store = new ObjectStoreFlowDispatchStore(engine); + + await expect(store.claim('time-relative:f:2026-07-25:offset7:c1')).resolves.toBe(true); + const row = rows.get('time-relative:f:2026-07-25:offset7:c1'); + expect(row).toBeDefined(); + expect(row?.dispatched_at).toEqual(expect.any(String)); + + await expect(store.claim('time-relative:f:2026-07-25:offset7:c1')).resolves.toBe(false); + expect(rows.size).toBe(1); + }); + + it('distinct keys claim independently', async () => { + const { engine } = fakeQl(); + const store = new ObjectStoreFlowDispatchStore(engine); + await expect(store.claim('k1')).resolves.toBe(true); + await expect(store.claim('k2')).resolves.toBe(true); + }); + + it('an insert lost to a concurrent claimer reads as false, not as a store error', async () => { + // First find sees no row; the insert then collides (a racing sweep won); + // the re-check finds the winner's row → the key IS claimed. + let finds = 0; + const engine: FlowDispatchStoreEngine = { + async find() { + finds++; + return finds === 1 ? [] : [{ id: 'k1' }]; + }, + async insert() { + throw new Error('UNIQUE constraint failed: sys_flow_dispatch.id'); + }, + }; + const store = new ObjectStoreFlowDispatchStore(engine); + await expect(store.claim('k1')).resolves.toBe(false); + }); + + it('a genuine store failure propagates (the engine decides the fallback)', async () => { + const engine: FlowDispatchStoreEngine = { + async find() { return []; }, + async insert() { throw new Error('no such table: sys_flow_dispatch'); }, + }; + const store = new ObjectStoreFlowDispatchStore(engine); + await expect(store.claim('k1')).rejects.toThrow('no such table'); + }); +}); + +describe('AutomationEngine.claim (#10220)', () => { + it('uses the persisted ledger when attached — dedup survives a "rebuild" (new engine, same store)', async () => { + const store = new InMemoryFlowDispatchStore(); + const a = testLogger(); + const engineA = new AutomationEngine(a.logger); + engineA.setFlowDispatchStore(store); + + await expect(engineA.claim('k1')).resolves.toBe(true); + await expect(engineA.claim('k1')).resolves.toBe(false); + + // Kernel rebuild: a FRESH engine instance over the same surviving store. + const b = testLogger(); + const engineB = new AutomationEngine(b.logger); + engineB.setFlowDispatchStore(store); + await expect(engineB.claim('k1')).resolves.toBe(false); + await expect(engineB.claim('k2')).resolves.toBe(true); + + // No degradation warning on the healthy path. + expect(a.warn).not.toHaveBeenCalled(); + expect(b.warn).not.toHaveBeenCalled(); + }); + + it('no ledger attached: in-process dedup, degradation warned exactly once', async () => { + const { logger, warn } = testLogger(); + const engine = new AutomationEngine(logger); + + await expect(engine.claim('k1')).resolves.toBe(true); + await expect(engine.claim('k1')).resolves.toBe(false); + await expect(engine.claim('k2')).resolves.toBe(true); + + const degradations = warn.mock.calls.filter( + (c) => typeof c[0] === 'string' && (c[0] as string).includes('IN-PROCESS ONLY'), + ); + expect(degradations).toHaveLength(1); + }); + + it('a ledger ERROR never blocks the claim: falls back to in-process for that key (availability over strict-once)', async () => { + const { logger, warn } = testLogger(); + const engine = new AutomationEngine(logger); + engine.setFlowDispatchStore({ + async claim() { throw new Error('ledger unreachable'); }, + }); + + // First claim: store fails → in-process has no record → dispatch allowed. + await expect(engine.claim('k1')).resolves.toBe(true); + // Second claim of the SAME key in the same process: still deduped. + await expect(engine.claim('k1')).resolves.toBe(false); + expect(warn).toHaveBeenCalledWith( + expect.stringContaining('falling back to in-process dedup'), + expect.anything(), + ); + }); +}); diff --git a/packages/services/service-automation/src/index.ts b/packages/services/service-automation/src/index.ts index 70a08f24c9..43d4a20bc9 100644 --- a/packages/services/service-automation/src/index.ts +++ b/packages/services/service-automation/src/index.ts @@ -18,6 +18,7 @@ export type { RegisteredConnector, SuspendedRun, SuspendedRunStore, + FlowDispatchStore, RunRecord, StepLogEntry, UnknownNodeTypeAuditEntry, @@ -63,6 +64,13 @@ export type { SuspendedRunStoreEngine, ObjectStoreSuspendedRunStoreOptions } fro // AutomationServicePlugin and exported for hosts wiring a custom store. export { SysAutomationRun } from './sys-automation-run.object.js'; +// Trigger dispatch idempotency (#10220). The persisted claim ledger behind +// `AutomationEngine.claim(key)` — the in-memory store is for tests / explicit +// memory-only hosts; the ObjectQL-backed store makes dedup survive rebuilds. +export { InMemoryFlowDispatchStore, ObjectStoreFlowDispatchStore } from './flow-dispatch-store.js'; +export type { FlowDispatchStoreEngine } from './flow-dispatch-store.js'; +export { SysFlowDispatch } from './sys-flow-dispatch.object.js'; + // Kernel plugin — seeds all built-in nodes; this is the only plugin needed for // a fully-functional automation capability. export { AutomationServicePlugin, createPackageFileLoader } from './plugin.js'; diff --git a/packages/services/service-automation/src/plugin.ts b/packages/services/service-automation/src/plugin.ts index 6086985084..516d07002a 100644 --- a/packages/services/service-automation/src/plugin.ts +++ b/packages/services/service-automation/src/plugin.ts @@ -18,11 +18,13 @@ import { describeThrownForLog, thrownMessageText } from './thrown-cause-diagnost import { installBuiltinNodes, rearmSuspendedWaitTimers } from './builtin/index.js'; import { resolveRunDataContext } from './runtime-identity.js'; import { SysAutomationRun } from './sys-automation-run.object.js'; +import { SysFlowDispatch } from './sys-flow-dispatch.object.js'; import { ObjectStoreSuspendedRunStore, DEFAULT_MAX_TERMINAL_RUNS_PER_FLOW, type SuspendedRunStoreEngine, } from './suspended-run-store.js'; +import { ObjectStoreFlowDispatchStore } from './flow-dispatch-store.js'; /** * #1928 — normalize an ObjectQL object's `fields` (a name-keyed map, or an @@ -521,8 +523,9 @@ export class AutomationServicePlugin implements Plugin { } /** - * Register {@link SysAutomationRun} with the `manifest` service so the - * suspended-run table migrates like every other `sys_*` object (ADR-0019). + * Register {@link SysAutomationRun} and {@link SysFlowDispatch} with the + * `manifest` service so the suspended-run and dispatch-ledger tables + * migrate like every other `sys_*` object (ADR-0019, #10220). * * Returns whether it landed. Callers must honour a `false` — a durable * store attached over an unregistered object writes to a table that does @@ -538,7 +541,7 @@ export class AutomationServicePlugin implements Plugin { scope: 'system', defaultDatasource: 'cloud', namespace: 'sys', - objects: [SysAutomationRun], + objects: [SysAutomationRun, SysFlowDispatch], }); return true; } catch (err) { @@ -708,6 +711,14 @@ export class AutomationServicePlugin implements Plugin { durableStore = candidate; this.engine.setSuspendedRunStore(durableStore); ctx.logger.info('[Automation] Suspended-run persistence enabled (sys_automation_run)'); + // #10220 — persisted dispatch-claim ledger. Attached under + // the same guard as the durable run store: it needs the same + // engine surface, and its object rode the same manifest + // registration, so `runObjectRegistered` vouches for both + // tables. Without it the engine's claim() degrades to + // in-process dedup and says so once. + this.engine.setFlowDispatchStore(new ObjectStoreFlowDispatchStore(dataEngine)); + ctx.logger.info('[Automation] Flow-dispatch idempotency ledger enabled (sys_flow_dispatch)'); } } else { ctx.logger.info('[Automation] No ObjectQL engine — suspended runs kept in-memory only'); diff --git a/packages/services/service-automation/src/sys-flow-dispatch.object.ts b/packages/services/service-automation/src/sys-flow-dispatch.object.ts new file mode 100644 index 0000000000..d0b342a765 --- /dev/null +++ b/packages/services/service-automation/src/sys-flow-dispatch.object.ts @@ -0,0 +1,90 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +import { ObjectSchema, Field } from '@objectstack/spec/data'; + +/** + * sys_flow_dispatch — Persisted idempotency ledger for trigger dispatches + * (#10220). + * + * A time-relative sweep (`config.timeRelative`) evaluates its date window on + * every tick and launches the flow once per matching record — but the sweep + * itself holds no cross-tick memory, so every re-scan of the same window + * re-dispatched the same records (measured: 15 duplicate reminders in ~70s on + * a 5s interval, and a kernel rebuild re-dispatches even under a daily cron). + * This table is that memory: one row per **claimed dispatch key**, written by + * {@link ObjectStoreFlowDispatchStore.claim} before the flow is launched. + * + * The key (the row `id`) names the MATCHED WINDOW's identity, derived from the + * same `DateWindow` the sweep matched against (maintainer ruling 2026-08-20 on + * #10220): offset mode keys on `(flowName, recordId, windowDay, offset)`; + * range mode keys on `(flowName, recordId, sweepDay, rangeSpec)` — so a range + * flow still "fires every day the record stays in range" (the documented + * `withinDays` semantic), just never twice in one day, and an offset flow + * re-fires when the record's date field moves to a new window day. + * + * Every key embeds a calendar day, so a row is claimable on exactly one sweep + * day and is dead weight afterwards — ADR-0057 telemetry retention reaps rows + * after 30 days (comfortably >= any near-term catch-up horizon for cloud#1288's + * catch-up sweeps, which this ledger unblocks; widen there if that work needs + * more). + * + * Writers: the automation engine's {@link FlowDispatchStore} (`claim()`), + * check-and-record under a system context. Readers: the same claim path, and + * operability surfaces ("what did this sweep dispatch?"). + * + * @namespace sys + */ +export const SysFlowDispatch = ObjectSchema.create({ + name: 'sys_flow_dispatch', + label: 'Flow Dispatch', + pluralLabel: 'Flow Dispatches', + icon: 'repeat', + isSystem: true, + managedBy: 'engine-owned', + // ADR-0057: pure telemetry — every row's key embeds the one sweep day it can + // be claimed on, so rows have no read value after the window passes. 30-day + // retention per the #10220 ruling (>= the cloud#1288 catch-up horizon). + lifecycle: { + class: 'telemetry', + retention: { maxAge: '30d' }, + }, + description: + 'Idempotency ledger for trigger dispatches (#10220): one row per claimed (flow, record, matched-window) key, so a re-scan or a rebuilt kernel never re-launches a flow for a window it already dispatched.', + displayNameField: 'id', + nameField: 'id', // [ADR-0079] canonical primary-title pointer (mirrors deprecated displayNameField) + highlightFields: ['id', 'dispatched_at'], + + fields: { + // The dispatch key IS the identity — using it as the primary key makes + // claim() a natural check-and-record (a concurrent duplicate insert fails + // on the id, and the claimer re-reads to see it lost the race). + id: Field.text({ label: 'Dispatch Key', required: true, readonly: true, group: 'System' }), + + dispatched_at: Field.datetime({ + label: 'Dispatched At', + required: true, + description: 'When the dispatch key was claimed (immediately before the flow launch it deduplicates).', + group: 'State', + }), + + created_at: Field.datetime({ + label: 'Created At', + required: true, + defaultValue: 'NOW()', + readonly: true, + group: 'System', + }), + }, + + indexes: [ + // Retention age sweep: the platform Reaper deletes rows older than + // `retention.maxAge` by created_at. + { fields: ['created_at'] }, + ], + + enable: { + // [ADR-0103] Engine-owned: written only by the automation engine's claim + // path (SYSTEM_CTX), never via the generic data API. Reads stay open. + apiMethods: ['get', 'list'], + }, +}); diff --git a/packages/spec/src/system/constants/platform-object-names.ts b/packages/spec/src/system/constants/platform-object-names.ts index 827504738c..f094248680 100644 --- a/packages/spec/src/system/constants/platform-object-names.ts +++ b/packages/spec/src/system/constants/platform-object-names.ts @@ -110,8 +110,8 @@ export const PLATFORM_OBJECTS_BY_PACKAGE: Readonly this.resolveService(ctx, 'job'), () => this.resolveDataEngine(ctx), ctx.logger, + undefined, // default wall clock + // #10220 — dispatch-idempotency claims go through the SAME + // automation service this plugin already resolves; the trigger + // computes the key and never learns the ledger's table name. An + // automation service predating claim() resolves to null and the + // trigger degrades (honestly, warned once) to in-process dedup. + () => { + const svc = this.resolveService>(ctx, 'automation'); + return svc && typeof svc.claim === 'function' ? (svc as FlowDispatchClaimSurface) : null; + }, ); automation.registerTrigger(trigger); ctx.logger.info('TimeRelativeTriggerPlugin: time-relative trigger registered'); diff --git a/packages/triggers/trigger-schedule/src/time-relative-trigger.test.ts b/packages/triggers/trigger-schedule/src/time-relative-trigger.test.ts index 13b6e792d0..3a02867ac8 100644 --- a/packages/triggers/trigger-schedule/src/time-relative-trigger.test.ts +++ b/packages/triggers/trigger-schedule/src/time-relative-trigger.test.ts @@ -5,7 +5,9 @@ import type { AutomationContext, JobSchedule, JobHandler } from '@objectstack/sp import { TimeRelativeTrigger, computeDateWindows, + computeWindowClaimScopes, buildWindowWhere, + type FlowDispatchClaimSurface, type TimeRelativeDataEngine, type FlowTriggerBinding, type JobServiceSurface, @@ -425,6 +427,227 @@ describe('TimeRelativeTrigger', () => { }); }); +// ─── Dispatch idempotency (#10220) ────────────────────────────────── + +/** + * Fake persisted claim ledger: a Set that OUTLIVES trigger instances, so + * sharing one across two triggers simulates a kernel rebuild against a + * surviving `sys_flow_dispatch` table. + */ +function fakeClaimLedger() { + const keys = new Set(); + const claims: string[] = []; + const surface: FlowDispatchClaimSurface = { + async claim(key) { + claims.push(key); + if (keys.has(key)) return false; + keys.add(key); + return true; + }, + }; + return { surface, keys, claims }; +} + +describe('TimeRelativeTrigger dispatch idempotency (#10220)', () => { + const JOB = 'flow-time-relative:renewal_alert'; + + it('offset mode: two sweeps over the same window dispatch once', async () => { + const rows: Row[] = [{ id: 'c1', end_date: '2026-07-25T09:00:00.000Z' }]; // T+7 from NOW + const job = fakeJobService(); + const { engine } = fakeDataEngine(rows); + const ledger = fakeClaimLedger(); + const trigger = new TimeRelativeTrigger( + () => job.service, () => engine, silentLogger(), NOW, () => ledger.surface, + ); + const launched: string[] = []; + + trigger.start(binding({ object: 'contracts', dateField: 'end_date', offsetDays: [7] }), async (ctx) => { + launched.push((ctx.record as Row).id as string); + }); + await flush(); + await job.fire(JOB); + await job.fire(JOB); + + expect(launched).toEqual(['c1']); + // The key names the MATCHED WINDOW's identity (windowDay + offset + record). + expect(ledger.claims[0]).toBe('time-relative:renewal_alert:2026-07-25:offset7:c1'); + expect(ledger.claims).toHaveLength(2); // second sweep asked and was refused + }); + + it('kernel rebuild (fresh trigger instance, same persisted ledger): still once', async () => { + const rows: Row[] = [{ id: 'c1', end_date: '2026-07-25T09:00:00.000Z' }]; + const ledger = fakeClaimLedger(); + const launched: string[] = []; + const desc = { object: 'contracts', dateField: 'end_date', offsetDays: [7] }; + + for (let boot = 0; boot < 2; boot++) { + const job = fakeJobService(); + const { engine } = fakeDataEngine(rows); + // A NEW trigger instance per boot — only the ledger survives. + const trigger = new TimeRelativeTrigger( + () => job.service, () => engine, silentLogger(), NOW, () => ledger.surface, + ); + trigger.start(binding(desc), async (ctx) => { + launched.push((ctx.record as Row).id as string); + }); + await flush(); + await job.fire(JOB); + } + + expect(launched).toEqual(['c1']); // the rebuild did not re-mint + }); + + it('range mode: twice in one day → once; the next day → fires again (withinDays prose preserved)', async () => { + const rows: Row[] = [{ id: 'c1', end_date: '2026-08-10T09:00:00.000Z' }]; // in [today, +30d] + let current = new Date('2026-07-18T12:00:00.000Z'); + const clock = () => current; + const job = fakeJobService(); + const { engine } = fakeDataEngine(rows); + const ledger = fakeClaimLedger(); + const trigger = new TimeRelativeTrigger( + () => job.service, () => engine, silentLogger(), clock, () => ledger.surface, + ); + const launched: string[] = []; + + trigger.start(binding({ object: 'contracts', dateField: 'end_date', withinDays: 30 }), async (ctx) => { + launched.push((ctx.record as Row).id as string); + }); + await flush(); + await job.fire(JOB); + await job.fire(JOB); // same sweep day — deduped + expect(launched).toEqual(['c1']); + expect(ledger.claims[0]).toBe('time-relative:renewal_alert:2026-07-18:within30:c1'); + + current = new Date('2026-07-19T12:00:00.000Z'); // next day — new sweepDay, new key + await job.fire(JOB); + expect(launched).toEqual(['c1', 'c1']); + expect(ledger.claims[ledger.claims.length - 1]).toBe('time-relative:renewal_alert:2026-07-19:within30:c1'); + }); + + it('offset mode: a dateField edit that moves the window fires again for the NEW window', async () => { + const row: Row = { id: 'c1', end_date: '2026-07-25T09:00:00.000Z' }; // T+7 from 07-18 + let current = new Date('2026-07-18T12:00:00.000Z'); + const clock = () => current; + const job = fakeJobService(); + const { engine } = fakeDataEngine([row]); + const ledger = fakeClaimLedger(); + const trigger = new TimeRelativeTrigger( + () => job.service, () => engine, silentLogger(), clock, () => ledger.surface, + ); + const launched: string[] = []; + + trigger.start(binding({ object: 'contracts', dateField: 'end_date', offsetDays: [7] }), async (ctx) => { + launched.push((ctx.record as Row).id as string); + }); + await flush(); + await job.fire(JOB); + expect(launched).toEqual(['c1']); + + // The due date is postponed → the record leaves the old window and, + // three days later, matches a NEW window day (07-28 = 07-21 + 7). + row.end_date = '2026-07-28T09:00:00.000Z'; + current = new Date('2026-07-21T12:00:00.000Z'); + await job.fire(JOB); + expect(launched).toEqual(['c1', 'c1']); // re-fired for the new window + expect(ledger.claims[ledger.claims.length - 1]).toBe('time-relative:renewal_alert:2026-07-28:offset7:c1'); + }); + + it('a claim-store failure never blocks the dispatch (availability over strict-once)', async () => { + const rows: Row[] = [{ id: 'c1', end_date: '2026-07-25T09:00:00.000Z' }]; + const job = fakeJobService(); + const { engine } = fakeDataEngine(rows); + const warn = vi.fn(); + const surface: FlowDispatchClaimSurface = { + async claim() { throw new Error('ledger table unreachable'); }, + }; + const trigger = new TimeRelativeTrigger( + () => job.service, () => engine, + { info: () => {}, warn, debug: () => {} }, + NOW, () => surface, + ); + const launched: string[] = []; + + trigger.start(binding({ object: 'contracts', dateField: 'end_date', offsetDays: [7] }), async (ctx) => { + launched.push((ctx.record as Row).id as string); + }); + await flush(); + await job.fire(JOB); + + expect(launched).toEqual(['c1']); // dispatched despite the failing claim + expect(warn).toHaveBeenCalledWith(expect.stringContaining('dispatch-claim failed')); + }); + + it('no claim surface: in-process dedup still holds within one process, and the degradation is warned once', async () => { + const rows: Row[] = [{ id: 'c1', end_date: '2026-07-25T09:00:00.000Z' }]; + const job = fakeJobService(); + const { engine } = fakeDataEngine(rows); + const warn = vi.fn(); + // Four-arg construction — the pre-#10220 shape every host without an + // automation claim() surface effectively uses. + const trigger = new TimeRelativeTrigger( + () => job.service, () => engine, + { info: () => {}, warn, debug: () => {} }, + NOW, + ); + const launched: string[] = []; + + trigger.start(binding({ object: 'contracts', dateField: 'end_date', offsetDays: [7] }), async (ctx) => { + launched.push((ctx.record as Row).id as string); + }); + await flush(); + await job.fire(JOB); + await job.fire(JOB); + + expect(launched).toEqual(['c1']); // deduped in-process + const degradations = warn.mock.calls.filter( + (c) => typeof c[0] === 'string' && (c[0] as string).includes('IN-PROCESS ONLY'), + ); + expect(degradations).toHaveLength(1); // said once, not per tick + }); + + it('records without an id are dispatched unconditionally (never claimed)', async () => { + const rows: Row[] = [{ end_date: '2026-07-25T09:00:00.000Z' }]; // no id + const job = fakeJobService(); + const { engine } = fakeDataEngine(rows); + const ledger = fakeClaimLedger(); + const trigger = new TimeRelativeTrigger( + () => job.service, () => engine, silentLogger(), NOW, () => ledger.surface, + ); + let launched = 0; + + trigger.start(binding({ object: 'contracts', dateField: 'end_date', offsetDays: [7] }), async () => { + launched++; + }); + await flush(); + await job.fire(JOB); + await job.fire(JOB); + + expect(launched).toBe(2); // not dedupable — unchanged pre-#10220 behaviour + expect(ledger.claims).toHaveLength(0); + }); +}); + +describe('computeWindowClaimScopes', () => { + const now = new Date('2026-07-18T12:00:00.000Z'); + + it('offset mode scopes on the window day + offset', () => { + const scopes = computeWindowClaimScopes({ object: 'c', dateField: 'd', offsetDays: [7, 30] }, now); + expect(scopes.map((s) => s.scope)).toEqual(['2026-07-25:offset7', '2026-08-17:offset30']); + }); + + it('range mode scopes on the SWEEP day + range spec (fires daily, never twice a day)', () => { + expect(computeWindowClaimScopes({ object: 'c', dateField: 'd', withinDays: 30 }, now)[0].scope) + .toBe('2026-07-18:within30'); + expect(computeWindowClaimScopes({ object: 'c', dateField: 'd', withinDays: -14 }, now)[0].scope) + .toBe('2026-07-18:within-14'); + }); + + it('windows are exactly computeDateWindows (one derivation, no drift)', () => { + const desc = { object: 'c', dateField: 'd', offsetDays: [60, 30, 7] }; + expect(computeWindowClaimScopes(desc, now).map((s) => s.window)).toEqual(computeDateWindows(desc, now)); + }); +}); + // ─── TimeRelativeTriggerPlugin ────────────────────────────────────── describe('TimeRelativeTriggerPlugin', () => { @@ -470,6 +693,31 @@ describe('TimeRelativeTriggerPlugin', () => { expect(registerTrigger).toHaveBeenCalledTimes(1); }); + it('wires the automation service claim() as the dispatch-idempotency surface (#10220)', async () => { + const registerTrigger = vi.fn(); + const claim = vi.fn(async () => true); + const job = fakeJobService(); + // Far-future date: the plugin wires the real wall clock, so the row + // must sit inside [real-today, +36500d] whenever this suite runs. + const { engine } = fakeDataEngine([{ id: 'c1', end_date: '2099-01-01T00:00:00.000Z' }]); + const fake = fakePluginCtx({ automation: { registerTrigger, claim }, job: job.service, objectql: engine }); + + const plugin = new TimeRelativeTriggerPlugin(); + await plugin.start(fake.ctx as never); + await fake.readyHandlers[0](); + + const trigger = registerTrigger.mock.calls[0][0] as TimeRelativeTrigger; + // Deterministic clock is a constructor-only injection, so drive the + // sweep through a descriptor whose window is computed from real "now": + // withinDays 36500 always includes the row's date. + trigger.start(binding({ object: 'contracts', dateField: 'end_date', withinDays: 36_500 }), async () => {}); + await flush(); + await job.fire('flow-time-relative:renewal_alert'); + + expect(claim).toHaveBeenCalledTimes(1); + expect(claim).toHaveBeenCalledWith(expect.stringMatching(/^time-relative:renewal_alert:\d{4}-\d{2}-\d{2}:within36500:c1$/)); + }); + it('skips gracefully when the automation service is absent', async () => { const job = fakeJobService(); const fake = fakePluginCtx({ job: job.service }); diff --git a/packages/triggers/trigger-schedule/src/time-relative-trigger.ts b/packages/triggers/trigger-schedule/src/time-relative-trigger.ts index d58d1ef081..27138da89d 100644 --- a/packages/triggers/trigger-schedule/src/time-relative-trigger.ts +++ b/packages/triggers/trigger-schedule/src/time-relative-trigger.ts @@ -37,11 +37,32 @@ export interface TimeRelativeDataEngine { getObject?(name: string): unknown; } +/** + * The slice of the automation service this trigger needs for dispatch + * idempotency (#10220): claim a dispatch key against the persisted + * `sys_flow_dispatch` ledger. `true` = this caller owns the dispatch; `false` = + * an earlier sweep (possibly in a previous process lifetime) already made it. + * Typed structurally — like {@link TimeRelativeDataEngine} — so this plugin + * never learns the ledger's table name and takes no build dependency on + * `@objectstack/service-automation`. + */ +export interface FlowDispatchClaimSurface { + claim(key: string): Promise; +} + /** Job-name namespace so time-relative sweeps never collide with plain schedule jobs. */ const JOB_PREFIX = 'flow-time-relative'; const MS_PER_DAY = 86_400_000; +/** + * TTL for the trigger's IN-PROCESS claim fallback (#10220): every dispatch key + * embeds a calendar day, so no key is producible more than ~48h after it was + * first claimable — pruning at that age keeps the fallback map bounded without + * ever forgetting a key a sweep could still produce. + */ +const LOCAL_CLAIM_TTL_MS = 48 * 60 * 60 * 1000; + /** A closed, inclusive instant window `[gte, lte]` as ISO-8601 strings. */ export interface DateWindow { /** Lower bound (inclusive), ISO-8601. */ @@ -80,21 +101,54 @@ function addUtcDays(d: Date, n: number): Date { * `date` field (compared as `YYYY-MM-DD` after the driver truncates) is inclusive. */ export function computeDateWindows(desc: TimeRelativeDescriptor, now: Date): DateWindow[] { + return computeWindowClaimScopes(desc, now).map((s) => s.window); +} + +/** + * A date window paired with the **claim scope** naming its identity for the + * dispatch dedup key (#10220, maintainer ruling 2026-08-20). + */ +export interface WindowClaimScope { + window: DateWindow; + /** + * Window-identity fragment of the dispatch key — what makes a re-scan of + * the SAME window dedup while a genuinely new window fires again: + * + * - offset mode → `:offset`: the window day is the date the + * record's field must fall on, so editing the field to a new day (or a + * different offset matching) yields a new key and legitimately re-fires. + * Re-scans of one window all derive the same day → deduped. + * - range mode → `:within`: keyed on the SWEEP day, not the + * (constant) field value, so the documented `withinDays` semantic — + * "fires every day the record stays in range" — remains true: each new + * day is a new key, but never twice in one day. + */ + scope: string; +} + +/** + * {@link computeDateWindows}, with each window's claim scope (#10220). One + * derivation for both so the matching rule and the dedup key can never drift. + */ +export function computeWindowClaimScopes(desc: TimeRelativeDescriptor, now: Date): WindowClaimScope[] { const today = startOfUtcDay(now); if (desc.offsetDays && desc.offsetDays.length > 0) { return desc.offsetDays.map((offset) => { const day = addUtcDays(today, offset); - return { gte: startOfUtcDay(day).toISOString(), lte: endOfUtcDay(day).toISOString() }; + const window = { gte: startOfUtcDay(day).toISOString(), lte: endOfUtcDay(day).toISOString() }; + return { window, scope: `${window.gte.slice(0, 10)}:offset${offset}` }; }); } const n = desc.withinDays ?? 0; - if (n >= 0) { - return [{ gte: startOfUtcDay(today).toISOString(), lte: endOfUtcDay(addUtcDays(today, n)).toISOString() }]; - } - // Negative: window extends into the past, still anchored to (and including) today. - return [{ gte: startOfUtcDay(addUtcDays(today, n)).toISOString(), lte: endOfUtcDay(today).toISOString() }]; + const sweepDay = today.toISOString().slice(0, 10); + const window: DateWindow = + n >= 0 + ? { gte: startOfUtcDay(today).toISOString(), lte: endOfUtcDay(addUtcDays(today, n)).toISOString() } + // Negative: window extends into the past, still anchored to (and including) today. + : { gte: startOfUtcDay(addUtcDays(today, n)).toISOString(), lte: endOfUtcDay(today).toISOString() }; + return [{ window, scope: `${sweepDay}:within${n}` }]; } /** @@ -147,17 +201,29 @@ export class TimeRelativeTrigger implements FlowTrigger { private readonly now: () => Date; /** flowName → job name registered for it, so stop() can cancel it. */ private readonly bound = new Map(); + /** Dispatch-idempotency claim surface (#10220), resolved lazily per sweep. */ + private readonly getClaimSurface: () => FlowDispatchClaimSurface | null; + /** + * In-process claim fallback when no claim surface resolves (#10220): + * key → claim time (epoch ms), TTL-pruned. Dedups re-scans within THIS + * process only — which is why falling to it is warned once, below. + */ + private readonly localClaims = new Map(); + /** Whether the in-process-only dedup degradation has been said (once). */ + private claimDegradationWarned = false; constructor( getJobService: () => JobServiceSurface | null, getDataEngine: () => TimeRelativeDataEngine | null, logger: TriggerLogger, now: () => Date = () => new Date(), + getClaimSurface: () => FlowDispatchClaimSurface | null = () => null, ) { this.getJobService = getJobService; this.getDataEngine = getDataEngine; this.logger = logger; this.now = now; + this.getClaimSurface = getClaimSurface; } start(binding: FlowTriggerBinding, callback: (ctx: AutomationContext) => Promise): void { @@ -265,11 +331,11 @@ export class TimeRelativeTrigger implements FlowTrigger { return; } - const windows = computeDateWindows(desc, this.now()); + const scopes = computeWindowClaimScopes(desc, this.now()); const seenIds = new Set(); - const matched: Array> = []; + const matched: Array<{ record: Record; claimKey: string | null }> = []; - for (const window of windows) { + for (const { window, scope } of scopes) { if (matched.length >= maxRecords) break; const where = buildWindowWhere(desc, window); const rows = @@ -286,7 +352,11 @@ export class TimeRelativeTrigger implements FlowTrigger { if (seenIds.has(id)) continue; seenIds.add(id); } - matched.push(row); + // #10220 — dispatch key: the MATCHED WINDOW's identity + the + // record. A row without an id can't be keyed; it is dispatched + // unconditionally, exactly as it was never dedupable before. + const claimKey = id != null ? `time-relative:${flowName}:${scope}:${String(id)}` : null; + matched.push({ record: row, claimKey }); if (matched.length >= maxRecords) break; } } @@ -300,7 +370,16 @@ export class TimeRelativeTrigger implements FlowTrigger { let launched = 0; let failed = 0; - for (const record of matched) { + let deduped = 0; + for (const { record, claimKey } of matched) { + // #10220 — idempotency gate: launch only if this (flow, record, + // window) key has not been dispatched before. A re-scan of the same + // window (denser schedule, kernel rebuild + persisted ledger, + // future catch-up sweep) skips instead of re-minting. + if (claimKey != null && !(await this.claimDispatch(flowName, claimKey))) { + deduped++; + continue; + } try { const ctx: AutomationContext = { record, @@ -326,10 +405,56 @@ export class TimeRelativeTrigger implements FlowTrigger { } this.logger.debug?.( - `[time-relative] flow '${flowName}' swept '${desc.object}': ${matched.length} matched, ${launched} launched, ${failed} failed`, + `[time-relative] flow '${flowName}' swept '${desc.object}': ${matched.length} matched, ${launched} launched, ${deduped} already dispatched, ${failed} failed`, ); } + /** + * Claim one dispatch key (#10220): `true` = launch, `false` = an earlier + * sweep already dispatched this (flow, record, window). + * + * Degradation contract: + * - Claim surface resolves (the automation service's `claim()`, backed by + * the persisted `sys_flow_dispatch` ledger) → its answer is used; if the + * CALL throws, the failure is logged and the dispatch proceeds — + * availability over strict-once: a broken ledger must never silently + * swallow reminders. + * - No claim surface (automation service missing, or one predating + * `claim()`) → in-process dedup only, warned ONCE: a silent fallback + * would hide that the once-per-window guarantee no longer survives a + * kernel rebuild. + */ + private async claimDispatch(flowName: string, key: string): Promise { + const surface = this.getClaimSurface(); + if (surface && typeof surface.claim === 'function') { + try { + return await surface.claim(key); + } catch (err) { + this.logger.warn( + `[time-relative] flow '${flowName}' dispatch-claim failed for key '${key}' — dispatching anyway ` + + `(availability over strict-once; the same window may re-fire until the claim store recovers): ${errMessage(err)}`, + ); + return true; + } + } + if (!this.claimDegradationWarned) { + this.claimDegradationWarned = true; + this.logger.warn( + `[time-relative] no dispatch-claim surface (automation service missing or without claim()) — ` + + `sweep dedup is IN-PROCESS ONLY and will NOT survive a kernel rebuild: ` + + `the same record/window can re-fire after a restart.`, + ); + } + const now = this.now().getTime(); + const cutoff = now - LOCAL_CLAIM_TTL_MS; + for (const [k, t] of this.localClaims) { + if (t < cutoff) this.localClaims.delete(k); + } + if (this.localClaims.has(key)) return false; + this.localClaims.set(key, now); + return true; + } + stop(flowName: string): void { const jobName = this.bound.get(flowName); if (!jobName) return;