From 94b8563368a1cfefc11d2cd3d5826d9aa691a690 Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 6 Sep 2026 01:38:32 +0000 Subject: [PATCH 1/2] fix(service-automation): decide the claim capability before the compare-and-set, never after it MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `ObjectStoreSuspendedRunStore.claimSuspension` decided `'unsupported'` — "no cross-replica advance guarantee is offered by this store" — from the SHAPE of the return value, one line after the conditional delete had already gone out. On an engine whose multi-delete does not resolve an affected-row count that made the refusal a statement about a write that had already landed: the compare-and-set was performed against the shared row and its verdict discarded, `claimAdvance` read `'unsupported'` as `unguarded`, and a replica that actually LOST the claim resumed anyway — the doubled side effect #14333 exists to prevent, on the one composition that declares itself unable to prevent it. Two arms, because `ObjectQL.delete` declares `Promise` and there is no read-only instrument for "does this engine's multi-delete return a count": - a one-time capability probe down the same route, against a sentinel predicate that matches no row, so an engine that cannot count is refused with nothing consumed and `claimAdvance`'s `unguarded` reading is true when it is taken; - after the write, `'unsupported'` is retired: a committed compare-and-set with an unreadable verdict is UNKNOWN, so the store throws and `claimAdvance` answers STORE_UNAVAILABLE — the resume is refused, not continued. The guarantee itself is not restored for an uncounted engine and the change does not claim it is; the count is contracted at `IDataDriver.deleteMany` and erased to `any` at the engine boundary, which is #16033. Refs #15832 (Note 2) Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01XpTx2tbq3pZRYAdoGt6E6Y --- .../claim-capability-probe-before-mutate.md | 16 + .../src/suspended-run-claim-probe.test.ts | 532 ++++++++++++++++++ .../src/suspended-run-store.test.ts | 17 +- .../src/suspended-run-store.ts | 205 ++++++- 4 files changed, 760 insertions(+), 10 deletions(-) create mode 100644 .changeset/claim-capability-probe-before-mutate.md create mode 100644 packages/services/service-automation/src/suspended-run-claim-probe.test.ts diff --git a/.changeset/claim-capability-probe-before-mutate.md b/.changeset/claim-capability-probe-before-mutate.md new file mode 100644 index 0000000000..96c34c184b --- /dev/null +++ b/.changeset/claim-capability-probe-before-mutate.md @@ -0,0 +1,16 @@ +--- +"@objectstack/service-automation": patch +--- + +`ObjectStoreSuspendedRunStore` no longer announces "no cross-replica advance guarantee is offered" *after* it has issued the guarded delete. + +`claimSuspension` is the cross-replica half of the resume idempotency guard: it removes the `sys_automation_run` row only if the run is still parked where this replica read it, and the affected-row count names the winner. The refusal for an engine that does not resolve such a count was decided on the SHAPE of the return value — one line after the compare-and-set had already gone out. On such an engine that made the refusal a statement about a write that had already landed: the conditional delete was performed against the shared row and its verdict discarded, `AutomationEngine.claimAdvance` read `'unsupported'` as `unguarded`, and a replica that **actually lost** the claim (0 rows affected) resumed anyway — running every downstream side effect a second time, on the one composition that declares itself unable to prevent that. + +The capability question is now settled before anything is claimed, and `'unsupported'` is retired as an answer once the row has been touched: + +- **A one-time capability probe, before the compare-and-set.** Once per store instance, `claimSuspension` issues one delete down the very route the claim takes (`multi: true` with a `where` carrying keys besides `id`, which is what dispatches to `driver.deleteMany`) against a sentinel predicate that matches no row — the same value in `id`, `node_id` and `correlation` at once. An engine that resolves something other than a count is refused with **nothing consumed**, so `claimAdvance`'s `unguarded` reading is true when it is taken. Concurrent first claims share one probe, and a probe that *throws* is deliberately not memoized: a store that was unreachable for one second must not answer for the life of the process. +- **After the write, an unreadable verdict is `STORE_UNAVAILABLE`, not `unguarded`.** If a probed-counting engine still resolves a non-count for a real claim, the compare-and-set is committed and its verdict is unrecoverable — a winner and a loser both find the row gone, so no follow-up read can tell them apart. The store throws instead of answering `'unsupported'`; `claimAdvance` already maps that to `STORE_UNAVAILABLE`, whose text is written for exactly this fact ("a failure can arrive after a committed delete"), and the resume is **refused** rather than continued. A claim that in fact won is then stranded until an operator retries — the deliberate direction, since a doubled side effect is the worse outcome. + +**What this does not do, stated so it is not read into it.** It does not give an uncounted engine the guarantee. `ObjectQL.delete` declares `Promise`, so "does a multi-delete return a count" has no contractual answer to look up and no read-only instrument to measure — a probe can observe the route once, never promise what the next call resolves to. Closing that gap belongs to the engine boundary, where the count is contracted one layer down (`IDataDriver.deleteMany`, `Promise`) and erased to `any` on the way up. On such a composition the store still degrades to an unguarded resume; what changed is that it says so before consuming anything, and the run's durable row is still removed by the consumption choke point exactly as before. + +Every measured shipped composition already resolves a count (memory, sql/better-sqlite3, sqlite-wasm, turso local and remote transport, sql with the security plugin composed), so the observable cost there is one extra `DELETE … WHERE` that matches nothing, once per process. It emits no hook dispatch, no realtime event and no row change: the per-row before phase is "zero matched rows is zero dispatches", the after phase iterates the same empty set, and `publishBulkDataEvent` returns at `matched === 0` by design. diff --git a/packages/services/service-automation/src/suspended-run-claim-probe.test.ts b/packages/services/service-automation/src/suspended-run-claim-probe.test.ts new file mode 100644 index 0000000000..6f496a1a79 --- /dev/null +++ b/packages/services/service-automation/src/suspended-run-claim-probe.test.ts @@ -0,0 +1,532 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * [#15832 note 2] `ObjectStoreSuspendedRunStore.claimSuspension` must never + * announce `'unsupported'` AFTER it has issued the compare-and-set. + * + * ## The defect, in the terms that make it a correctness card + * + * The refusal used to be decided on the SHAPE of the return value: + * + * ```ts + * const affected = await this.engine.delete(TABLE, { where, multi: true, … }); + * if (typeof affected !== 'number') { warn(…); return 'unsupported'; } + * ``` + * + * `'unsupported'` means "no cross-replica advance guarantee is offered by this + * store". Said HERE it is a statement about a write that has already landed + * against the shared row: the conditional delete went out, its verdict was + * discarded, and {@link AutomationEngine.claimAdvance} reads `'unsupported'` + * as `unguarded` — so a replica that ACTUALLY LOST the compare-and-set (0 rows + * affected) resumed anyway. That is the doubled side effect #14333 exists to + * prevent, occurring on the one composition that declares itself unable to + * prevent it. + * + * ⛔ The redundant delete round-trip the old shape also bought (`claimAdvance` + * reads `unguarded`, so `forgetSuspendedRun` issues a second, unconditional + * delete of the same row) is real and is NOT what this file is about. A change + * that only removed it would leave every case below red. + * + * ## The invariant this file pins + * + * > No path through `claimSuspension` answers `'unsupported'` once a delete + * > carrying the run's condition has been issued. + * + * Two arms, because the question "does this engine's multi-delete resolve a + * count?" has no contractual answer to look up — `ObjectQL.delete` declares + * `Promise` — and therefore no read-only instrument: + * + * - **before the write**: a one-time capability probe down the SAME route, + * against a predicate that matches no row. An engine that does not count is + * refused with nothing consumed, which is what makes `claimAdvance`'s + * `unguarded` reading TRUE when it is taken. + * - **after the write**: `'unsupported'` is retired as an answer. A committed + * compare-and-set whose verdict is unreadable is UNKNOWN, not unguarded — + * a winner and a loser both find the row gone, so nothing can tell them + * apart afterwards — and the store THROWS, which `claimAdvance` already + * turns into `STORE_UNAVAILABLE` and a REFUSED resume. + * + * ⚠️ What is deliberately NOT claimed: that an uncounted engine gains the + * guarantee. It does not, and cannot from inside this store — the count is + * contracted one layer down (`IDataDriver.deleteMany`, `Promise`) and + * erased to `any` at the engine boundary, which is #16033. What changes here + * is that the store's declaration stops being false at the moment it is made. + * + * ## Population — what each case actually drives + * + * Fourteen cases over three populations: + * + * 1. **Store-level, one fake data engine** (`createProbeAwareEngine`, whose + * `delete` is bound to the producer's own dispatch predicate, so it cannot + * accept a call `ObjectQL.delete` refuses). Three engine return shapes are + * driven: a counting one, a uniformly non-counting one (`undefined`), and + * an INCONSISTENT one that counts for the probe and does not for the claim + * — the only shape on which the second arm above is reachable. + * 2. **Two `AutomationEngine` replicas over ONE + * `ObjectStoreSuspendedRunStore`**, over one of those fake engines. Two + * replicas is the whole modelled fleet: it is the smallest number on which + * "the loser resumes too" is observable, and the side effects are counted + * off a shared ledger rather than off call spies. + * 3. **A real kernel** — `ObjectKernel` + `ObjectQLPlugin` + `SqlDriver` + * (better-sqlite3) over a real `sys_automation_run` table — for the two + * facts a fake cannot witness: that the shipped composition takes the + * COUNTED path, and that the probe's predicate consumes nothing there. + * ⛔ Only better-sqlite3 is installable in this container, so postgres, + * mysql, mongodb and a hosted Turso endpoint are UNMEASURED here, as they + * were for this card's phase-1 sweep. + */ + +import { mkdtempSync, rmSync } from 'node:fs'; +import { tmpdir } from 'node:os'; +import { join } from 'node:path'; + +import { describe, it, expect, afterEach } from 'vitest'; +import { defineActionDescriptor } from '@objectstack/spec/automation'; +import { RESUME_AUTHORITY_SERVICE } from '@objectstack/spec/contracts'; +import { ObjectKernel } from '@objectstack/core'; +import { ObjectQLPlugin, type ObjectQL } from '@objectstack/objectql'; +import { SqlDriver } from '@objectstack/driver-sql'; +// The PRODUCER's own delete-dispatch decision — the fake below routes `delete` +// through it so it physically cannot accept a call `ObjectQL.delete` refuses +// (`scripts/check-engine-double-contract.mjs`), and the cases read it directly +// to say WHICH route a recorded call would have taken. +import { assertEngineDeleteDispatch } from '@objectstack/metadata-core'; + +import { AutomationEngine } from './engine.js'; +import { AutomationServicePlugin } from './plugin.js'; +import { ObjectStoreSuspendedRunStore, type SuspendedRunStoreEngine } from './suspended-run-store.js'; +import type { RunRecord, SuspendedRun } from './engine.js'; + +const silent = () => ({ info() {}, warn() {}, error() {}, debug() {}, child: silent }) as any; + +/** Rows keyed by id, with `where` equality — the `sys_automation_run` table. */ +function createProbeAwareEngine( + /** + * What a `multi` delete resolves to, given how many rows it matched and the + * options bag it was called with. Omitted → the honest count, which is what + * every shipped driver measured for this card answers. + */ + multiResult?: (matched: number, options: any) => unknown, +): SuspendedRunStoreEngine & { rows: Map; deletes: any[] } { + const rows = new Map(); + const deletes: any[] = []; + const matches = (row: any, where: any) => + !where || Object.entries(where).every(([k, v]) => { + if (k.startsWith('$')) throw new Error(`fake driver: unsupported operator ${k}`); + return row[k] === v; + }); + return { + rows, + deletes, + async find(_object, options) { + const out = [...rows.values()].filter(r => matches(r, options?.where)); + return typeof options?.limit === 'number' ? out.slice(0, options.limit) : out; + }, + async insert(_object, data) { rows.set(String(data.id), { ...data }); return data; }, + async update(_object, data, options) { + const id = options?.where?.id ?? data.id; + rows.set(String(id), { ...(rows.get(String(id)) ?? { id }), ...data }); + return rows.get(String(id)); + }, + async delete(_object, options) { + // Bound to the producer's predicate, never a hand-written + // approximation of it: drop `multi: true` from the store and this + // THROWS, exactly as a running server does. + const dispatch = assertEngineDeleteDispatch(options as any); + deletes.push(options); + if (dispatch.kind === 'multi') { + const doomed = [...rows.values()].filter(r => matches(r, options?.where)); + for (const r of doomed) rows.delete(String(r.id)); + return multiResult ? multiResult(doomed.length, options) : doomed.length; + } + rows.delete(String(dispatch.id)); + return true; + }, + }; +} + +/** + * Recognise the one-time capability probe by its SHAPE, not by its literal + * sentinel: a `multi` delete whose three condition columns all carry the same + * value. Pinning the characters would pin a private constant; pinning the + * shape pins the property that makes the call safe — a row would have to carry + * that one string in `id` AND `node_id` AND `correlation` to match it. + */ +function isCapabilityProbe(options: any): boolean { + const where = options?.where ?? {}; + const keys = Object.keys(where).sort().join(','); + return assertEngineDeleteDispatch(options).kind === 'multi' + && keys === 'correlation,id,node_id' + && where.id === where.node_id + && where.node_id === where.correlation; +} + +const baseRun = (over: Partial = {}): SuspendedRun => ({ + runId: 'run_abc', + flowName: 'approval_flow', + flowVersion: 1, + nodeId: 'approve_step', + variables: { $runId: 'run_abc' }, + steps: [], + context: { object: 'crm_deal', userId: 'u1', tenantId: 'org_1' } as any, + startedAt: '2026-01-01T00:00:00.000Z', + startTime: 1735689600000, + correlation: 'areq_1', + ...over, +}); + +describe('#15832 the capability question is settled BEFORE the compare-and-set', () => { + it('⭐ an engine that does not count is refused with NOTHING consumed — the row survives the refusal', async () => { + // The composition the card is about: a multi-delete that resolves + // something other than an affected-row count. + const data = createProbeAwareEngine(() => undefined); + const lines: string[] = []; + const store = new ObjectStoreSuspendedRunStore(data, { warn: (m: string) => lines.push(m) } as any); + await store.save(baseRun()); + + expect(await store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' })) + .toBe('unsupported'); + + // ⭐ THE CARD. Before this change the compare-and-set had already gone + // out and taken the row with it, so this read was `false` — the store + // said "no guarantee is offered" about a guard it had just executed. + expect(data.rows.has('run_abc')).toBe(true); + expect(data.rows.get('run_abc').node_id).toBe('approve_step'); + // And not one delete carried the run's condition. + expect(data.deletes.filter(d => d.where?.id === 'run_abc')).toEqual([]); + // The only delete issued is the probe, which can match nothing. + expect(data.deletes).toHaveLength(1); + expect(isCapabilityProbe(data.deletes[0])).toBe(true); + + // The degradation is still DECLARED, and now truthfully. + expect(lines).toHaveLength(1); + expect(lines[0]).toContain('not an affected-row count'); + expect(lines[0]).toContain('no cross-replica advance guarantee'); + expect(lines[0]).toContain('NO delete was issued'); + }); + + it('the probe cannot consume anything: neither a live suspension nor a terminal history row', async () => { + const data = createProbeAwareEngine(() => undefined); + const store = new ObjectStoreSuspendedRunStore(data, silent()); + // Both row families this table holds, written by their real writers. + await store.save(baseRun()); + await store.recordTerminal({ + runId: 'abc2', flowName: 'approval_flow', status: 'completed', + startedAt: '2026-01-01T00:00:00.000Z', triggerType: 'manual', + } as RunRecord); + const before = [...data.rows.keys()].sort(); + expect(before).toHaveLength(2); + + await store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' }); + + expect([...data.rows.keys()].sort()).toEqual(before); + }); + + it('the probe takes the SAME route the claim takes — `multi`, not by-id', async () => { + // If it took the by-id route it would answer about `driver.delete`, + // whose contract is a boolean, and every counted engine would read as + // uncounted. The route IS the question. + const data = createProbeAwareEngine(); + const store = new ObjectStoreSuspendedRunStore(data, silent()); + await store.claimSuspension('run_abc', { nodeId: 'approve_step' }); + + expect(assertEngineDeleteDispatch(data.deletes[0])).toEqual({ kind: 'multi' }); + expect(isCapabilityProbe(data.deletes[0])).toBe(true); + }); + + it('costs ONE probe per store, not one per resume — three claims, one probe', async () => { + const data = createProbeAwareEngine(() => undefined); + const lines: string[] = []; + const store = new ObjectStoreSuspendedRunStore(data, { warn: (m: string) => lines.push(m) } as any); + + for (let i = 0; i < 3; i++) { + expect(await store.claimSuspension('run_abc', { nodeId: 'approve_step' })).toBe('unsupported'); + } + + expect(data.deletes).toHaveLength(1); + // Same posture the answer already had: a composition that cannot + // express the condition cannot express it for the life of the process. + expect(lines).toHaveLength(1); + }); + + it('concurrent first claims share ONE probe', async () => { + const data = createProbeAwareEngine(); + const store = new ObjectStoreSuspendedRunStore(data, silent()); + await store.save(baseRun()); + + const [first, second] = await Promise.all([ + store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' }), + store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' }), + ]); + + // Memoizing the VALUE rather than the promise would probe twice here. + expect(data.deletes.filter(isCapabilityProbe)).toHaveLength(1); + // And the claim itself is unchanged: exactly one of them consumed it. + expect([first, second].filter(o => o === 'claimed')).toHaveLength(1); + expect([first, second].filter(o => o === 'lost')).toHaveLength(1); + }); + + it('a counting engine pays the probe once and then claims exactly as before', async () => { + const data = createProbeAwareEngine(); + const lines: string[] = []; + const store = new ObjectStoreSuspendedRunStore(data, { warn: (m: string) => lines.push(m) } as any); + await store.save(baseRun()); + + expect(await store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' })) + .toBe('claimed'); + expect(await store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' })) + .toBe('lost'); + + // probe + claim + claim, and no probe on the second claim. + expect(data.deletes).toHaveLength(3); + expect(data.deletes.filter(isCapabilityProbe)).toHaveLength(1); + expect(isCapabilityProbe(data.deletes[0])).toBe(true); + // Nothing is declared: this composition offers the guarantee. + expect(lines).toEqual([]); + }); + + it('a THROWN probe is not memoized — a transient outage is not a permanent verdict', async () => { + let failing = true; + const data = createProbeAwareEngine(); + const throwing = { + ...data, + async delete(object: string, options: any) { + if (failing) throw new Error('sqlite: database is locked'); + return data.delete!(object, options); + }, + } as SuspendedRunStoreEngine & { rows: Map; deletes: any[] }; + const store = new ObjectStoreSuspendedRunStore(throwing, silent()); + await store.save(baseRun()); + + // The question could not be asked at all — which `claimAdvance` maps to + // STORE_UNAVAILABLE, the same answer the claim itself produced when it + // was the call that threw. + await expect(store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' })) + .rejects.toThrow('database is locked'); + // ⛔ And it refused before touching the row. + expect(data.rows.has('run_abc')).toBe(true); + + failing = false; + expect(await store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' })) + .toBe('claimed'); + }); + + it('an engine with no delete() still short-circuits ahead of the probe', async () => { + const data = createProbeAwareEngine(); + const noDelete = { find: data.find, insert: data.insert, update: data.update } as SuspendedRunStoreEngine; + const lines: string[] = []; + const store = new ObjectStoreSuspendedRunStore(noDelete, { warn: (m: string) => lines.push(m) } as any); + + expect(await store.claimSuspension('run_abc', { nodeId: 'approve_step' })).toBe('unsupported'); + expect(lines[0]).toContain('engine has no delete()'); + expect(data.deletes).toEqual([]); + }); +}); + +describe('#15832 after the write, `unsupported` is retired — an unreadable verdict is UNKNOWN', () => { + /** + * The ONLY shape on which this arm is reachable: an engine that answers a + * count for the probe and something else for a real claim. No measured + * composition does this; the arm exists because the probe cannot promise + * what the next call resolves to, and a residual nobody can rule out must + * fail SAFE rather than silently. + */ + const inconsistent = () => + createProbeAwareEngine((matched, options) => (isCapabilityProbe(options) ? matched : undefined)); + + it('⭐ the store THROWS rather than answering `unsupported` once the row has been touched', async () => { + const data = inconsistent(); + const lines: string[] = []; + const store = new ObjectStoreSuspendedRunStore(data, { warn: (m: string) => lines.push(m) } as any); + await store.save(baseRun()); + + await expect(store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' })) + .rejects.toThrow(/UNRECOVERABLE/); + + // The compare-and-set really did go out — that is what makes + // `'unsupported'` unsayable here rather than merely untidy. + expect(data.deletes.filter(d => d.where?.id === 'run_abc')).toHaveLength(1); + // ⛔ And it did NOT declare "no guarantee is offered": that sentence is + // the one `claimAdvance` reads as `unguarded`. + expect(lines).toEqual([]); + }); + + it('⭐ THE HARM: a replica whose claim verdict is unreadable is REFUSED, not resumed', async () => { + // Two replicas over one store, side effects counted off a shared + // ledger. On `main` both resumed and `notify` fired TWICE. + const data = inconsistent(); + const store = new ObjectStoreSuspendedRunStore(data, silent()); + const opened: string[] = []; + const fired: string[] = []; + const a = replica(store, opened, fired); + const b = replica(store, opened, fired); + + const runId = (await a.execute('expense_approval')).runId!; + expect(opened).toEqual(['lv1']); + + const both = await Promise.all([ + a.resume(runId, { [RESUME_AUTHORITY_SERVICE]: true } as any), + b.resume(runId, { [RESUME_AUTHORITY_SERVICE]: true } as any), + ]); + + // ⭐ The doubled side effect #14333 exists to prevent does not happen. + expect(fired).toEqual([]); + expect(both.every(r => !r.success)).toBe(true); + // Unknown, never "gone for good": the same envelope the strict load's + // failure uses, and the same remedy — retry. + expect(both.map(r => r.code)).toEqual(['STORE_UNAVAILABLE', 'STORE_UNAVAILABLE']); + expect(both[0].error).toContain('UNKNOWN'); + }); +}); + +describe('#15832 the declared degradation still works — an uncounted engine resumes and cleans up', () => { + it('the run advances, the durable row is still removed, and the engine declares once', async () => { + // ⚠️ The guarantee is NOT restored by this card: on an engine that + // cannot count, two replicas can still both advance. What changed is + // that the store says so before consuming anything. Pinned here so a + // future reader does not mistake the case above for a claim that the + // guarantee is now universal. + const data = createProbeAwareEngine(() => undefined); + const store = new ObjectStoreSuspendedRunStore(data, silent()); + const lines: string[] = []; + const opened: string[] = []; + const fired: string[] = []; + const engine = replica(store, opened, fired, lines); + + const runId = (await engine.execute('expense_approval')).runId!; + const resumed = await engine.resume(runId, { [RESUME_AUTHORITY_SERVICE]: true } as any); + + expect(resumed.success).toBe(true); + expect(fired).toEqual(['notify']); + // ⛔ No durability regression: `claimSuspension` no longer deletes on + // this path, and `forgetSuspendedRun`'s unconditional delete — which + // `unguarded` has always reached — still removes the consumed row. + expect(data.rows.has(runId)).toBe(false); + // The engine's own one-time degradation line, now true when it is said. + expect(lines.filter(l => l.includes('no cross-replica advance guarantee'))).toHaveLength(1); + }); +}); + +/** One replica over the shared store, appending to the shared ledgers. */ +function replica(store: any, opened: string[], fired: string[], warnLines?: string[]): AutomationEngine { + const logger: any = warnLines + ? { info() {}, warn: (m: string) => warnLines.push(m), error() {}, debug() {}, child: () => logger } + : silent(); + const engine = new AutomationEngine(logger, store); + engine.registerNodeExecutor({ + type: 'approval_level', + descriptor: defineActionDescriptor({ + type: 'approval_level', version: '1.0.0', name: 'Approval level', + supportsPause: true, resumeAuthority: 'service', + }), + async execute(node) { + opened.push(node.id); + return { success: true, suspend: true, correlation: `req_${node.id}` }; + }, + }); + engine.registerNodeExecutor({ + type: 'notify_action', + descriptor: defineActionDescriptor({ type: 'notify_action', version: '1.0.0', name: 'Notify' }), + async execute(node) { fired.push(node.id); return { success: true }; }, + }); + engine.registerFlow('expense_approval', { + name: 'expense_approval', label: 'Expense approval', type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'lv1', type: 'approval_level', label: 'Department head' }, + { id: 'notify', type: 'notify_action', label: 'Notify finance' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'lv1' }, + { id: 'e2', source: 'lv1', target: 'notify' }, + { id: 'e3', source: 'notify', target: 'end' }, + ], + } as never); + return engine; +} + +/** + * The two facts a fake engine cannot witness, measured against a real + * `sys_automation_run` table through a real `ObjectQL`. + * + * ⛔ better-sqlite3 only — postgres, mysql, mongodb and a hosted Turso endpoint + * are UNMEASURED by this file and are not implied by it. + */ +describe('#15832 real ObjectQL + SqlDriver: the shipped composition takes the COUNTED path', () => { + let dir: string | undefined; + const kernels: ObjectKernel[] = []; + + afterEach(async () => { + for (const k of kernels.splice(0)) { + try { await k.shutdown(); } catch { /* noop */ } + } + if (dir) { rmSync(dir, { recursive: true, force: true }); dir = undefined; } + }); + + async function boot() { + dir = mkdtempSync(join(tmpdir(), 'os-15832-')); + const kernel = new ObjectKernel({ logger: { level: 'fatal' } }); + kernels.push(kernel); + await kernel.use(new ObjectQLPlugin()); + await kernel.use(new AutomationServicePlugin()); + await kernel.bootstrap(); + + const ql = kernel.getService('objectql'); + const driver = new SqlDriver({ + client: 'better-sqlite3', + connection: { filename: join(dir, 'data.db') }, + useNullAsDefault: true, + }); + await driver.connect(); + ql.registerDriver(driver, true); + await ql.syncSchemas(); + return ql; + } + + it('claims and loses off a real affected-row count, having probed first', async () => { + const ql = await boot(); + const seen: any[] = []; + // A pass-through recorder — the calls are the real engine's, only + // observed on the way in. + const recorded = new Proxy(ql as any, { + get(target, prop, receiver) { + if (prop !== 'delete') return Reflect.get(target, prop, receiver); + return async (object: string, options: any) => { seen.push(options); return target.delete(object, options); }; + }, + }) as unknown as SuspendedRunStoreEngine; + const store = new ObjectStoreSuspendedRunStore(recorded, silent()); + await store.save(baseRun()); + + expect(await store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' })) + .toBe('claimed'); + // A second claim is the LOSER's read: the row is gone, so the count is + // 0 — and 0 is a number, which is the whole point of the probe. + expect(await store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' })) + .toBe('lost'); + + expect(seen.filter(isCapabilityProbe)).toHaveLength(1); + expect(isCapabilityProbe(seen[0])).toBe(true); + }); + + it("the probe's own predicate resolves a COUNT of 0 and consumes nothing", async () => { + const ql = await boot(); + const store = new ObjectStoreSuspendedRunStore(ql as unknown as SuspendedRunStoreEngine, silent()); + await store.save(baseRun()); + + // The probe shape, issued directly at the real engine so the reading is + // about ObjectQL rather than about this store. + const sentinel = '__objectstack_suspended_run_claim_capability_probe__'; + const affected = await (ql as any).delete('sys_automation_run', { + where: { id: sentinel, node_id: sentinel, correlation: sentinel }, + multi: true, + context: { isSystem: true }, + }); + + // NOT MEASURED by reading source: this is the real engine's own answer. + expect(typeof affected).toBe('number'); + expect(affected).toBe(0); + // …and the parked row is untouched. + expect(await store.load('run_abc')).not.toBeNull(); + }); +}); diff --git a/packages/services/service-automation/src/suspended-run-store.test.ts b/packages/services/service-automation/src/suspended-run-store.test.ts index c422e251b4..0b2df2c701 100644 --- a/packages/services/service-automation/src/suspended-run-store.test.ts +++ b/packages/services/service-automation/src/suspended-run-store.test.ts @@ -844,14 +844,20 @@ describe('#14333 ObjectStoreSuspendedRunStore.claimSuspension — the production expect(await store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' })) .toBe('claimed'); - expect(seen).toHaveLength(1); + // [#15832] TWO calls now, and the first one is not a claim: the + // capability probe runs once per store instance BEFORE anything is + // consumed. `suspended-run-claim-probe.test.ts` owns what it is and + // what it may cost; here it is only skipped past so this case keeps + // asserting the same thing it always did about the CLAIM. + expect(seen).toHaveLength(2); + const claim = seen[1]; // The condition really is carried: id AND the parking, not id alone. - expect(seen[0].where).toEqual({ id: 'run_abc', node_id: 'approve_step', correlation: 'areq_1' }); + expect(claim.where).toEqual({ id: 'run_abc', node_id: 'approve_step', correlation: 'areq_1' }); // THE decision, taken by the producer's predicate over the very options // bag the store built. `by-id` would bind only the primary key and // silently discard the condition; `reject` is what a missing `multi` // produces, and it THROWS in a running server. - expect(assertEngineDeleteDispatch(seen[0])).toEqual({ kind: 'multi' }); + expect(assertEngineDeleteDispatch(claim)).toEqual({ kind: 'multi' }); // …and it actually removed the row. expect(engine.rows.has('run_abc')).toBe(false); }); @@ -867,8 +873,9 @@ describe('#14333 ObjectStoreSuspendedRunStore.claimSuspension — the production await parkRun(store, { correlation: undefined }); expect(await store.claimSuspension('run_abc', { nodeId: 'approve_step' })).toBe('claimed'); - expect(seen[0].where).toEqual({ id: 'run_abc', node_id: 'approve_step' }); - expect(assertEngineDeleteDispatch(seen[0])).toEqual({ kind: 'multi' }); + // [#15832] `seen[0]` is the one-time capability probe; the claim is next. + expect(seen[1].where).toEqual({ id: 'run_abc', node_id: 'approve_step' }); + expect(assertEngineDeleteDispatch(seen[1])).toEqual({ kind: 'multi' }); }); it('maps the affected-row COUNT to the outcome: 1 is claimed, 0 is lost', async () => { diff --git a/packages/services/service-automation/src/suspended-run-store.ts b/packages/services/service-automation/src/suspended-run-store.ts index 03b860aa39..04ac9f8760 100644 --- a/packages/services/service-automation/src/suspended-run-store.ts +++ b/packages/services/service-automation/src/suspended-run-store.ts @@ -41,6 +41,36 @@ const TABLE = 'sys_automation_run'; const HISTORY_PREFIX = 'run_'; const SYSTEM_CTX = { isSystem: true, positions: [], permissions: [] } as const; +/** + * [#15832] The sentinel the counted-multi-delete capability probe addresses — + * see {@link ObjectStoreSuspendedRunStore.probeCountedMultiDelete}. + * + * It is written into `id`, `node_id` AND `correlation` at once, so a row would + * have to carry this exact string in all three columns to match. No producer + * writes it into any of them: `id` is the engine's minted `runId` (or a + * {@link HISTORY_PREFIX}-prefixed terminal id), `node_id` is an authored flow + * node id, and `correlation` is minted by the pausing executor. + * + * ⛔ The zero-match property is NOT bought with an empty `$in` or any other + * "matches nothing by construction" operator. `{ id: { $in: [] } }` reads as + * the safer spelling and is the more dangerous one: it is a driver-by-driver + * question whether an empty `IN` compiles at all, and the failure direction of + * a builder that drops an empty clause is a `DELETE` over the whole table. + * Three ANDed equalities compile the same way everywhere and fail closed. + */ +const CLAIM_CAPABILITY_PROBE_SENTINEL = '__objectstack_suspended_run_claim_capability_probe__'; + +/** + * [#15832] What the one-time probe learned about this engine's counted + * multi-delete route — `ObjectQL.delete` declares `Promise`, so this is + * the only place the answer exists. + */ +type CountedMultiDeleteCapability = + /** The route resolved a number: the compare-and-set verdict is readable. */ + | { readonly counted: true } + /** It resolved something else; `observed` is that value's `typeof`. */ + | { readonly counted: false; readonly observed: string }; + /** * Default per-flow cap on terminal run-history rows (#2585). A busy * per-record-change flow otherwise persists one row per execution forever — @@ -253,6 +283,16 @@ export class ObjectStoreSuspendedRunStore implements SuspendedRunStore { * conditional advance — see {@link claimSuspension}'s docblock. */ private claimUnsupportedWarned = false; + /** + * [#15832] The in-flight or settled one-time capability probe — memoized as + * the PROMISE, not as its value, so two resumes arriving in the same tick + * take one probe between them rather than one each. + * + * ⛔ A REJECTED probe is deliberately un-memoized (see + * {@link probeCountedMultiDelete}): a store that was unreachable for one + * second must not answer for the life of the process. + */ + private countedMultiDelete?: Promise; /** * [#10101] Memoized shared platform-row organization resolver over the same * engine the rows are written through — answers "which column carries the @@ -356,7 +396,8 @@ export class ObjectStoreSuspendedRunStore implements SuspendedRunStore { * 1. an engine with no `delete()` at all — the same composition * {@link delete} already degrades on; * 2. a multi-row result that is not a count, where "did I win?" has no - * answer to read; + * answer to read — settled BEFORE the compare-and-set since #15832, by + * {@link probeCountedMultiDelete}; see the section below; * 3. ⚠️ a DRIVER with no `deleteMany` — which this method never sees as an * answer at all. `ObjectQL.delete` resolves the predicate route and then * finds no `deleteMany` to call, so it throws @@ -377,6 +418,61 @@ export class ObjectStoreSuspendedRunStore implements SuspendedRunStore { * already promises. ⛔ Deliberately NOT extended to {@link delete}'s warn one * screen up: that line predates this card, fires once per CONSUMPTION rather * than once per resume attempt, and is not this change's to re-shape. + * + * ## [#15832] Reason 2 is settled BEFORE the row is touched, and `'unsupported'` + * is no longer an answer this method can give AFTER it + * + * Until #15832 the shape test lived where the count arrives — i.e. one line + * after the compare-and-set had gone out. On an engine whose multi-delete + * does not resolve a count that made the refusal a LIE about its own effect: + * the conditional delete had been performed against the shared row and its + * verdict discarded, so a replica that ACTUALLY LOST (0 rows affected) was + * told `'unsupported'`, which {@link AutomationEngine.claimAdvance} reads as + * `unguarded`, and it resumed. That is the doubled side effect #14333 exists + * to prevent, occurring on the one composition that declares itself unable + * to prevent it. (The extra round-trip `forgetSuspendedRun` then issues is + * real too, and ⛔ it is not why this changed.) + * + * So the question is asked of a call that touches no suspension: + * {@link probeCountedMultiDelete}, once per store instance, down the very + * route the claim takes, against a predicate that matches no row. + * + * ⚠️ **What that probe cannot do, stated plainly.** It cannot promise what + * the NEXT call resolves to. `ObjectQL.delete` declares `Promise` + * (`packages/objectql/src/engine.ts`) and what surfaces is `opCtx.result`, + * which any middleware may rewrite per call — so "the engine returns a + * count" is not a fact the probe can establish, only a fact it can OBSERVE + * once. There is no read-only way to establish it either: the shape is + * undeclared at this boundary, and the only instrument that answers is a + * call down the same route. Closing THAT gap is #16033's — the count is + * contracted one layer down (`IDataDriver.deleteMany`, + * `packages/spec/src/contracts/data-driver.ts`, `Promise`) and + * erased to `any` at the engine boundary this store talks to. + * + * ⇒ The probe alone would therefore be exactly the "looks like it decides in + * advance" change it must not be. It is half of the fix, and the half it + * buys is the DECLARATION: an engine that cannot count is refused with + * nothing consumed, which is what makes `claimAdvance`'s `unguarded` + * reading true when it is taken. The other half is the residual, below. + * + * ⭐ **The residual is closed by RETIRING `'unsupported'` after the write.** + * If a probed-counting engine still resolves a non-count for a real claim, + * the compare-and-set is committed and its verdict is unrecoverable — a + * winner and a loser both see the row gone, so no follow-up read can tell + * them apart. That is not "no guarantee was offered"; it is UNKNOWN, which + * this store says the way every other unknown is said here: it THROWS. + * {@link AutomationEngine.claimAdvance} already catches exactly that and + * answers `STORE_UNAVAILABLE`, whose text is already written for this fact + * ("a failure can arrive after a committed delete"), and the resume is + * REFUSED rather than continued. ⇒ A replica that lost is never told it is + * unguarded once the row has been touched, whatever the probe concluded. + * + * ⛔ Refusing is not free and is not pretended to be: a claim that in fact + * WON is then stranded (the row is gone and this resume does not continue + * it) until an operator retries. That is the deliberate direction — #14333's + * whole premise is that a doubled side effect is the worse outcome — and it + * is reachable only on an engine that answers inconsistently between the + * probe and the claim, which no shipped composition does. */ async claimSuspension(runId: string, parkedAt: SuspensionParkedAt): Promise { if (typeof this.engine.delete !== 'function') { @@ -385,19 +481,118 @@ export class ObjectStoreSuspendedRunStore implements SuspendedRunStore { ); return 'unsupported'; } + // [#15832] BEFORE the row is touched. A rejection propagates: it means the + // question could not be asked at all, which `claimAdvance` already maps to + // STORE_UNAVAILABLE — the same answer the claim itself produced when it + // was the call that threw, and the same one a driver with no `deleteMany` + // produces (reason 3 above). + const capability = await this.countedMultiDeleteCapability(); + if (!capability.counted) { + this.warnClaimUnsupported( + `the data engine's multi-row delete resolved ${capability.observed}, not an affected-row count, so the ` + + `conditional advance for suspended run '${runId}' cannot be decided — NO delete was issued, so no ` + + `suspension was consumed and no claim verdict was discarded`, + ); + return 'unsupported'; + } const where: Record = { id: runId, node_id: parkedAt.nodeId }; if (parkedAt.correlation !== undefined) where.correlation = parkedAt.correlation; const affected = await this.engine.delete(TABLE, { where, multi: true, context: SYSTEM_CTX }); if (typeof affected !== 'number') { - this.warnClaimUnsupported( - `the data engine's multi-row delete resolved ${typeof affected}, not an affected-row count, so the ` + - `conditional advance for suspended run '${runId}' cannot be decided`, + // ⛔ NOT `'unsupported'`, and this is the whole of #15832. The + // compare-and-set is COMMITTED; its verdict is unreadable and + // unrecoverable. Answering "no guarantee is offered" here would hand a + // losing replica an `unguarded` resume having just consumed — or failed + // to consume — the row it is racing for. + throw new Error( + `the data engine's multi-row delete resolved ${typeof affected}, not an affected-row count, AFTER the ` + + `compare-and-set for suspended run '${runId}' had already been issued. This engine answered a count ` + + `for the capability probe, so the shape is inconsistent between calls and this claim's verdict is ` + + `UNRECOVERABLE — a winner and a loser both find the row gone. Refusing the resume rather than ` + + `guessing: retry once the engine answers consistently.`, ); - return 'unsupported'; } return affected > 0 ? 'claimed' : 'lost'; } + /** + * [#15832] Ask, ONCE per store instance, whether this engine's counted + * multi-delete route resolves a count — before anything is claimed on it. + * + * See {@link claimSuspension}'s docblock for why the question is asked at + * all, and for the boundary of what an answer here is worth. + * + * ## The probe is a mutation-SHAPED call that mutates nothing + * + * The route is the decision, not the statement: `ObjectQL.delete` sends a + * `where` carrying keys besides `id` together with `multi: true` to + * `driver.deleteMany`, and that is the only route whose result is a count. + * So the probe must be a delete — a `find` would answer about a different + * method. What it must NOT be is a delete that can consume anything, and it + * is not: the predicate is {@link CLAIM_CAPABILITY_PROBE_SENTINEL} in all + * three condition columns at once. + * + * What a zero-match multi-delete costs on the real engine, read off + * `ObjectQL.delete` rather than assumed: + * + * - **no row changes** — the predicate matches none; + * - **no hook dispatch** — the caller's `where` is present, so the + * unscoped-multi-write dispatch does not fire; the per-row before phase + * is explicitly "zero matched rows is zero dispatches"; and the per-row + * after phase iterates the same empty set; + * - **no realtime event** — `publishBulkDataEvent` returns at `matched === 0` + * with a `debug` line, precisely so an idle sweep is not a webhook + * delivery saying "0 records"; + * - **no summary recompute** — that is the by-id arm's, keyed on a pre-image + * this path never reads. + * + * ⇒ one `DELETE … WHERE` that matches nothing, once per process. It also + * happens to remove the redundant round-trip the old shape bought on every + * resume of an uncounted engine, but ⛔ that is a side benefit, not the + * reason. + * + * ⛔ A THROWN probe is not memoized. "The store was unreachable" is a + * transient fact and must not become this store's permanent answer; the + * rejection reaches {@link claimSuspension}'s caller, and the next resume + * asks again. A SETTLED probe — counted or not — is memoized either way, + * because that one is a property of the composition. + * + * ⛔ And a throwaway ROW is deliberately not inserted to make the probe + * match something. It would answer a strictly stronger question (that the + * count counts, not merely that it is a number), and `typeof affected !== + * 'number'` is the whole of the condition this store branches on — while + * buying an INSERT on a platform object and a stranded row whenever a + * process dies between the two statements. + */ + private countedMultiDeleteCapability(): Promise { + return this.countedMultiDelete ?? this.probeCountedMultiDelete(); + } + + private probeCountedMultiDelete(): Promise { + const pending = (async (): Promise => { + const affected = await this.engine.delete!(TABLE, { + where: { + id: CLAIM_CAPABILITY_PROBE_SENTINEL, + node_id: CLAIM_CAPABILITY_PROBE_SENTINEL, + correlation: CLAIM_CAPABILITY_PROBE_SENTINEL, + }, + multi: true, + context: SYSTEM_CTX, + }); + return typeof affected === 'number' ? { counted: true } : { counted: false, observed: typeof affected }; + })(); + // Memoized BEFORE it settles, so concurrent resumes share one probe. + this.countedMultiDelete = pending; + // Forget a REJECTED probe so the next resume re-asks. The `catch` is on a + // derived promise purely to keep the memoized rejection from surfacing as + // an unhandled one; `pending` itself is what callers await, so they still + // see the failure. + pending.catch(() => { + if (this.countedMultiDelete === pending) this.countedMultiDelete = undefined; + }); + return pending; + } + /** * [#14333] Say ONCE, per store instance, that this composition cannot * express the conditional advance. See {@link claimSuspension}'s docblock From 90f388825539475dee6513cef9cb9c1eca6cddac Mon Sep 17 00:00:00 2001 From: Claude Date: Sun, 6 Sep 2026 02:00:40 +0000 Subject: [PATCH 2/2] wip(#15832): preserve in-progress Note 2 pin work after a container restart NOT a finished change and NOT verified by the PM. The dispatched os-dev seat was extending the claim-probe pins and the engine-double ledger when the container restarted and killed it. This commit preserves that work; the merge of origin/main below it was the seat's own. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01XpTx2tbq3pZRYAdoGt6E6Y --- .../src/suspended-run-claim-probe.test.ts | 4 +++- scripts/engine-double-contract.pinned.json | 10 ++++++++++ 2 files changed, 13 insertions(+), 1 deletion(-) diff --git a/packages/services/service-automation/src/suspended-run-claim-probe.test.ts b/packages/services/service-automation/src/suspended-run-claim-probe.test.ts index 6f496a1a79..605092ec3d 100644 --- a/packages/services/service-automation/src/suspended-run-claim-probe.test.ts +++ b/packages/services/service-automation/src/suspended-run-claim-probe.test.ts @@ -90,7 +90,7 @@ import { SqlDriver } from '@objectstack/driver-sql'; // through it so it physically cannot accept a call `ObjectQL.delete` refuses // (`scripts/check-engine-double-contract.mjs`), and the cases read it directly // to say WHICH route a recorded call would have taken. -import { assertEngineDeleteDispatch } from '@objectstack/metadata-core'; +import { assertEngineDeleteDispatch, assertEngineUpdateDispatch } from '@objectstack/metadata-core'; import { AutomationEngine } from './engine.js'; import { AutomationServicePlugin } from './plugin.js'; @@ -124,6 +124,8 @@ function createProbeAwareEngine( }, async insert(_object, data) { rows.set(String(data.id), { ...data }); return data; }, async update(_object, data, options) { + // Refuses what a real server refuses (`check:engine-double-contract`). + assertEngineUpdateDispatch(data, options as any); const id = options?.where?.id ?? data.id; rows.set(String(id), { ...(rows.get(String(id)) ?? { id }), ...data }); return rows.get(String(id)); diff --git a/scripts/engine-double-contract.pinned.json b/scripts/engine-double-contract.pinned.json index 81ef3511f3..ef26ede04e 100644 --- a/scripts/engine-double-contract.pinned.json +++ b/scripts/engine-double-contract.pinned.json @@ -3381,6 +3381,16 @@ "verb": "update", "pinned": 1 }, + { + "file": "packages/services/service-automation/src/suspended-run-claim-probe.test.ts", + "verb": "delete", + "pinned": 1 + }, + { + "file": "packages/services/service-automation/src/suspended-run-claim-probe.test.ts", + "verb": "update", + "pinned": 1 + }, { "file": "packages/services/service-automation/src/suspended-run-store.test.ts", "verb": "delete",