diff --git a/.changeset/concurrent-replica-resume-advance-claim.md b/.changeset/concurrent-replica-resume-advance-claim.md new file mode 100644 index 0000000000..112aa499db --- /dev/null +++ b/.changeset/concurrent-replica-resume-advance-claim.md @@ -0,0 +1,44 @@ +--- +"@objectstack/service-automation": minor +--- + +fix(service-automation): make the resume advance a conditional claim on the suspended-run store, so two replicas cannot both advance one run (#14333) + +`AutomationEngine.resumeInternal` guarded a duplicate resume with `this.resuming`, +an in-process `Set`. That is a complete guard for exactly one deployment shape: a +single process. Behind a load balancer, two decisions on one run arriving in the +same instant on two replicas each passed their **own** `resuming` check, both read +the same fresh row out of the shared store, both consumed it, and both traversed +forward — so every downstream side effect ran twice. #13617 closed the sequential +half of this family (a replica resuming from a snapshot it had gone stale on); it +deliberately did not close the concurrent one. + +Measured before the fix on the two-engines-over-one-shared-store harness, at +`packages/services/service-automation/src/concurrent-replica-resume-race.test.ts`: +**25 of 25** raced runs advanced twice — one action fired twice and one approval +level opened twice per run — for both reachable shapes the report named (parallel / +any-of approvers, and duplicated automated approve calls). A single approver per +level deciding sequentially does **not** race, and is pinned as the negative +control. + +`SuspendedRunStore` therefore gains `claimSuspension(runId, parkedAt)`: consume the +durable record **only** if it is still parked at the node the caller read (and, when +the caller has one, still carrying that correlation), atomically, answering +`'claimed'` / `'lost'` / `'unsupported'`. The winner advances; the loser is refused +`RESUME_IN_PROGRESS` — the existing code, because the remedy is identical to the +in-process refusal's and `plugin-approvals` already branches on it that way — and +runs nothing. The per-process `resuming` set stays as the cheap first gate; it is +not replaced, and the single-replica path is unchanged. + +Both shipped stores implement it: `InMemorySuspendedRunStore` tests and removes +with no `await` between the two, and `ObjectStoreSuspendedRunStore` issues one +`DELETE … WHERE id = ? AND node_id = ?` through the data engine's documented +compare-and-set route (`multi: true` with a full `where`), reading the affected-row +count. ⛔ No platform-object schema change: `node_id` and `correlation` are columns +`sys_automation_run` already carried. + +The member is **optional**, so no existing implementation is broken, and its absence +is a declared degradation rather than a silent one: an engine whose store cannot +express the condition says once, at `warn`, that resume idempotency is in-process +only and what that costs — the same posture `AutomationEngine.claim()` already takes +when no persisted flow-dispatch ledger is attached. diff --git a/content/docs/permissions/tenant-audit-census.mdx b/content/docs/permissions/tenant-audit-census.mdx index 87c684e694..9d95e28107 100644 --- a/content/docs/permissions/tenant-audit-census.mdx +++ b/content/docs/permissions/tenant-audit-census.mdx @@ -98,7 +98,7 @@ are reported as `undecidable` rather than assumed either way. The same holds twice over for the context. An options argument spelled as a literal can be read; one spelled `options`, `{ ...opts }`, or handed through a -forwarding shim cannot, and **67 of the 217 sites are spelled that way**. A +forwarding shim cannot, and **67 of the 218 sites are spelled that way**. A context resolved from an inline literal or a local `const` can be tested for `isSystem`; one arriving from a helper call cannot. @@ -147,10 +147,10 @@ reproduce them. Where it disagrees, it disagrees on the page: | carried figure | where it survives | this census | | :--- | :--- | ---: | -| 175 write call sites | quoted in the merged changeset | **217** | +| 175 write call sites | quoted in the merged changeset | **218** | | 24 carrying no tenant context | quoted in the merged changeset | **9** provable and tenancy-enabled; **32** more whose options argument is unreadable | -| 127 of 175 statically decidable, 48 runtime-parameter-name sites | restated on the `isSystem`-scoping card | **145 of 217** decidable, **72** undecidable | -| 135 (77%) silenced by the `isSystem` guard before the posture gate | the lost issue body — **no surviving corroboration** | **not reproduced**: 99 decidably elevated, 0 decidably not, 101 undecidable | +| 127 of 175 statically decidable, 48 runtime-parameter-name sites | restated on the `isSystem`-scoping card | **146 of 218** decidable, **72** undecidable | +| 135 (77%) silenced by the `isSystem` guard before the posture gate | the lost issue body — **no surviving corroboration** | **not reproduced**: 100 decidably elevated, 0 decidably not, 101 undecidable | | 141 and 132, two independent re-derivations | the card that filed this work | — | **The differences are not reconciled, and deliberately so.** The old census's @@ -161,17 +161,17 @@ at any commit. Two structural facts do plausibly widen this reading against any hand or regex one, and both are counted in the generated tables below: the 45 sites reached -through an erased (`any`) receiver, and the 37 that name their object through a +through an erased (`any`) receiver, and the 38 that name their object through a `const` rather than inline. An instrument that read either the way a person does would report a smaller number and would not say so. The fourth row is the one worth flagging to anyone citing it. **The 135 / 77% figure has no surviving corroboration anywhere in the tree.** This census reads -99 of 217 (45%) as decidably elevated, with 101 more whose elevation is a +100 of 218 (46%) as decidably elevated, with 101 more whose elevation is a run-time fact — so the claim is neither confirmed nor refuted, and the honest answer is that a static reading cannot settle it. -⇒ **Cite `9 / 217`, and say what it is**: the sites whose options argument was +⇒ **Cite `9 / 218`, and say what it is**: the sites whose options argument was READ and holds no tenant context, against a decidably tenancy-enabled object. That is the control's provable yield surface. ⛔ Do not cite it as "the sites without tenant context" — **32 further sites** have an options argument this @@ -183,29 +183,29 @@ cannot read, and they are neither in nor out. | what | count | | :--- | ---: | -| write call sites on the application surface | **217** | -| …whose object name is statically decidable | 145 | +| write call sites on the application surface | **218** | +| …whose object name is statically decidable | 146 | | …whose object name is chosen at run time | 72 | -| …against an object with tenancy ENABLED | 145 | +| …against an object with tenancy ENABLED | 146 | | …against an object that declares tenancy off | 0 | -| threading a tenant context | 133 | +| threading a tenant context | 134 | | PROVABLY carrying none (options read, no context key) | **17** | | …of those, against a decidably tenancy-enabled object | **9** | | options argument UNREADABLE — may or may not carry one | 67 | | …of those, against a decidably tenancy-enabled object | 32 | -| threading a decidably ELEVATED (`isSystem`) context | 99 | +| threading a decidably ELEVATED (`isSystem`) context | 100 | | threading a context that is decidably NOT elevated | 0 | | threading a context whose elevation is a run-time fact | 101 | | how the instrument reached the site | count | | :--- | ---: | -| receiver carried a readable engine type | 172 | +| receiver carried a readable engine type | 173 | | receiver erased, placed by the object NAME | 19 | | receiver erased, placed by an `object: string` PARAMETER | 15 | | receiver erased, placed by an `UNTYPED_RECEIVERS` row | 11 | | object name spelled inline | 108 | -| object name spelled through a `const` | 37 | +| object name spelled through a `const` | 38 | | object name is an `object: string` parameter | 19 | | object name is some other run-time expression | 53 | @@ -224,13 +224,13 @@ holds still. They are required to be HERE and to say WHEN they were true; their values are not compared. The reasoning, and the measurement behind it, are in `scripts/check-tenant-audit-census.mjs`. -Measured on 2026-09-01 at `d3ebf3b55`. +Measured on 2026-09-02 at `5daab8df0`. | corpus scale (not enforced) | count | | :--- | ---: | -| tracked non-test sources scanned | 534 | +| tracked non-test sources scanned | 539 | | engine-shaped types recognised | 56 | | declared objects in the registry | 297 | -| same-named calls subtracted as non-engine | 119 | +| same-named calls subtracted as non-engine | 130 | {/* END GENERATED: tenant-audit-census */} diff --git a/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md b/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md index f9458b7a03..fc16789018 100644 --- a/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md +++ b/docs/audits/2026-08-tenant-audit-write-call-sites.counts.md @@ -29,17 +29,17 @@ silent, and `node scripts/tenant-audit-census.mjs --write` is the resolution. | Measure | Value | |---|---:| -| Write call sites | 217 | -| Object name statically decidable | 145 | +| Write call sites | 218 | +| Object name statically decidable | 146 | | Object name chosen at run time | 72 | -| Against a tenancy-enabled object | 145 | +| Against a tenancy-enabled object | 146 | | Against an object declaring tenancy off | 0 | -| Threading a tenant context | 133 | +| Threading a tenant context | 134 | | Provably carrying none | 17 | | …and decidably tenancy-enabled | 9 | | Options argument unreadable | 67 | | …and decidably tenancy-enabled | 32 | -| Threading a decidably elevated context | 99 | +| Threading a decidably elevated context | 100 | | Threading a decidably non-elevated context | 0 | | Threading a context of undecidable elevation | 101 | @@ -52,14 +52,14 @@ holds still. They are required to be HERE and to say WHEN they were true; their values are not compared. The reasoning, and the measurement behind it, are in `scripts/check-tenant-audit-census.mjs`. -Measured on 2026-09-01 at `d3ebf3b55`. +Measured on 2026-09-02 at `5daab8df0`. | corpus scale (not enforced) | count | | :--- | ---: | -| tracked non-test sources scanned | 534 | +| tracked non-test sources scanned | 539 | | engine-shaped types recognised | 56 | | declared objects in the registry | 297 | -| same-named calls subtracted as non-engine | 119 | +| same-named calls subtracted as non-engine | 130 | ## Every site @@ -158,7 +158,7 @@ Measured on 2026-09-01 at `d3ebf3b55`. | `packages/services/service-automation/src/builtin/crud-nodes.ts` | `insert` | `objectName` | undecidable | context, elevation undecidable | 1 | | `packages/services/service-automation/src/builtin/crud-nodes.ts` | `update` | `objectName` | undecidable | context, elevation undecidable | 1 | | `packages/services/service-automation/src/flow-dispatch-store.ts` | `insert` | `sys_flow_dispatch` | enabled | elevated | 1 | -| `packages/services/service-automation/src/suspended-run-store.ts` | `delete` | `sys_automation_run` | enabled | elevated | 2 | +| `packages/services/service-automation/src/suspended-run-store.ts` | `delete` | `sys_automation_run` | enabled | elevated | 3 | | `packages/services/service-automation/src/suspended-run-store.ts` | `insert` | `sys_automation_run` | enabled | elevated | 2 | | `packages/services/service-automation/src/suspended-run-store.ts` | `update` | `sys_automation_run` | enabled | elevated | 2 | | `packages/services/service-datasource/src/datasource-admin-plugin.ts` | `delete` | `sys_metadata` | enabled | PROVABLY NONE | 1 | diff --git a/packages/services/service-automation/src/concurrent-replica-resume-race.test.ts b/packages/services/service-automation/src/concurrent-replica-resume-race.test.ts new file mode 100644 index 0000000000..fec1f3d61c --- /dev/null +++ b/packages/services/service-automation/src/concurrent-replica-resume-race.test.ts @@ -0,0 +1,613 @@ +// Copyright (c) 2026 ObjectStack. Licensed under the Apache-2.0 license. + +/** + * #14333 — two CONCURRENT resumes of one run, on two replicas over one shared + * store, must advance it exactly once. + * + * ## Why this is a sibling file and not more cases in the staleness harness + * + * `multi-replica-resume-staleness.test.ts` (#13617) is the two-engines-over- + * one-shared-store harness of record, and it carries a REVERT-PROOF ledger + * keyed to its own mutation: restore the cache-first read at the top of + * `loadSuspendedRunStrict` and that file goes exactly `4 red / 4 green`, case + * by named case. Adding cases there would silently invalidate those counts — + * the ledger would describe a file that no longer exists — so the concurrent + * half lives here, off its own mutation, and that file's ledger stays true. + * + * ## The defect this file measures + * + * #13617 closed the SEQUENTIAL failure (a replica resuming from a snapshot it + * had gone stale on). It did not close the CONCURRENT one. `resumeInternal` + * guards a duplicate resume with `this.resuming`, an in-process `Set`. Two + * decisions on ONE run arriving in the same instant on TWO replicas each pass + * their OWN `resuming` check, both read the same fresh row out of the shared + * store, both consume it, and both traverse forward — so every downstream side + * effect runs twice. + * + * ## What "both advanced" is asserted on + * + * OBSERVABLE EFFECTS, never internal bookkeeping. The flow below puts a real + * side-effect node between the two approval levels: + * + * start -> lv1 (pauses) -> notify (fires) -> lv2 (pauses) -> end + * + * so one advance past `lv1` fires `notify` once and opens `lv2` once. The + * defect's signature is `notify` in the `fired` ledger TWICE and `lv2` in the + * `opened` ledger twice — an action fired twice and a node executed twice, + * which is the `sys_approval_request` duplication the family was reported as. + * "Both callers entered `resumeInternal`" is deliberately NOT the assertion. + * + * ## The remedy this file pins + * + * A conditional advance on `SuspendedRunStore` — `claimSuspension`, "consume + * the row only if it is still parked at node N with correlation C" — so the + * winner advances and the loser is told `RESUME_IN_PROGRESS` and runs nothing. + * The per-process `this.resuming` stays as the cheap first gate; it is not + * replaced. + * + * ## REVERT-PROOF — three mutations, all measured on the committed tree + * + * Each was confirmed ON DISK before a single result was read (anchored counts + * plus the blob hash) and restored inside a `trap ... EXIT INT TERM`, with the + * restore proven by an empty `git diff HEAD` and a blob hash equal to HEAD's. + * The population is these three files: this one, `suspended-run-store.test.ts` + * and `multi-replica-resume-staleness.test.ts` — 61 tests. + * + * **(E) the engine stops asking.** Replace the `claimAdvance` call in + * `resumeInternal` with the unconditional `forgetSuspendedRun(run, 'resumed')` + * it had before this card: `Tests 9 failed | 52 passed (61)`. Seven here + * (SHAPE A and SHAPE B on `expected [ 'notify', 'notify' ] to deeply equal + * [ 'notify' ]`, SIZED on `{ trials: 25, doubled: 25, extraOpens: 25 }`, both + * CONDITION cases, the loser's `debug` trace and the declared degradation) and + * two in `suspended-run-store.test.ts` (the two-engines race over the durable + * store, and the throwing claim). `multi-replica-resume-staleness.test.ts` + * stays 8/8: the mutation is targeted, and #13617's own ledger is untouched. + * + * **(C) the condition stops being a condition.** Delete BOTH comparisons from + * `InMemorySuspendedRunStore.claimSuspension`, leaving an existence-only + * consume: `Tests 2 failed | 59 passed (61)` — exactly the two CONDITION cases + * below, and nothing else. That is the point of them. Every other race in this + * file lets the loser lose by finding no row at all, which an existence check + * satisfies too; before those two existed this mutation was measured GREEN + * across the whole branch. + * + * **(C2) the production store loses its predicate.** Delete `multi: true` from + * the one `delete` call in `ObjectStoreSuspendedRunStore.claimSuspension`: + * `Tests 7 failed | 54 passed (61)`, every one of them in + * `suspended-run-store.test.ts`, failing with the PRODUCER's own refusal — + * "Delete names one row by primary key, but options.where also carries + * predicate keys 'node_id', 'correlation' ... For a conditional + * (compare-and-set) write, declare the predicate path". Against a running + * server that spelling throws and `claimAdvance` turns it into + * `STORE_UNAVAILABLE` on EVERY resume; before that suite existed, this + * one-token regression was measured GREEN across the whole branch. + */ + +import { describe, it, expect } from 'vitest'; +import { defineActionDescriptor } from '@objectstack/spec/automation'; +import { RESUME_AUTHORITY_SERVICE } from '@objectstack/spec/contracts'; +import { AutomationEngine } from './engine.js'; +import { InMemorySuspendedRunStore } from './suspended-run-store.js'; +import type { SuspendedRun, SuspendedRunStore } from './engine.js'; + +function silentLogger(): any { + return { info() {}, warn() {}, error() {}, debug() {}, child() { return silentLogger(); } }; +} + +/** A logger that keeps every line, so a degradation can be read back. */ +function capturingLogger(lines: string[]): any { + const l: any = { + info: (m: string) => lines.push(`info ${m}`), + warn: (m: string) => lines.push(`warn ${m}`), + error: (m: string) => lines.push(`error ${m}`), + debug: (m: string) => lines.push(`debug ${m}`), + child: () => l, + }; + return l; +} + +/** + * Two approval levels with a real side effect between them, so ONE advance + * past `lv1` is distinguishable from two by what ran, not by what was called. + */ +const APPROVAL_FLOW = { + 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: 'lv2', type: 'approval_level', label: 'General manager' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'lv1' }, + { id: 'e2', source: 'lv1', target: 'notify' }, + { id: 'e3', source: 'notify', target: 'lv2' }, + { id: 'e4', source: 'lv2', target: 'end' }, + ], +} as any; + +/** The shared ledgers a run writes into, standing in for real side effects. */ +interface Ledgers { + /** Approval levels opened — one `sys_approval_request` row each. */ + opened: string[]; + /** Downstream actions fired — the doubled effect this card is about. */ + fired: string[]; +} + +/** + * One replica: a fresh engine over the SHARED store, appending to the SHARED + * ledgers. Two replicas therefore see one another's effects, exactly as two + * app instances over one database see one another's rows. + */ +function replica(store: SuspendedRunStore | undefined, led: Ledgers, logger = silentLogger()): AutomationEngine { + 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) { + led.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) { + led.fired.push(node.id); + return { success: true }; + }, + }); + engine.registerFlow('expense_approval', APPROVAL_FLOW); + return engine; +} + +const ledgers = (): Ledgers => ({ opened: [], fired: [] }); + +/** The approve an approvals service issues once it has recorded a decision. */ +function approve(engine: AutomationEngine, runId: string, signal: Record = {}) { + return engine.resume(runId, { ...signal, [RESUME_AUTHORITY_SERVICE]: true } as any); +} + +/** Park a fresh run at `lv1` on `submitter`, returning its id. */ +async function parkAtLv1(submitter: AutomationEngine): Promise { + const submitted = await submitter.execute('expense_approval'); + expect(submitted.status).toBe('paused'); + return submitted.runId!; +} + +describe('#14333 concurrent resumes of one run on two replicas advance it exactly once', () => { + // ── SHAPE A: parallel / any-of approvers ──────────────────────────────── + // + // Two DIFFERENT approvers on the same level, both authorized to advance it + // (any-of), click approve in the same instant. The load balancer puts one + // decision on replica A and the other on replica B. Each engine's own + // `resuming` set is empty, so neither sees the other. + + it('SHAPE A (any-of approvers): the level advances once and fires the action once', async () => { + const store = new InMemorySuspendedRunStore(); + const led = ledgers(); + const a = replica(store, led); + const b = replica(store, led); + + const runId = await parkAtLv1(a); + expect(led.opened).toEqual(['lv1']); + + const [byAlice, byBob] = await Promise.all([ + approve(a, runId, { approver: 'alice' }), + approve(b, runId, { approver: 'bob' }), + ]); + + // THE OBSERVABLE EFFECT. One advance past `lv1`: the action fires once + // and exactly one `lv2` request is opened. Unguarded, this reads + // `['notify','notify']` and `['lv1','lv2','lv2']`. + expect(led.fired).toEqual(['notify']); + expect(led.opened).toEqual(['lv1', 'lv2']); + + // Exactly one winner, and the loser is TOLD — not silently dropped. + const winners = [byAlice, byBob].filter((r) => r.success); + const losers = [byAlice, byBob].filter((r) => !r.success); + expect(winners).toHaveLength(1); + expect(losers).toHaveLength(1); + expect(winners[0]!.status).toBe('paused'); + expect(losers[0]!.code).toBe('RESUME_IN_PROGRESS'); + // A refusal carries no run status — nothing was dispatched for it. + expect(losers[0]!.status).toBeUndefined(); + + // And the run is still one run, parked once, as either replica reads it. + expect(await a.hasSuspendedRun(runId)).toBe(true); + expect(await b.hasSuspendedRun(runId)).toBe(true); + }); + + // ── SHAPE B: automated approve calls ──────────────────────────────────── + // + // No human at all: an auto-approve rule (or an at-least-once delivery of + // one decision) issues the SAME approve twice, microseconds apart, landing + // on two replicas. Identical signals, so nothing downstream could tell the + // duplicate from the original after the fact. + + it('SHAPE B (automated approve calls): a duplicated automated approve advances once', async () => { + const store = new InMemorySuspendedRunStore(); + const led = ledgers(); + const a = replica(store, led); + const b = replica(store, led); + + const runId = await parkAtLv1(a); + const automated = { approver: 'auto_rule_expense_under_1000' }; + + const [first, second] = await Promise.all([ + approve(a, runId, automated), + approve(b, runId, automated), + ]); + + expect(led.fired).toEqual(['notify']); + expect(led.opened).toEqual(['lv1', 'lv2']); + expect([first, second].filter((r) => r.success)).toHaveLength(1); + expect([first, second].filter((r) => r.code === 'RESUME_IN_PROGRESS')).toHaveLength(1); + }); + + // ── The exposure, sized ───────────────────────────────────────────────── + // + // How OFTEN, not just whether. Every trial is one run raced by two + // replicas; `doubled` counts the trials whose action fired more than once. + // The failure message carries the measurement, so the number is recorded + // by the run itself rather than by a comment. + + it('SIZED: 25 raced runs produce 25 single advances and zero doubled effects', async () => { + const TRIALS = 25; + let doubled = 0; + let extraOpens = 0; + for (let i = 0; i < TRIALS; i++) { + const store = new InMemorySuspendedRunStore(); + const led = ledgers(); + const a = replica(store, led); + const b = replica(store, led); + const runId = await parkAtLv1(a); + await Promise.all([approve(a, runId), approve(b, runId)]); + if (led.fired.length > 1) doubled++; + extraOpens += led.opened.filter((id) => id === 'lv2').length - 1; + } + expect({ trials: TRIALS, doubled, extraOpens }).toEqual({ trials: TRIALS, doubled: 0, extraOpens: 0 }); + }); + + // ── NEGATIVE CONTROL: the shape the card called not obviously reachable ── + // + // A single approver per level, deciding one at a time: the second decision + // is only issued after the first has returned. There is no window in which + // two resumes overlap, so this cannot race however the replicas are + // scheduled — and the second call finds the pause already consumed. + + it('NEGATIVE CONTROL: a single approver per level, sequential, cannot race', async () => { + const store = new InMemorySuspendedRunStore(); + const led = ledgers(); + const a = replica(store, led); + const b = replica(store, led); + + const runId = await parkAtLv1(a); + + // Decision 1, fully returned before decision 2 is issued. + const first = await approve(a, runId); + expect(first.success).toBe(true); + expect(first.status).toBe('paused'); + expect(led.fired).toEqual(['notify']); + expect(led.opened).toEqual(['lv1', 'lv2']); + + // Decision 2 for `lv2`, on the other replica — advances, once. + const second = await approve(b, runId); + expect(second.success).toBe(true); + expect(second.status).toBeUndefined(); + expect(led.fired).toEqual(['notify']); + expect(led.opened).toEqual(['lv1', 'lv2']); + }); + + // ── NEGATIVE CONTROL: the single-replica path is untouched ────────────── + + it('NEGATIVE CONTROL: one replica, two concurrent approves — the in-process guard still refuses one', async () => { + const store = new InMemorySuspendedRunStore(); + const led = ledgers(); + const only = replica(store, led); + + const runId = await parkAtLv1(only); + const [first, second] = await Promise.all([approve(only, runId), approve(only, runId)]); + + expect(led.fired).toEqual(['notify']); + expect(led.opened).toEqual(['lv1', 'lv2']); + expect([first, second].filter((r) => r.success)).toHaveLength(1); + const refused = [first, second].find((r) => !r.success)!; + // `this.resuming` — the cheap first gate, kept, not replaced. + expect(refused.code).toBe('RESUME_IN_PROGRESS'); + expect(refused.error).toContain('already being resumed'); + }); + + it('NEGATIVE CONTROL: one replica, sequential approves walk the flow to completion', async () => { + const store = new InMemorySuspendedRunStore(); + const led = ledgers(); + const only = replica(store, led); + + const runId = await parkAtLv1(only); + expect((await approve(only, runId)).status).toBe('paused'); + const final = await approve(only, runId); + + expect(final.success).toBe(true); + expect(final.status).toBeUndefined(); + expect(led.opened).toEqual(['lv1', 'lv2']); + expect(led.fired).toEqual(['notify']); + expect(await only.hasSuspendedRun(runId)).toBe(false); + }); + + it('NEGATIVE CONTROL: with no store at all, behaviour is purely in-memory', async () => { + // The historical default (LiteKernel, tests, dev): nothing is shared, + // so there is no second replica to race and the in-process guard is + // the whole guarantee. No degradation is declared here — there is no + // cross-replica guarantee to weaken. + const led = ledgers(); + const lines: string[] = []; + const solo = replica(undefined, led, capturingLogger(lines)); + + const runId = await parkAtLv1(solo); + expect((await approve(solo, runId)).status).toBe('paused'); + expect(led.opened).toEqual(['lv1', 'lv2']); + expect(led.fired).toEqual(['notify']); + expect(lines.filter((l) => l.includes('advance guarantee'))).toEqual([]); + }); + + // ── THE CONDITION ITSELF: "still parked at node N", not "still present" ── + // + // Everything above races two claims that arrive while the row is still at + // the node both replicas read, so the LOSER loses by finding no row at all. + // That cannot tell a compare-and-set from an existence check — measured: + // with both comparisons deleted from `InMemorySuspendedRunStore. + // claimSuspension`, every test on the branch stayed green. + // + // The window the condition actually closes is the other one: a loser whose + // claim lands AFTER the winner has already advanced and RE-PARKED. Now a + // row exists again, at a different parking, and an existence-only consume + // deletes the parking another replica is standing on and traverses forward + // from a snapshot that is two beats stale. The two tests below hold the + // loser's claim until exactly that moment — one per comparison, so a + // mutation that deletes only one of them still reddens. + + /** A promise with its resolver, for holding a claim open. */ + function latch(): { held: Promise; release: () => void } { + let release!: () => void; + const held = new Promise((r) => { release = () => r(); }); + return { held, release }; + } + + /** + * A replica's own view of the SHARED store — the real topology: two + * processes, one database, each process holding its own client. Only the + * TIMING of `claimSuspension` is this wrapper's business; the claim itself + * is the shared store's, taken against the shared row. + */ + function delayedClaimClient( + shared: InMemorySuspendedRunStore, + held: Promise, + seen: Array<{ nodeId: string; correlation?: string }>, + ): SuspendedRunStore { + return { + save: (r: SuspendedRun) => shared.save(r), + load: (id: string) => shared.load(id), + delete: (id: string) => shared.delete(id), + list: () => shared.list(), + async claimSuspension(runId, parkedAt) { + seen.push({ ...parkedAt }); + await held; + return shared.claimSuspension(runId, parkedAt); + }, + }; + } + + /** The same view with no delay — the winner's client. */ + function client(shared: InMemorySuspendedRunStore): SuspendedRunStore { + return { + save: (r: SuspendedRun) => shared.save(r), + load: (id: string) => shared.load(id), + delete: (id: string) => shared.delete(id), + list: () => shared.list(), + claimSuspension: (id, at) => shared.claimSuspension(id, at), + }; + } + + it('THE CONDITION (node): a claim landing after the winner RE-PARKED is lost, not granted', async () => { + const shared = new InMemorySuspendedRunStore(); + const led = ledgers(); + const gate = latch(); + const seen: Array<{ nodeId: string; correlation?: string }> = []; + const a = replica(client(shared), led); + const b = replica(delayedClaimClient(shared, gate.held, seen), led); + + const runId = await parkAtLv1(a); + expect(led.opened).toEqual(['lv1']); + + // B's decision arrives first and reads the run at `lv1` — then its claim + // stalls (a slow client, a queued statement, a paused container). + const bDecision = approve(b, runId); + + // A's decision lands and completes: `notify` fires once and the run + // RE-PARKS at `lv2`. A row for this run exists again. + const aResult = await approve(a, runId); + expect(aResult.success).toBe(true); + expect(aResult.status).toBe('paused'); + expect(led.fired).toEqual(['notify']); + expect(led.opened).toEqual(['lv1', 'lv2']); + + // The precondition this test rests on, asserted rather than assumed: B + // is claiming the parking it READ, which is the one A has left. + expect(seen).toEqual([{ nodeId: 'lv1', correlation: 'req_lv1' }]); + + // Now B's claim reaches the store. + gate.release(); + const bResult = await bDecision; + + // It must LOSE. Existence-only, it wins: the `lv2` row is present, so + // it is deleted and B traverses forward from its stale `lv1` snapshot. + expect(bResult.success).toBe(false); + expect(bResult.code).toBe('RESUME_IN_PROGRESS'); + expect(bResult.status).toBeUndefined(); + + // The observable effects are unchanged by the loser: one `notify`, one + // `lv2`. Existence-only reads `[ 'notify', 'notify' ]` here. + expect(led.fired).toEqual(['notify']); + expect(led.opened).toEqual(['lv1', 'lv2']); + + // ⛔ And the winner's parking SURVIVED. This is the half a doubled + // effect alone would not catch: an existence-only consume destroys the + // live `lv2` suspension, stranding the run for good. + expect(await a.hasSuspendedRun(runId)).toBe(true); + expect(await b.hasSuspendedRun(runId)).toBe(true); + const stored = await shared.load(runId); + expect(stored?.nodeId).toBe('lv2'); + }); + + it('THE CONDITION (correlation): a re-entry at the SAME node with a new correlation is lost', async () => { + // The one shape where a run legitimately re-parks at the node it just + // left: `map` re-entry, whose correlation carries the child run id + // (`map:`), so the node id alone cannot separate the + // parking a replica read from the parking that replaced it. Only the + // correlation comparison can. + const shared = new InMemorySuspendedRunStore(); + const gate = latch(); + const seen: Array<{ nodeId: string; correlation?: string }> = []; + // Each re-entry records the item it just finished, then parks for the + // next one — so an item recorded twice is the doubled effect here. + const done: string[] = []; + const items = ['item_1', 'item_2', 'item_3']; + + function mapReplica(store: SuspendedRunStore): AutomationEngine { + const engine = new AutomationEngine(silentLogger(), store); + engine.registerNodeExecutor({ + type: 'map_items', + descriptor: defineActionDescriptor({ + type: 'map_items', version: '1.0.0', name: 'Map items', + supportsPause: true, resumeAuthority: 'service', + }), + async execute() { + const next = items[done.length]; + if (next === undefined) return { success: true }; + done.push(next); + return { success: true, suspend: true, correlation: `map:${next}` }; + }, + }); + engine.registerFlow('sweep', { + name: 'sweep', + label: 'Sweep', + type: 'autolaunched', + nodes: [ + { id: 'start', type: 'start', label: 'Start' }, + { id: 'sweep_items', type: 'map_items', label: 'Sweep items' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'sweep_items' }, + { id: 'e2', source: 'sweep_items', target: 'end' }, + ], + } as any); + return engine; + } + + const a = mapReplica(client(shared)); + const b = mapReplica(delayedClaimClient(shared, gate.held, seen)); + + const runId = (await a.execute('sweep')).runId!; + expect(done).toEqual(['item_1']); + + // B reads the run parked at `sweep_items` / `map:item_1`, then stalls. + const bDecision = approve(b, runId); + + // A completes the item: the node RE-RUNS and re-parks at the SAME node + // with a NEW correlation. + const aResult = await approve(a, runId); + expect(aResult.status).toBe('paused'); + expect(done).toEqual(['item_1', 'item_2']); + expect((await shared.load(runId))?.nodeId).toBe('sweep_items'); + expect((await shared.load(runId))?.correlation).toBe('map:item_2'); + expect(seen).toEqual([{ nodeId: 'sweep_items', correlation: 'map:item_1' }]); + + gate.release(); + const bResult = await bDecision; + + // Node id alone says "still parked at sweep_items" and would GRANT it. + // The correlation is the whole difference. + expect(bResult.success).toBe(false); + expect(bResult.code).toBe('RESUME_IN_PROGRESS'); + // No item swept twice, and item_3 was not pulled forward. + expect(done).toEqual(['item_1', 'item_2']); + // The live parking survived. + expect((await shared.load(runId))?.correlation).toBe('map:item_2'); + }); + + it('the loser is traced at DEBUG — an ordinary outcome, not a degradation', async () => { + // The caller is told in the result, so this line exists only for an + // operator reconstructing a race. At `warn` a busy any-of level would + // emit a steady stream of records describing correct behaviour. + const store = new InMemorySuspendedRunStore(); + const led = ledgers(); + const lines: string[] = []; + const a = replica(store, led, capturingLogger(lines)); + const b = replica(store, led, capturingLogger(lines)); + + const runId = await parkAtLv1(a); + await Promise.all([approve(a, runId), approve(b, runId)]); + + const claimLines = lines.filter((l) => l.includes('lost the advance claim')); + expect(claimLines).toHaveLength(1); + expect(claimLines[0]!.startsWith('debug ')).toBe(true); + expect(claimLines[0]).toContain("at node 'lv1'"); + // ⛔ Not a degradation: nothing about the advance claim is raised to + // `warn` or `error` on this path. (Scoped to the claim family on + // purpose — a directly-constructed engine also emits the unrelated + // #4792 "node-type vocabulary was never sealed" warn, and widening + // this to "no warn at all" would pin that instead of this.) + const raised = lines.filter( + (l) => (l.startsWith('warn ') || l.startsWith('error ')) && + (l.includes('advance claim') || l.includes('advance guarantee')), + ); + expect(raised).toEqual([]); + }); + + // ── The seam is LOUD when a store cannot express the guarantee ────────── + + it('a store without claimSuspension gets a ONE-TIME declared degradation, and still resumes', async () => { + // Third-party stores predating this member cannot express the + // conditional advance. They must not silently offer no guarantee: the + // engine says so once per store, at `warn` (a weakened guarantee is a + // FUNCTIONAL degradation — nothing claimed-persisted fails to land — + // the same call `claim()`'s missing-ledger branch makes one screen up). + const led = ledgers(); + const lines: string[] = []; + const backing = new InMemorySuspendedRunStore(); + // A store predating `claimSuspension`: the four original members and + // nothing else. Delegated to a real one so the run genuinely resumes — + // the point is that it resumes UNGUARDED and is told so, not that it + // breaks. + const store: SuspendedRunStore = { + save: (r: SuspendedRun) => backing.save(r), + load: (id: string) => backing.load(id), + delete: (id: string) => backing.delete(id), + list: () => backing.list(), + }; + const engine = replica(store, led, capturingLogger(lines)); + + const runId = await parkAtLv1(engine); + expect((await approve(engine, runId)).status).toBe('paused'); + expect((await approve(engine, runId)).status).toBeUndefined(); + + const declared = lines.filter((l) => l.includes('advance guarantee')); + expect(declared).toHaveLength(1); + expect(declared[0]!.startsWith('warn ')).toBe(true); + expect(declared[0]).toContain('claimSuspension'); + }); +}); diff --git a/packages/services/service-automation/src/engine.ts b/packages/services/service-automation/src/engine.ts index f2c33d8b57..cc6fbf5a8b 100644 --- a/packages/services/service-automation/src/engine.ts +++ b/packages/services/service-automation/src/engine.ts @@ -1323,6 +1323,56 @@ export interface SuspensionRestoreResult { consumedAt?: string; } +/** + * [#14333] Where a suspension was parked when the caller READ it — the + * condition a {@link SuspendedRunStore.claimSuspension} compare-and-set is + * taken against. + * + * Both fields come off the {@link SuspendedRun} the caller loaded, never off + * live engine state: the whole point is to ask the shared store whether the + * row still says what this replica last saw it say. + */ +export interface SuspensionParkedAt { + /** The node the row said the run was parked at. */ + nodeId: string; + /** + * The correlation the pausing executor minted, when the row carries one. + * A store compares it only when it is set: a row persisted with no + * correlation has nothing to compare, and the node condition still holds. + */ + correlation?: string; +} + +/** + * [#14333] What {@link SuspendedRunStore.claimSuspension} answers. + * + * Three values, not a boolean, because "I cannot express this condition" is a + * different fact from "you lost" — and an implementation that answered `false` + * for it would stall every resume, while one that answered `true` would offer + * no guarantee while claiming one. The third value is how a store says so out + * loud; the engine turns it into a declared, one-time degradation. + */ +export type SuspensionClaimOutcome = + /** THIS call consumed the durable row — the caller owns the advance. */ + | 'claimed' + /** The row was already gone, or had moved on — another advance won it. */ + | 'lost' + /** This store cannot express the condition; no guarantee was offered. */ + | 'unsupported'; + +/** + * [#14333] What {@link AutomationEngine.claimAdvance} concluded — the engine's + * reading of a {@link SuspensionClaimOutcome} once the cases a store never + * sees are folded in (no store attached, a run the store never accepted, a + * store that cannot express the condition, a store that could not be reached). + * Module-private: it is a control-flow verdict, not a contract. + */ +type AdvanceClaim = + | { kind: 'claimed' } + | { kind: 'lost' } + | { kind: 'unguarded' } + | { kind: 'unavailable'; message: string }; + export interface SuspendedRunStore { /** Persist (insert or replace) a suspended run. */ save(run: SuspendedRun): Promise; @@ -1332,6 +1382,31 @@ export interface SuspendedRunStore { delete(runId: string): Promise; /** List all currently-stored suspended runs. */ list(): Promise; + /** + * [#14333] CONDITIONALLY consume a suspension: remove the durable record + * only if it is still parked where the caller read it — "delete only if + * still at node N". + * + * This is the cross-replica half of the resume idempotency guard. The + * engine's own `resuming` set is per-PROCESS, so two decisions on one run + * arriving in the same instant on two replicas each pass their own check, + * both read the same fresh row, and both traverse forward — running every + * downstream side effect twice. The compare-and-set belongs where the + * shared state lives, so it lives here. + * + * The condition, normatively: the stored row for `runId` is still parked + * at `parkedAt.nodeId`, AND — when `parkedAt.correlation` is set — still + * carries that correlation. An implementation MUST perform the test and + * the removal as ONE atomic operation against the shared state; a read + * followed by a separate delete re-opens the very window this closes. + * + * OPTIONAL, and its absence is a DECLARED degradation rather than a silent + * one: the engine says once per store that the advance guarantee is + * in-process only and then resumes exactly as it did before this member + * existed. Same posture as {@link FlowDispatchStore}'s missing-ledger + * branch — a weakened guarantee is announced, never assumed. + */ + claimSuspension?(runId: string, parkedAt: SuspensionParkedAt): Promise; /** * Persist a TERMINAL run (completed / failed) as durable history for the * "Runs" observability surface. Optional — the in-memory / test defaults @@ -1729,6 +1804,15 @@ export class AutomationEngine implements IAutomationService { * weakened guarantee, but repeating it every sweep tick is log spam. */ private dispatchClaimDegradationWarned = false; + /** + * [#14333] Whether this engine has already said that its cross-replica + * advance guarantee is missing — the attached {@link SuspendedRunStore} + * has no {@link SuspendedRunStore.claimSuspension}, or answered + * `'unsupported'`. Once per instance, for the reason one field up: a + * silent fallback hides a permanently weakened guarantee, and repeating it + * per resume is log spam. + */ + private advanceClaimDegradationWarned = false; constructor(logger: Logger, store?: SuspendedRunStore, options?: AutomationEngineOptions) { this.logger = logger; @@ -1892,13 +1976,27 @@ export class AutomationEngine implements IAutomationService { * reads as terminal now and as still-suspended after a restart. See the * catch below for the full verdict. */ - private async forgetSuspendedRun(run: SuspendedRun, reason: SuspensionReleaseReason): Promise { + private async forgetSuspendedRun( + run: SuspendedRun, + reason: SuspensionReleaseReason, + /** + * [#14333] Set only by the resume path, and only when + * {@link claimAdvance} has ALREADY removed the durable row as its + * compare-and-set. The unconditional delete below would then be a + * second round-trip that can only re-delete what is gone — while the + * rest of this choke point (the cache entry, the cache-only + * qualifier, and the executor's release notification) is still owed + * exactly once. Nothing else may pass it: for every other consumption + * the durable row is still there and this method is what removes it. + */ + durableRecordAlreadyConsumed = false, + ): Promise { this.suspendedRuns.delete(run.runId); // [#13617] The qualifier goes with the entry it qualifies — this is the // one choke point every consumption passes through, so nothing can leave // a run marked "the store never took this" after its map entry is gone. this.cacheOnlySuspensions.delete(run.runId); - if (this.store) { + if (this.store && !durableRecordAlreadyConsumed) { try { await this.store.delete(run.runId); } catch (err) { @@ -1953,6 +2051,103 @@ export class AutomationEngine implements IAutomationService { await this.releaseSuspension(run, reason); } + /** + * [#14333] Claim the right to advance this run past the node it is parked + * at — the CROSS-REPLICA half of the resume idempotency guard. + * + * {@link resuming} is a per-PROCESS `Set`, which is a complete guard for + * exactly one deployment shape: a single process. Put two replicas behind + * a load balancer over one store and two decisions on one run arriving in + * the same instant each pass their OWN `resuming` check, both read the + * same fresh row through {@link loadSuspendedRunStrict}, both consume it, + * and both traverse forward — every downstream side effect runs twice. + * Measured on the two-engines-over-one-shared-store harness at 25/25 + * raced runs before this claim existed. + * + * So the decision is taken where the shared state lives: + * {@link SuspendedRunStore.claimSuspension} removes the row only if it is + * still parked at the node this replica read, atomically, and the loser is + * TOLD. `resuming` stays as the cheap first gate — it is not replaced, and + * it still refuses a same-process duplicate without any store round-trip. + * + * Four cases never reach the store, and each is a deliberate `unguarded`: + * - **no store** — nothing is shared, so there is no second replica to + * race and `resuming` is already the whole guarantee. Not a + * degradation: there is no cross-replica guarantee to weaken. + * - **a run the store never accepted** ({@link cacheOnlySuspensions}, + * #13617) — the store holds no row for it, so a compare-and-set would + * answer `lost` for a run that is legitimately resumable in this + * process, converting {@link persistSuspendedRun}'s documented + * degradation (a failed save costs cross-RESTART durability, not + * in-process resumability) into an unresumable run. + * - **a store without the member** — a third-party store predating it. + * - **a store answering `'unsupported'`** — one that cannot express the + * condition against its backing engine. + * + * The last two are announced once per engine rather than assumed silently, + * which is the whole difference between a degradation and a lie. + */ + private async claimAdvance(run: SuspendedRun): Promise { + const store = this.store; + if (!store) return { kind: 'unguarded' }; + if (this.cacheOnlySuspensions.has(run.runId)) return { kind: 'unguarded' }; + if (typeof store.claimSuspension !== 'function') { + this.warnAdvanceClaimDegraded('the attached suspended-run store has no claimSuspension()'); + return { kind: 'unguarded' }; + } + let outcome: SuspensionClaimOutcome; + try { + outcome = await store.claimSuspension(run.runId, { + nodeId: run.nodeId, + ...(run.correlation !== undefined ? { correlation: run.correlation } : {}), + }); + } catch (err) { + // Handed to the CALLER, so not a degradation and deliberately NOT a + // log site (AGENTS.md "Degradation log levels": a failure the + // requester was told about does not look normal from the outside). + // The envelope is the same STORE_UNAVAILABLE the strict load + // answers, for the same reason — existence is UNKNOWN, not "gone + // for good" (#4420). + // + // ⛔ What this branch may NOT claim is that the suspension was not + // consumed. A throw can arrive AFTER a committed delete — a dropped + // connection on the way back from a landed statement — so the one + // thing known here is that THIS resume did not continue the run. + // The result text says exactly that and no more; a retry is what + // resolves the ambiguity, answering RUN_NOT_FOUND if the row is in + // fact gone. + return { kind: 'unavailable', message: (err as Error).message }; + } + if (outcome === 'unsupported') { + this.warnAdvanceClaimDegraded("the attached suspended-run store answered 'unsupported'"); + return { kind: 'unguarded' }; + } + return outcome === 'claimed' ? { kind: 'claimed' } : { kind: 'lost' }; + } + + /** + * [#14333] Say ONCE, per engine instance, that resume idempotency is + * in-process only. + * + * #4632 verdict: FUNCTIONAL — `warn`, not `error`. Nothing that claims to + * be persisted fails to land here; what is smaller than advertised is a + * GUARANTEE, and the next person to run two replicas finds out the way + * this line says they will. Byte-for-byte the call {@link claim}'s + * missing-ledger branch makes one screen up, and for the same reason: a + * silent fallback hides a permanently weakened guarantee, while an + * `error` here would train everyone to skim `error`. + */ + private warnAdvanceClaimDegraded(cause: string): void { + if (this.advanceClaimDegradationWarned) return; + this.advanceClaimDegradationWarned = true; + this.logger.warn( + `[automation] no cross-replica advance guarantee (${cause}) — resume idempotency is ` + + `IN-PROCESS ONLY: two decisions on one run arriving at the same moment on two replicas ` + + `can both advance it, running every downstream side effect twice. Attach a suspended-run ` + + `store that implements claimSuspension() to close it.`, + ); + } + /** * Tell the executor of the node a run was paused at that the pause is over, * so it can disarm whatever it armed on entry (#5512). @@ -4964,7 +5159,66 @@ export class AutomationEngine implements IAutomationService { // signal above is pure in-memory work, not downstream work.) // This is also where the paused node learns its pause is over and // disarms what it armed on entry (#5512) — see forgetSuspendedRun. - await this.forgetSuspendedRun(run, 'resumed'); + // + // [#14333] And the consumption is now a CLAIM, not an assertion. + // `resuming` above answered "is this process already resuming it"; + // this answers "did any process get here first", against the shared + // store, atomically. Placed exactly where the unconditional consume + // was: every refusal above it still refuses without consuming, and + // the ordering #13937 has not ruled on is untouched. + const claim = await this.claimAdvance(run); + if (claim.kind === 'lost') { + // An ORDINARY outcome, not a degradation: this is the guard + // doing its job, and the caller is told in the result. `debug` + // rather than `warn` for exactly that reason — a losing racer + // per approval on a busy any-of level would otherwise be a + // steady `warn` stream describing correct behaviour. + this.logger.debug( + `[automation] resume of run '${runId}' lost the advance claim at node '${run.nodeId}' — ` + + `another replica consumed this suspension first; nothing was dispatched here.`, + ); + // The SAME code the in-process guard answers, deliberately: the + // observation differs (another process, not this one) but the + // remedy is identical — a concurrent resume already has this + // run, the caller's own recorded outcome stands, and nothing + // here should be retried. `plugin-approvals` already branches + // on it that way (`resumeRecordedOutcome`), and the transport + // already maps it to 409. A distinct code would be vocabulary + // nothing reads — add one the day a caller needs the + // difference. + return { + success: false, + code: 'RESUME_IN_PROGRESS', + error: + `Run '${runId}' is already being resumed — another replica consumed its suspension ` + + `at node '${run.nodeId}' first`, + }; + } + if (claim.kind === 'unavailable') { + // Same envelope, same reason as the strict load's failure + // above: existence is UNKNOWN, not "gone for good", and the + // identical call is expected to work once the store recovers + // (#4420). + // + // The text states only what is KNOWN. The strict load's failure + // can say the suspension was not consumed, because it fails + // before anything is consumed; this one cannot — a claim throws + // just as readily after a committed delete as before one, so + // "NOT consumed" would be a claim about a fact this seam does + // not have. What it does know is that this resume did not + // continue the run, and that a retry settles the rest. + return { + success: false, + code: 'STORE_UNAVAILABLE', + error: + `Durable suspended-run store unreachable while claiming the advance for run '${runId}' — ` + + `this resume did NOT continue the run, and whether the store consumed the suspension is ` + + `UNKNOWN (a failure can arrive after a committed delete). Retry once the store is ` + + `available: a retry that finds the run still parked continues it, and one that finds the ` + + `suspension gone answers RUN_NOT_FOUND: ${claim.message}`, + }; + } + await this.forgetSuspendedRun(run, 'resumed', claim.kind === 'claimed'); const steps = run.steps; const context = run.context; diff --git a/packages/services/service-automation/src/index.ts b/packages/services/service-automation/src/index.ts index 4a0af058ea..e9efd24379 100644 --- a/packages/services/service-automation/src/index.ts +++ b/packages/services/service-automation/src/index.ts @@ -18,6 +18,13 @@ export type { RegisteredConnector, SuspendedRun, SuspendedRunStore, + // [#14333] The conditional-advance vocabulary. `claimSuspension` is the + // cross-replica half of the resume idempotency guard, so a host supplying + // its own store has to be able to NAME its parameter and its three-valued + // answer — including the `'unsupported'` a store uses to say out loud that + // it cannot express the condition. + SuspensionParkedAt, + SuspensionClaimOutcome, FlowDispatchStore, // [ADR-0126 §7.2] The packaged-flow activation ledger port and its row — // the durable off-switch that REPLACES the retired process-local 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 1fddd2a086..c422e251b4 100644 --- a/packages/services/service-automation/src/suspended-run-store.test.ts +++ b/packages/services/service-automation/src/suspended-run-store.test.ts @@ -5,6 +5,15 @@ import { AutomationEngine } from './engine.js'; import { ObjectStoreSuspendedRunStore, type SuspendedRunStoreEngine } from './suspended-run-store.js'; import type { RunRecord, SuspendedRun } from './engine.js'; import { defineActionDescriptor } from '@objectstack/spec/automation'; +import { RESUME_AUTHORITY_SERVICE } from '@objectstack/spec/contracts'; +// [#14333] The PRODUCER's own delete-dispatch decision, imported rather than +// re-approximated. `createFakeEngine` below routes its `delete` through it, so +// the double physically cannot accept a call `ObjectQL.delete` refuses — the +// property `scripts/check-engine-double-contract.mjs` exists to hold, and the +// thing that makes the compare-and-set spelling in +// `ObjectStoreSuspendedRunStore.claimSuspension` testable at all: drop +// `multi: true` from it and this fake THROWS, exactly as a running server does. +import { assertEngineDeleteDispatch } from '@objectstack/metadata-core'; /** * The `resumeAuthority: 'any'` declaration every pausing fixture below needs @@ -64,13 +73,21 @@ function createFakeEngine( return rows.get(String(id)); }, async delete(_object, options) { - if (options?.multi) { + // [#14333] Dispatch through the producer's predicate, never a + // hand-written approximation of it: `by-id` when a truthy scalar + // `where.id` is the WHOLE predicate, `multi` when `options.multi` + // carries the rest of the `where` to the many-row route, and a + // THROW otherwise — the shape `ObjectQL.delete` refuses. Before + // this, a `where` of `{ id, node_id }` with no `multi` silently + // took the by-id branch here and returned `true`, so a store that + // had lost its condition read as an untestable `'unsupported'`. + const dispatch = assertEngineDeleteDispatch(options as any); + 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 doomed.length; } - const id = options?.where?.id; - rows.delete(String(id)); + rows.delete(String(dispatch.id)); return true; }, }; @@ -785,3 +802,245 @@ describe('ObjectStoreSuspendedRunStore — organization attribution (cloud#1395) expect(engine.rows.get('run_abc2').organization_id).toBe('org_1'); }); }); + +/** + * [#14333] The PRODUCTION conditional advance — `ObjectStoreSuspendedRunStore`, + * against the counting `multi` fake engine above. + * + * ## Why this suite exists + * + * The engine-side pin (`concurrent-replica-resume-race.test.ts`) drives two + * `AutomationEngine`s over one `InMemorySuspendedRunStore`, which proves the + * engine asks the store and honours the answer. It says NOTHING about the store + * every production deployment actually runs. Measured, on the branch before + * this suite existed: deleting `multi: true` from `claimSuspension`'s one + * `delete` call left every test green — while against a running server that + * spelling is the dispatch `reject` verdict, which throws, which `claimAdvance` + * turns into `STORE_UNAVAILABLE` on EVERY resume. A one-token regression that + * refuses every production resume must not be able to pass. + * + * So the spelling is asserted through the producer's own predicate + * (`assertEngineDeleteDispatch`, wired into `createFakeEngine` above and read + * directly below), not by matching a literal token: a test that greps for + * `multi: true` pins the characters, this pins the DECISION `ObjectQL.delete` + * will make about the same options bag. + */ +describe('#14333 ObjectStoreSuspendedRunStore.claimSuspension — the production conditional advance', () => { + /** A paused row for `run_abc`, parked at `approve_step` / `areq_1`. */ + const parkRun = async (store: ObjectStoreSuspendedRunStore, over: Partial = {}) => { + await store.save({ ...baseRun(), ...over }); + }; + + it('spells the compare-and-set the way ObjectQL dispatches to deleteMany, not the by-id route', async () => { + const engine = createFakeEngine(); + const seen: any[] = []; + const spy = { ...engine, async delete(object: string, options: any) { + seen.push(options); + return engine.delete!(object, options); + } } as SuspendedRunStoreEngine & { rows: Map }; + const store = new ObjectStoreSuspendedRunStore(spy, createTestLogger()); + await parkRun(store); + + expect(await store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' })) + .toBe('claimed'); + + expect(seen).toHaveLength(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' }); + // 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' }); + // …and it actually removed the row. + expect(engine.rows.has('run_abc')).toBe(false); + }); + + it('omits `correlation` from the condition when the caller has none, and still takes the multi route', async () => { + const engine = createFakeEngine(); + const seen: any[] = []; + const spy = { ...engine, async delete(object: string, options: any) { + seen.push(options); + return engine.delete!(object, options); + } } as SuspendedRunStoreEngine & { rows: Map }; + const store = new ObjectStoreSuspendedRunStore(spy, createTestLogger()); + 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' }); + }); + + it('maps the affected-row COUNT to the outcome: 1 is claimed, 0 is lost', async () => { + const engine = createFakeEngine(); + const store = new ObjectStoreSuspendedRunStore(engine, createTestLogger()); + await parkRun(store); + + // One row matches the whole condition. + expect(await store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' })) + .toBe('claimed'); + // The row is gone, so the second claim matches nothing. + expect(await store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' })) + .toBe('lost'); + }); + + it('a row that MOVED to another node is lost, not claimed — the row still exists', async () => { + const engine = createFakeEngine(); + const store = new ObjectStoreSuspendedRunStore(engine, createTestLogger()); + await parkRun(store, { nodeId: 'second_step', correlation: 'areq_2' }); + + // A replica still holding the earlier parking claims against it. + expect(await store.claimSuspension('run_abc', { nodeId: 'approve_step', correlation: 'areq_1' })) + .toBe('lost'); + // ⛔ And the winner's parking is intact: an existence-only consume would + // have deleted the row that another replica is standing on. + expect(engine.rows.has('run_abc')).toBe(true); + expect(engine.rows.get('run_abc').node_id).toBe('second_step'); + }); + + it('a row re-parked at the SAME node with a new correlation is lost (the map re-entry shape)', async () => { + const engine = createFakeEngine(); + const store = new ObjectStoreSuspendedRunStore(engine, createTestLogger()); + await parkRun(store, { nodeId: 'items', correlation: 'map:child_2' }); + + expect(await store.claimSuspension('run_abc', { nodeId: 'items', correlation: 'map:child_1' })) + .toBe('lost'); + expect(engine.rows.has('run_abc')).toBe(true); + expect(engine.rows.get('run_abc').correlation).toBe('map:child_2'); + }); + + it('an engine with no delete() answers unsupported, and says so ONCE per store', async () => { + const engine = createFakeEngine(); + const noDelete = { find: engine.find, insert: engine.insert, update: engine.update } as SuspendedRunStoreEngine; + const lines: string[] = []; + const store = new ObjectStoreSuspendedRunStore(noDelete, { 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'); + } + // Once per STORE, not once per call: the composition cannot express the + // condition for the life of the process, so repeating it every resume + // is repetition of a permanent fact. + expect(lines).toHaveLength(1); + expect(lines[0]).toContain('no cross-replica advance guarantee'); + expect(lines[0]).toContain('engine has no delete()'); + }); + + it('a multi-row result that is not a COUNT answers unsupported, and says so ONCE per store', async () => { + const engine = createFakeEngine(); + // A driver whose many-row delete resolves something other than a number: + // "did I win?" has no answer to read, so the store must not guess. + const nonCounting = { ...engine, async delete() { return true; } } as SuspendedRunStoreEngine; + const lines: string[] = []; + const store = new ObjectStoreSuspendedRunStore(nonCounting, { 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(lines).toHaveLength(1); + expect(lines[0]).toContain('not an affected-row count'); + expect(lines[0]).toContain('no cross-replica advance guarantee'); + }); + + // ── End to end: two engines over ONE ObjectStoreSuspendedRunStore ──────── + // + // The engine-side pin runs over the in-memory store. This is the same race + // driven through the store production actually uses, so the predicate + // spelling, the count mapping and the engine's fold are exercised together. + + const RACE_FLOW = { + 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: 'lv2', type: 'approval_level', label: 'General manager' }, + { id: 'end', type: 'end', label: 'End' }, + ], + edges: [ + { id: 'e1', source: 'start', target: 'lv1' }, + { id: 'e2', source: 'lv1', target: 'notify' }, + { id: 'e3', source: 'notify', target: 'lv2' }, + { id: 'e4', source: 'lv2', target: 'end' }, + ], + } as any; + + function raceReplica(store: ObjectStoreSuspendedRunStore, opened: string[], fired: string[]) { + const engine = new AutomationEngine(createTestLogger(), 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', RACE_FLOW); + return engine; + } + + it('two engines over ONE durable store advance a raced run exactly once', async () => { + const dataEngine = createFakeEngine(); + const store = new ObjectStoreSuspendedRunStore(dataEngine, createTestLogger()); + const opened: string[] = []; + const fired: string[] = []; + const a = raceReplica(store, opened, fired); + const b = raceReplica(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 observable effect, through the production store: one advance. + expect(fired).toEqual(['notify']); + expect(opened).toEqual(['lv1', 'lv2']); + expect(both.filter((r) => r.success)).toHaveLength(1); + const loser = both.find((r) => !r.success)!; + expect(loser.code).toBe('RESUME_IN_PROGRESS'); + expect(loser.status).toBeUndefined(); + // The winner's new parking survived the loser's claim. + expect(dataEngine.rows.get(runId).node_id).toBe('lv2'); + }); + + it('a claim that THROWS refuses the resume as STORE_UNAVAILABLE and says only what it knows', async () => { + const dataEngine = createFakeEngine(); + const store = new ObjectStoreSuspendedRunStore(dataEngine, createTestLogger()); + const opened: string[] = []; + const fired: string[] = []; + const engine = raceReplica(store, opened, fired); + const runId = (await engine.execute('expense_approval')).runId!; + + (store as any).claimSuspension = async () => { throw new Error('sqlite: database is locked'); }; + const refused = await engine.resume(runId, { [RESUME_AUTHORITY_SERVICE]: true } as any); + + expect(refused.success).toBe(false); + // Unknown, never "gone for good" (#4420) — the same call is expected to + // work once the store recovers. + expect(refused.code).toBe('STORE_UNAVAILABLE'); + expect(refused.status).toBeUndefined(); + // ⛔ It must NOT assert the suspension was not consumed: a throw can + // arrive after a committed delete. It states what it knows and hands + // the ambiguity to a retry. + expect(refused.error).toContain('this resume did NOT continue the run'); + expect(refused.error).toContain('UNKNOWN'); + expect(refused.error).not.toContain('the suspension was NOT consumed'); + // Nothing ran, and the run is still parked for that retry. + expect(fired).toEqual([]); + expect(opened).toEqual(['lv1']); + expect(dataEngine.rows.get(runId).node_id).toBe('lv1'); + }); +}); diff --git a/packages/services/service-automation/src/suspended-run-store.ts b/packages/services/service-automation/src/suspended-run-store.ts index ae1ccc9ffe..0ad54b2841 100644 --- a/packages/services/service-automation/src/suspended-run-store.ts +++ b/packages/services/service-automation/src/suspended-run-store.ts @@ -9,7 +9,13 @@ import type { Logger } from '@objectstack/spec/contracts'; // recorder-local re-derivation here was rejected by name (Option B): it would // be a third answer to a question the codebase already answered two ways. import { createRecordOrganizationResolver, type RecordOrganizationResolver } from '@objectstack/metadata-core'; -import type { RunRecord, SuspendedRun, SuspendedRunStore } from './engine.js'; +import type { + RunRecord, + SuspendedRun, + SuspendedRunStore, + SuspensionClaimOutcome, + SuspensionParkedAt, +} from './engine.js'; /** * Durable persistence for suspended flow runs (ADR-0019). @@ -127,6 +133,25 @@ export class InMemorySuspendedRunStore implements SuspendedRunStore { this.runs.delete(runId); } + /** + * [#14333] Conditional consume. The test and the removal sit between the + * same two `await` points — there is none between them — so on a single + * JavaScript thread they ARE one atomic operation against `this.runs`, which + * is what lets two engines sharing one instance of this store produce + * exactly one winner. Spelling it as `load()` then `delete()` would put a + * microtask boundary in the middle and reopen the window. + */ + async claimSuspension(runId: string, parkedAt: SuspensionParkedAt): Promise { + const run = this.runs.get(runId); + if (!run) return 'lost'; + if (run.nodeId !== parkedAt.nodeId) return 'lost'; + // Compared only when the caller has one, per the contract: a row persisted + // without a correlation has nothing to compare and the node test stands. + if (parkedAt.correlation !== undefined && run.correlation !== parkedAt.correlation) return 'lost'; + this.runs.delete(runId); + return 'claimed'; + } + async list(): Promise { return [...this.runs.values()].map(jsonClone); } @@ -208,6 +233,11 @@ export interface ObjectStoreSuspendedRunStoreOptions { */ export class ObjectStoreSuspendedRunStore implements SuspendedRunStore { private readonly maxTerminalRunsPerFlow: number; + /** + * [#14333] Whether this store has already said it cannot express the + * conditional advance — see {@link claimSuspension}'s docblock. + */ + private claimUnsupportedWarned = false; /** * [#10101] Memoized shared platform-row organization resolver over the same * engine the rows are written through — answers "which column carries the @@ -280,6 +310,93 @@ export class ObjectStoreSuspendedRunStore implements SuspendedRunStore { await this.engine.delete(TABLE, { where: { id: runId }, context: SYSTEM_CTX }); } + /** + * [#14333] Conditional consume against `sys_automation_run`, as ONE + * statement: `DELETE ... WHERE id = ? AND node_id = ?` (plus `correlation` + * when the caller has one), whose affected-row count names the winner. + * + * ⛔ No schema change, and none is needed: `node_id` and `correlation` are + * columns {@link serialize} has always written, so the condition is + * expressible against the row exactly as it stands. (A version column on + * `sys_automation_run` was the other candidate remedy and is deliberately + * NOT taken — a platform-object schema change is the maintainer's floor.) + * + * The spelling is `multi: true` with a full `where`: `ObjectQL.delete` + * dispatches a `where` carrying keys BESIDES `id` to `driver.deleteMany` + * with the composed AST — the engine's own compare-and-set route, documented + * as such in `engine-delete-dispatch.ts` — and `IDataDriver.deleteMany` is + * contracted to resolve an affected COUNT. A pure-`id` `where` would take + * the by-id route instead and silently discard the condition, which is the + * one shape that must never happen here. ⛔ Dropping `multi: true` does not + * degrade to the by-id route either: with unhonoured keys in the `where` and + * no `multi`, the dispatch verdict is `reject` and `ObjectQL.delete` THROWS + * `ENGINE_DELETE_REJECT_MESSAGE`. That is pinned in + * `suspended-run-store.test.ts` through the producer's own predicate rather + * than by asserting the token, so a double can never be looser than the + * engine here. + * + * THREE ways a composition can leave this store unable to offer the + * guarantee, and only the first two are an `'unsupported'` ANSWER: + * + * 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; + * 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 + * `ENGINE_DELETE_REJECT_MESSAGE` rather than returning anything; the + * engine's `claimAdvance` maps that throw to `STORE_UNAVAILABLE`, so such + * a composition REFUSES every resume instead of degrading to an + * unguarded one. All five shipped drivers implement `deleteMany` (memory, + * sql, mongodb, turso; sqlite-wasm inherits it from `SqlDriver`), so this + * is third-party exposure only — recorded here because a host wiring its + * own driver is the one reader who can hit it, and the symptom (every + * resume 503) does not name its cause. + * + * The two answers it CAN give are said once per store instance, not once per + * call: a composition that cannot express the condition cannot express it for + * the rest of the process, so a line per resume would be pure repetition of a + * permanent fact — the same call {@link AutomationEngine.claim}'s + * missing-ledger branch makes, and what the engine's own degradation line + * 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. + */ + async claimSuspension(runId: string, parkedAt: SuspensionParkedAt): Promise { + if (typeof this.engine.delete !== 'function') { + this.warnClaimUnsupported( + `engine has no delete(); the conditional advance for suspended run '${runId}' cannot be expressed`, + ); + 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`, + ); + return 'unsupported'; + } + return affected > 0 ? 'claimed' : 'lost'; + } + + /** + * [#14333] Say ONCE, per store instance, that this composition cannot + * express the conditional advance. See {@link claimSuspension}'s docblock + * for why once and not per call. + */ + private warnClaimUnsupported(cause: string): void { + if (this.claimUnsupportedWarned) return; + this.claimUnsupportedWarned = true; + this.logger?.warn?.( + `[automation] ObjectStoreSuspendedRunStore: ${cause} — no cross-replica advance guarantee is offered ` + + `by this store for the life of this process.`, + ); + } + async list(): Promise { const rows = await this.engine.find(TABLE, { where: { status: 'paused' }, limit: 1000, context: SYSTEM_CTX, diff --git a/scripts/engine-double-contract.baseline.json b/scripts/engine-double-contract.baseline.json index 123fafbac4..f4bfac2a3e 100644 --- a/scripts/engine-double-contract.baseline.json +++ b/scripts/engine-double-contract.baseline.json @@ -856,14 +856,6 @@ "why": "MEASURED (#5480): the `update` slice of this gate is NEW — `resolveEngineUpdateDispatch` did not exist before #5480, so no double in the repo could route through it and the whole discovered set enters this ledger in one act. Not newly written looseness and not a raised ratchet: it is the first measurement of a contract that had no producer-side predicate to measure against, which is exactly what this script's header used to list under deliberately-not-covered (\"update's twin dispatch ... needs its own producer-side predicate extracted first\"). Discovered at line 36. The package already depends on @objectstack/objectql (`devDependencies`), so this is a one-line pin whenever a batch takes it — deferred here because #5480's slice is the producer-side predicate plus the gate that reads it, and flipping ~100 unmeasured suites red belongs in the per-package batches that follow, exactly as #5629 did for delete. WHAT THIS ENTRY DOES NOT CLAIM: unlike the #5629 delete batch above it carries NO per-file dormancy probe. Nothing here says the looseness is unexercised — only that the double is structurally free to be looser than ObjectQL.update on the shape a hand-written guard always drops (`where: { id: { $in: [...] } }` looks like an id and is a multi-row predicate) — a shape the producer refuses in `data.id` too since objectstack#5748 put the payload half through the SAME scalar test, so `data.id` still outranks `where` and `multi`, but only when it IS a scalar id.", "closes": "open the fake's update with assertEngineUpdateDispatch(data, options) and run the package's suite" }, - { - "file": "packages/services/service-automation/src/suspended-run-store.test.ts", - "verb": "delete", - "unguarded": 1, - "kind": "DEBT", - "why": "HAND-MIRRORS the guard already, which is the second copy of the contract this gate exists to remove. MEASURED (#5393): the devDependency that used to block replacing the copy now EXISTS — @objectstack/objectql was added to @objectstack/service-automation's devDependencies when `builtin/crud-bulk-intent.test.ts` was pinned, and the graph is acyclic (see the sibling entries). What is left is replacing the mirrored `if` with the producer's predicate.", - "closes": "replace the mirrored `if` with assertEngineDeleteDispatch(options) — the devDependency is already declared" - }, { "file": "packages/services/service-automation/src/suspended-run-store.test.ts", "verb": "update", diff --git a/scripts/engine-double-contract.pinned.json b/scripts/engine-double-contract.pinned.json index 89a5672181..ee25bbcc8e 100644 --- a/scripts/engine-double-contract.pinned.json +++ b/scripts/engine-double-contract.pinned.json @@ -3191,6 +3191,11 @@ "verb": "findOne", "pinned": 1 }, + { + "file": "packages/services/service-automation/src/suspended-run-store.test.ts", + "verb": "delete", + "pinned": 1 + }, { "file": "packages/services/service-datasource/src/__tests__/datasource-admin-plugin.test.ts", "verb": "findOne",