fix(trigger-schedule,service-automation): persisted sys_flow_dispatch ledger makes time-relative sweeps idempotent per matched window

.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<boolean>` 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).

packages/services/service-automation/src/engine.ts

@@ -1133,6 +1133,27 @@ export interface SuspendedRunStore {
  loadTerminal?(runId: string): Promise<RunRecord | null>;
 }
 
+/**
+ * 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<boolean>;
+}
+
+/**
+ * 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<string>();
+ /**
+ * 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<string, number>();
+ /**
+ * 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<boolean> {
+ 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

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<any[]>;
+ insert(object: string, data: any, options?: any): Promise<any>;
+}
+
+/** In-memory {@link FlowDispatchStore} — process-lifetime dedup only. */
+export class InMemoryFlowDispatchStore implements FlowDispatchStore {
+ private readonly keys = new Set<string>();
+
+ async claim(key: string): Promise<boolean> {
+ 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<boolean> {
+ 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<void> {
+ await this.engine.find(TABLE, { where: {}, limit: 1, context: SYSTEM_CTX });
+ }
+}