fix(plugin-approvals): re-issuing a restored approval continuation proves the pause past the first 1000 parked runs

.changeset/28271-restored-continuation-past-first-page.md

@@ -0,0 +1,14 @@
+---
+'@objectstack/plugin-approvals': patch
+'@objectstack/service-automation': patch
+---
+
+fix(plugin-approvals): re-issuing a restored approval continuation proves the pause past the first 1000 parked runs
+
+Clause-②: no (a re-issue on a store with more than 1000 parked runs now proves the pause as it already does below that; no schema, accept set, error code or result field changes)
+Metadata-protocol: none
+
+`continueRestoredRun` proves the pause it continues is the one the request's outcome was issued at, by reading where the run is parked. It read that from `listSuspendedRunsDurable()`. On a database-backed store that listing reads one page of 1000 `paused` rows, and a process adds only the runs it parked itself. So when another replica (or this process before a restart) parked the run and more than 1000 runs were parked ahead of it, the run was on no page. Every re-issue was refused with `INVALID_STATE` "parked at no node this engine can see", every retry read the same page, and the recorded outcome never took effect.
+
+- **Now.** The proof asks the engine for that one run. The new `AutomationEngine.suspensionOf(runId)` reads the run's suspension by id, through the same strict, store-authoritative read `hasSuspendedRun` uses, and answers its flow, node and correlation, or `undefined` when it is not suspended. It rejects when the store cannot be read.
+- **Unchanged.** Below 1000 parked runs the re-issue proceeds as before. A run that is not parked, or is parked at another node, is still refused, and the refusal still names the node. An engine without the read is still refused; the refusal now names `suspensionOf` as the missing read.

packages/plugins/plugin-approvals/src/approval-service.ts

@@ -270,24 +270,36 @@ export interface ApprovalResumeSurface {
    */
   unresumableSuspensionCause?(runId: string): Promise<string | undefined>;
   /**
-   * [#15389] Where each suspended run is currently parked. Read by
-   * {@link ApprovalService.continueRestoredRun} to prove the pause it was asked
-   * to continue is the one THIS request's recorded outcome was issued at, and
-   * not merely some live pause on the same run — `hasSuspendedRun` answers a
-   * boolean and cannot tell the two apart. Which node that is depends on the signal, not
-   * only on the row: see {@link ApprovalService.expectedPauseNode}.
+   * [#15389] Where each suspended run is currently parked, durable-first.
    *
    * ⚠️ Declares a method `AutomationEngine` ALREADY implements publicly
-   * (`listSuspendedRunsDurable`); it widens no engine surface. Durable-first,
-   * so it sees a pause parked by another replica or before a restart.
+   * (`listSuspendedRunsDurable`); it widens no engine surface. [#28271]
+   * {@link ApprovalService.continueRestoredRun} no longer reads it: on a
+   * DB-backed store it is one capped page, and it now asks
+   * {@link suspensionOf} for the one run instead.
+   */
+  listSuspendedRunsDurable?(): Promise<Array<{ runId: string; flowName: string; nodeId: string; correlation?: string }>>;
+  /**
+   * [#15389, #28271] Where the run `runId` is parked, or `undefined` when it
+   * is not suspended. Read by {@link ApprovalService.continueRestoredRun} to
+   * prove the pause it was asked to continue is the one THIS request's
+   * recorded outcome was issued at, and not merely some live pause on the
+   * same run — `hasSuspendedRun` answers a boolean and cannot tell the two
+   * apart. Which node that is depends on the signal, not only on the row:
+   * see {@link ApprovalService.expectedPauseNode}.
+   *
+   * ⚠️ Declares the method `AutomationEngine` implements publicly
+   * (`suspensionOf`): one keyed read of the store, so it sees a pause parked
+   * by another replica or before a restart however many runs are parked
+   * ahead of it — where {@link listSuspendedRunsDurable}, which this proof
+   * read until #28271, is one capped page of a DB-backed store. Read-only;
+   * rejects when the store cannot be read.
    *
    * Optional, and its absence is FAIL-CLOSED: a caller that cannot prove node
    * identity refuses the continuation rather than proceeding on the weaker
-   * check. The engine degrades a store outage to its in-memory list rather
-   * than throwing, and that degradation is safe HERE for the same reason — a
-   * pause it cannot see is a pause this verb will not act on.
+   * check.
    */
-  listSuspendedRunsDurable?(): Promise<Array<{ runId: string; flowName: string; nodeId: string; correlation?: string }>>;
+  suspensionOf?(runId: string): Promise<{ runId: string; flowName: string; nodeId: string; correlation?: string } | undefined>;
   /**
    * [#28230] The run parked at a `map` node on the item run `itemRunId`, or
    * `undefined` when there is none. Read by
@@ -7381,23 +7393,27 @@ export class ApprovalService implements IApprovalService {
     // node refused the one population the journal was built to serve, with a
     // message that named a cause the code had not taken.
     //
-    // FAIL-CLOSED in every direction: no reader, no listing, no matching entry,
-    // an entry at another node, or an expected node this service cannot derive
-    // — all refuse. The engine degrades a store outage to its in-memory list
-    // instead of throwing, so a pause this cannot see is a pause this verb
-    // declines to act on. A false refusal costs the operator a retry; a false
-    // admission advances a flow nobody decided.
+    // FAIL-CLOSED in every direction: no reader, no pause, a pause at another
+    // node, or an expected node this service cannot derive — all refuse. A
+    // store that cannot be read rejects the read, as GUARD 2's does. A false
+    // refusal costs the operator a retry; a false admission advances a flow
+    // nobody decided.
+    //
+    // [#28271] The read is KEYED by the run id. It used to be the run's entry
+    // in `listSuspendedRunsDurable()`, which on a DB-backed store is one page
+    // of 1000 parked runs plus what this process parked itself: a run another
+    // process parked behind more than 1000 others was on no page, so every
+    // retry was refused the same way and the recorded outcome never took effect.
     const expected = await this.expectedPauseNode(raw, signal, requestId, runId);
-    if (typeof this.automation?.listSuspendedRunsDurable !== 'function') {
+    if (typeof this.automation?.suspensionOf !== 'function') {
       throw new Error(
         `INVALID_STATE: this automation engine cannot report WHERE run '${runId}' is parked ` +
-        `(no listSuspendedRunsDurable), so the pause cannot be proved to be the one ${signal.what} on ` +
+        `(no suspensionOf), so the pause cannot be proved to be the one ${signal.what} on ` +
         `request ${requestId} was refused on — refusing, because continuing the wrong pause advances a ` +
         `flow with no decision behind it`,
       );
     }
-    const parkedAt = (await this.automation.listSuspendedRunsDurable())
-      .find(r => String(r.runId) === String(runId))?.nodeId;
+    const parkedAt = (await this.automation.suspensionOf(runId))?.nodeId;
     if (parkedAt !== expected.nodeId) {
       throw new Error(
         `INVALID_STATE: run '${runId}' is parked at ` +

packages/plugins/plugin-approvals/src/restored-approval-continuation.test.ts

@@ -39,7 +39,7 @@
  */
 
 import { describe, it, expect, beforeEach } from 'vitest';
-import { AutomationEngine, InMemorySuspendedRunStore, resolveRunDataContext } from '@objectstack/service-automation';
+import { AutomationEngine, InMemorySuspendedRunStore, ObjectStoreSuspendedRunStore, resolveRunDataContext } from '@objectstack/service-automation';
 // [#4550] The engine double below routes its write verbs through ObjectQL's OWN
 // dispatch predicates rather than a hand-mirrored copy.
 import { assertEngineDeleteDispatch, assertEngineUpdateDispatch } from '@objectstack/objectql';
@@ -1035,3 +1035,157 @@ describe('ruling A — a decision resumes the run as the person who decided', ()
     expect(seen.after_reject, 'the replay fell back to the trigger user').toEqual({ userId: 'u1', permissions: ['deal_manager'] });
   });
 });
+
+/**
+ * GUARD 3 proves where the run is parked past the first page of parked runs
+ * (#28271).
+ *
+ * It used to find the run in `listSuspendedRunsDurable()`. A DB-backed store
+ * lists one 1000-row page of `paused` rows in driver order, and a process adds
+ * only what it parked itself. So when another process (a replica, or this one
+ * before a restart) parked the run behind more than 1000 others, the run was
+ * not on the listing, and every re-issue was refused with "parked at no node
+ * this engine can see" — the retry the refusal promises read the same page.
+ * Both replicas below share one store and one table set: `a` strands and
+ * restores the run, `b` re-issues it.
+ */
+describe('#28271 — GUARD 3 reads where the run is parked by its id, not from one page', () => {
+  let data: ReturnType<typeof makeFakeEngine>;
+  let marks: string[];
+  let rejectBranchThrows: string | undefined;
+  let a: ReturnType<typeof wire>;
+  let b: ReturnType<typeof wire>;
+
+  /** One process: a real engine over `store`, the approval node, and its own approvals service. */
+  function wire(store: ObjectStoreSuspendedRunStore) {
+    const service = new ApprovalService({ engine: data as any, logger: noopLogger });
+    const automation = new AutomationEngine(noopLogger as any, store);
+    registerApprovalNode(automation, service, noopLogger as any);
+    automation.registerNodeExecutor({
+      type: 'mark',
+      async execute(node: any) {
+        if (node.id === 'mark_rejected' && rejectBranchThrows) throw new Error(rejectBranchThrows);
+        marks.push(node.id);
+        return { success: true };
+      },
+    } as never);
+    automation.registerFlow('deal_approval', DEAL_APPROVAL as never);
+    automation.registerFlow('revise_flow', REVISE_FLOW as never);
+    service.attachAutomation(automation);
+    return { automation, service };
+  }
+
+  /** `count` parked runs of some other flow, stored ahead of everything this test parks. */
+  const parkOthers = (count: number) => {
+    const rows = data.tables.get('sys_automation_run') ?? (data.tables.set('sys_automation_run', []), data.tables.get('sys_automation_run')!);
+    for (let i = 0; i < count; i++) {
+      rows.push({ id: `other-${String(i).padStart(4, '0')}`, flow_name: 'other_flow', node_id: 'wait', status: 'paused' });
+    }
+  };
+
+  /** `b`'s engine with one member taken away, everything else delegated. */
+  const without = (member: string) => new Proxy(b.automation, {
+    get(target, prop) {
+      if (prop === member) return undefined;
+      const value = Reflect.get(target, prop, target);
+      return typeof value === 'function' ? value.bind(target) : value;
+    },
+  });
+
+  beforeEach(() => {
+    marks = []; rejectBranchThrows = undefined;
+    data = makeFakeEngine();
+    const store = new ObjectStoreSuspendedRunStore(data as any, noopLogger as any);
+    a = wire(store);
+    b = wire(store);
+  });
+
+  const rowOf = async (id: string) =>
+    (await data.find('sys_approval_request', { where: { id } }))[0];
+
+  /** On `a`: park, strand a reject decision, and (unless told not to) re-arm its pause. */
+  async function strandOnA({ restore = true } = {}) {
+    await a.automation.execute('deal_approval', {
+      object: 'crm_deal', record: { id: 'd1', amount: 100 }, userId: 'submitter',
+    } as never);
+    const req = (await data.find('sys_approval_request', { where: { status: 'pending' } }))[0];
+    rejectBranchThrows = DOWNSTREAM_FAILURE;
+    const err = await a.service
+      .decide(req.id, { decision: 'reject', actorId: 'u1', comment: 'no' }, SYSTEM_CTX)
+      .then(() => null, (e: Error) => e);
+    expect(err?.message).toMatch(/^RESUME_FAILED/);
+    if (restore) {
+      expect((await a.automation.restoreConsumedSuspension(req.flow_run_id, { requestedBy: 'ops' })).restored).toBe(true);
+    }
+    rejectBranchThrows = undefined;
+    return req;
+  }
+
+  it('past the first page, the pause another process re-armed is proved and the decision is re-issued', async () => {
+    parkOthers(1000);
+    const req = await strandOnA();
+    // The premise: `b` holds nothing in memory, and the run is not on the page.
+    expect(await b.automation.hasSuspendedRun(req.flow_run_id), 'the pause is back, in the store').toBe(true);
+    expect((await b.automation.listSuspendedRunsDurable()).some(r => r.runId === req.flow_run_id),
+      'and not on the listing\'s one page').toBe(false);
+
+    const out = await b.service.continueRestoredRun(req.id, { requestedBy: 'ops', reason: 'record recreated' });
+    expect(out.resumed, JSON.stringify(out)).toBe(true);
+    expect(out.decision).toBe('reject');
+    expect(out.source).toBe('journal');
+    expect(marks, 'the reject branch ran').toEqual(['mark_rejected']);
+    expect(await b.automation.hasSuspendedRun(req.flow_run_id), 'the continuation consumed the pause').toBe(false);
+  });
+
+  it('CONTROL: within the first page it is re-issued as before', async () => {
+    parkOthers(998);
+    const req = await strandOnA();
+    expect((await b.automation.listSuspendedRunsDurable()).some(r => r.runId === req.flow_run_id)).toBe(true);
+
+    const out = await b.service.continueRestoredRun(req.id, { requestedBy: 'ops' });
+    expect(out.resumed, JSON.stringify(out)).toBe(true);
+    expect(marks).toEqual(['mark_rejected']);
+  });
+
+  it('CONTROL: past the first page, a run parked at another node is still refused, and the refusal names that node', async () => {
+    parkOthers(1000);
+    await a.automation.execute('revise_flow', {
+      object: 'crm_deal', record: { id: 'd7', amount: 3 }, userId: 'submitter',
+    } as never);
+    const rev = (await data.find('sys_approval_request', { where: { status: 'pending' } }))[0];
+    await a.service.sendBack(rev.id, { actorId: 'u1', comment: 'fix it' } as any, SYSTEM_CTX);
+    expect((await rowOf(rev.id)).status).toBe('returned');
+
+    const refused = await b.service.continueRestoredRun(rev.id).then(() => null, (e: Error) => e);
+    expect(refused?.message).toMatch(
+      /^INVALID_STATE: run '.*' is parked at node 'wait_revision', but the send-back on request .* was issued from its own approval node 'review'/,
+    );
+    expect(await data.find('sys_approval_request', { where: { flow_run_id: rev.flow_run_id } }),
+      'no new pending round was opened').toHaveLength(1);
+  });
+
+  it('CONTROL: past the first page, a run that is not parked is still refused — by GUARD 3 too, where GUARD 2 cannot ask', async () => {
+    parkOthers(1000);
+    const req = await strandOnA({ restore: false });
+
+    const byGuard2 = await b.service.continueRestoredRun(req.id).then(() => null, (e: Error) => e);
+    expect(byGuard2?.message).toMatch(/is not suspended/);
+
+    // An engine that cannot answer `hasSuspendedRun` skips GUARD 2; GUARD 3's
+    // keyed read finds no pause and refuses.
+    b.service.attachAutomation(without('hasSuspendedRun') as any);
+    const byGuard3 = await b.service.continueRestoredRun(req.id).then(() => null, (e: Error) => e);
+    expect(byGuard3?.message).toMatch(/^INVALID_STATE: run '.*' is parked at no node this engine can see/);
+    expect(marks, 'nothing ran').toEqual([]);
+  });
+
+  it('CONTROL: an engine that cannot say where a run is parked is refused', async () => {
+    const req = await strandOnA();
+    b.service.attachAutomation(without('suspensionOf') as any);
+
+    const refused = await b.service.continueRestoredRun(req.id).then(() => null, (e: Error) => e);
+    expect(refused?.message).toMatch(/^INVALID_STATE: this automation engine cannot report WHERE run '.*' is parked \(no suspensionOf\)/);
+    expect(marks, 'nothing ran').toEqual([]);
+    expect(await b.automation.hasSuspendedRun(req.flow_run_id), 'the pause is left as it was').toBe(true);
+  });
+});

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

@@ -6905,6 +6905,26 @@ export class AutomationEngine implements IAutomationService {
         return { runId: parent.runId, flowName: parent.flowName, nodeId: parent.nodeId, correlation: parent.correlation };
     }
 
+    /**
+     * [#28271] Where `runId` is parked — its flow, node and correlation — or
+     * `undefined` when it is not suspended. One keyed read through the strict,
+     * store-authoritative loader {@link hasSuspendedRun} uses, never a listing:
+     * {@link listSuspendedRunsDurable} reads one capped page of a DB-backed
+     * store, so past 1000 parked runs a run another process parked may not be
+     * on it.
+     *
+     * Read-only. THROWS when the store cannot be read, as
+     * {@link hasSuspendedRun} does: an outage means "unknown", never "not
+     * parked".
+     */
+    async suspensionOf(
+        runId: string,
+    ): Promise<{ runId: string; flowName: string; nodeId: string; correlation?: string } | undefined> {
+        const run = await this.loadSuspendedRunStrict(runId);
+        if (!run) return undefined;
+        return { runId: run.runId, flowName: run.flowName, nodeId: run.nodeId, correlation: run.correlation };
+    }
+
     /**
      * Credit a completed child run's totals to the parent step waiting on it
      * (#4354).