From 9c57d63459e7cb3aa75dbd72963517fda99a1ba6 Mon Sep 17 00:00:00 2001 From: Ruben van der Linde Date: Thu, 3 Sep 2026 11:52:36 +0200 Subject: [PATCH] fix(flow): the heartbeat recovers a missed signal instead of rolling forever A UserTask's completion signal was REFUSED on the acceptance rig (the assignee group did not exist at signal time), and the suspended run's 30-minute heartbeat then fired, re-suspended, and never advanced: resume_at rolled 08:07 -> 08:37 -> ... while the task sat `completed` and the group had long been created. The heartbeat exists precisely to recover a missed wake; it recovered nothing. The node was never the problem. UserTaskNode has always re-read its task on re-entry and applied the outcome when terminal. What wedged is upstream: persistResult() dropped EVERY parked node's resume slot whenever a pass ended anything but `suspended`. FlowResumeState::storableWhen(suspended:) read NOT-SUSPENDED as "nothing left to continue from", which conflates it with TERMINAL -- and a pass legitimately ends `queued` while a node parked in an EARLIER pass still waits: the in-request advance of a sibling branch finalises `queued` whenever other enabled work remains, and a refused claim does the same. The parked node then lost the uuid of the task it was waiting on. Its next wake found an empty slot and -- correctly, by its own idempotency guard -- created a NEW task. From that moment the ORIGINAL task's completion could never address the node's slot, its signal was refused against the new slot's recorded assignee (the refusal observed on the rig), every heartbeat re-read the new open task and re-suspended, and a duplicate task sat in somebody's inbox. Slots now survive every pass end the run can still advance from, and drop only on a terminal one. A recovered delivery is recorded on the task's audit as `heartbeat-recovered`, attributed to the task's completedBy, so the trail no longer ends at the refusal. The symmetric cases need no new mechanism, and the change says so. A task completed while the run was not yet suspended (the race) loses its signal -- signal() refuses a non-suspended run -- and the node then parks on a non-null heartbeat, so the next wake re-reads it; the race costs at most one heartbeat period. A task concluded by TaskSequenceService reaches its terminal state on the same row the re-read observes, so it is covered with no sequence-specific handling. No second delivery mechanism is added: the wake, the re-read and the outcome application are the paths that already existed. Proven red before the fix: FlowHeartbeatRecoveryTest drives the real engine, dispatcher, node, stream walk, claims and commit path over in-memory mappers, so the pass-to-pass persistence that loses the slot is exercised exactly as the worker exercises it. Co-Authored-By: Claude Opus 5 (1M context) --- lib/Service/Flow/FlowResumeState.php | 29 +- lib/Service/Flow/FlowRunService.php | 54 +- lib/Service/Flow/FlowTaskBridge.php | 54 ++ lib/Service/Flow/Nodes/PortalTaskNode.php | 10 + lib/Service/Flow/Nodes/UserTaskNode.php | 11 + .../changes/flow-heartbeat-recovery/design.md | 97 ++++ .../flow-heartbeat-recovery/proposal.md | 76 +++ .../specs/flow-heartbeat-recovery/spec.md | 111 ++++ .../changes/flow-heartbeat-recovery/tasks.md | 37 ++ .../Flow/FlowHeartbeatRecoveryTest.php | 514 ++++++++++++++++++ .../Unit/Service/Flow/FlowResumeStateTest.php | 39 ++ .../Unit/Service/Flow/PortalTaskNodeTest.php | 46 ++ tests/Unit/Service/Flow/UserTaskNodeTest.php | 44 ++ 13 files changed, 1105 insertions(+), 17 deletions(-) create mode 100644 openspec/changes/flow-heartbeat-recovery/design.md create mode 100644 openspec/changes/flow-heartbeat-recovery/proposal.md create mode 100644 openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md create mode 100644 openspec/changes/flow-heartbeat-recovery/tasks.md create mode 100644 tests/Unit/Service/Flow/FlowHeartbeatRecoveryTest.php diff --git a/lib/Service/Flow/FlowResumeState.php b/lib/Service/Flow/FlowResumeState.php index 50583d44b1..e94457c5bf 100644 --- a/lib/Service/Flow/FlowResumeState.php +++ b/lib/Service/Flow/FlowResumeState.php @@ -207,24 +207,37 @@ public static function fromArray(mixed $stored): self { /** * The storable form, or null when there is nothing worth storing. * - * Only a SUSPENDED run has anywhere to continue from. A terminal one does - * not, so keeping its slots would put a stale cursor in front of anyone - * reading the run to find out what happened — and the dispatcher has already - * cleared every node that returned, so anything still held belongs to a node - * the run never came back to. + * Kept for every run that can still advance, dropped only on a terminal + * one. The first version of this rule said "only a SUSPENDED run has + * anywhere to continue from", and that conflated NOT-SUSPENDED with + * TERMINAL: a pass can end `queued` — an in-request advance whose sibling + * still has enabled work, a claim refused on contention — while a node + * parked in an EARLIER pass still holds live progress. Dropping the slots + * there is how the heartbeat wedge happened: a user-task node lost the + * uuid of the task it was waiting on, asked again on the next wake, and + * the original task's completion could never address the node's slot + * again — its signal was refused against the new slot's assignee, and the + * run rolled its heartbeat forever. + * + * A terminal run still drops them: keeping its slots would put a stale + * cursor in front of anyone reading the run to find out what happened — + * and the dispatcher has already cleared every node that returned, so + * anything still held belongs to a node the run never came back to. * * Lives here rather than in the run service because it is a question about * this value, not about persistence: the state knows when it is worth * keeping. * - * @param boolean $suspended Whether the walk ended suspended. + * @param boolean $live Whether the run can still advance (any non-terminal + * status — suspended, queued, running). * * @return array>|null The slots, or null to drop them. * * @spec openspec/specs/flow-engine/spec.md#requirement-a-node-must-be-able-to-resume-from-where-it-stopped + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-live-run-keeps-every-parked-nodes-resume-slot */ - public function storableWhen(bool $suspended): ?array { - if ($suspended === false || $this->byNode === []) { + public function storableWhen(bool $live): ?array { + if ($live === false || $this->byNode === []) { return null; } diff --git a/lib/Service/Flow/FlowRunService.php b/lib/Service/Flow/FlowRunService.php index 5f4da8706e..43346f0f82 100644 --- a/lib/Service/Flow/FlowRunService.php +++ b/lib/Service/Flow/FlowRunService.php @@ -907,7 +907,7 @@ public function execute(FlowRun $run, array $flow, object $subject, ?array $seed $resuming = ($run->getStatus() === FlowRun::STATUS_SUSPENDED); if ($resuming === true) { // Stored items win on resume (below), but the subject's own fields - // on them are a trigger-time snapshot. {@see self::refreshSubjectItems()} + // on them are a trigger-time snapshot: see refreshSubjectItems(). $this->refreshSubjectItems(run: $run, subject: $subject); } @@ -1091,6 +1091,7 @@ private function failUnresolvableVersion(FlowRun $run): FlowRun { * @return FlowRun The updated run. * * @spec openspec/changes/or-flow-runs/specs/flow-runs/spec.md + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-live-run-keeps-every-parked-nodes-resume-slot */ private function persistResult(FlowRun $run, array $result): FlowRun { $status = (string)($result['status'] ?? FlowRun::STATUS_FAILED); @@ -1141,14 +1142,7 @@ private function persistResult(FlowRun $run, array $result): FlowRun { // other node also reads from. unset($context[FlowNodeResumeState::CONTEXT_KEY]); - $resumeState = ($context[FlowResumeState::CONTEXT_KEY] ?? null); - unset($context[FlowResumeState::CONTEXT_KEY]); - if ($resumeState instanceof FlowResumeState === true) { - $storable = $resumeState->storableWhen(suspended: ($status === FlowRun::STATUS_SUSPENDED)); - if ($storable !== null) { - $context[FlowResumeState::CONTEXT_KEY] = $storable; - } - } + $this->keepResumeSlots(context: $context, status: $status); // A signal is consumed by the walk it woke. Kept, it would still be // sitting there the NEXT time this run suspends on a signal, and that @@ -1186,6 +1180,48 @@ private function persistResult(FlowRun $run, array $result): FlowRun { return $persisted; }//end persistResult() + /** + * Fold the walk's per-node resume slots back into the storable context. + * + * 🔴 LIVE, NOT SUSPENDED. The rule used to be "only a suspended run has + * anywhere to continue from", which quietly conflated NOT-SUSPENDED with + * TERMINAL. A pass legitimately ends `queued` while a node parked in an + * EARLIER pass is still waiting: the in-request advance of one branch + * finalises `queued` whenever a sibling has enabled work, and a claim + * refused on contention does the same. Dropping the slots there costs a + * task-waiting node the uuid of the task it is waiting on — and that loss + * IS the heartbeat wedge. The node's next wake finds an empty slot, so + * (correctly, by its own idempotency guard) it asks again; from then on + * the ORIGINAL task's completion can never address the node's slot, its + * signal is refused against the new slot's recorded assignee, and the run + * re-suspends on its heartbeat forever while a duplicate task sits in + * somebody's inbox. + * + * A terminal run still drops them, for the reason it always did: anything + * still held belongs to a node the run never came back to. + * + * @param array $context The context being persisted, modified in place. + * @param string $status The status the walk ended in. + * + * @return void + * + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-live-run-keeps-every-parked-nodes-resume-slot + */ + private function keepResumeSlots(array &$context, string $status): void { + $resumeState = ($context[FlowResumeState::CONTEXT_KEY] ?? null); + unset($context[FlowResumeState::CONTEXT_KEY]); + + if ($resumeState instanceof FlowResumeState === false) { + return; + } + + $storable = $resumeState->storableWhen(live: (in_array($status, FlowRun::TERMINAL, true) === false)); + if ($storable !== null) { + $context[FlowResumeState::CONTEXT_KEY] = $storable; + } + + }//end keepResumeSlots() + /** * The correlation key a suspended run can be addressed by, or null. * diff --git a/lib/Service/Flow/FlowTaskBridge.php b/lib/Service/Flow/FlowTaskBridge.php index 1c1e751ffb..885f68cb51 100644 --- a/lib/Service/Flow/FlowTaskBridge.php +++ b/lib/Service/Flow/FlowTaskBridge.php @@ -186,6 +186,60 @@ public function record(string $uuid, string $action, ?string $actor, string $rea $this->tasks->record(uuid: $uuid, action: $action, actor: $actor, reason: $reason); }//end record() + /** + * Record that a heartbeat, not the completion's signal, delivered a + * terminal task's answer to its run. + * + * The heartbeat exists precisely to recover a missed wake — a completion + * whose signal was refused (the assignee guard, a group that did not exist + * yet) or lost. When it does recover one, the audit must say so: the + * guarded signal seam records a refusal, and without this entry the trail + * ends there, reading as though the answer never reached the run at all. + * Attributed to whoever completed the task, because the fact being + * recorded is THEIR answer arriving — late, by poll — not the cron job's. + * + * Best-effort by design: the recovery itself is the node applying the + * outcome, and a failure to write the audit row must never turn a + * recovered run back into a wedged one. + * + * @param Task $task The terminal task whose outcome the heartbeat applied. + * + * @return void + * + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-heartbeat-recovered-delivery-is-recorded-on-the-tasks-audit + */ + public function recordHeartbeatRecovery(Task $task): void { + try { + $this->tasks->record( + uuid: (string)$task->getUuid(), + action: 'heartbeat-recovered', + actor: $task->getCompletedBy(), + reason: sprintf( + 'The completion signal never reached run %s; the heartbeat re-read this task and applied its outcome.', + (string)$task->getRunUuid() + ) + ); + + $this->logger->info( + message: '[FlowTaskBridge] Heartbeat recovered a missed completion signal', + context: [ + 'file' => __FILE__, + 'line' => __LINE__, + 'task' => (string)$task->getUuid(), + 'run' => (string)$task->getRunUuid(), + 'node' => (string)$task->getNodeId(), + 'completedBy' => (string)($task->getCompletedBy() ?? ''), + ] + ); + } catch (Throwable $failure) { + $this->logger->warning( + message: '[FlowTaskBridge] Could not record a heartbeat recovery on task ' . $task->getUuid() + . '; the outcome itself was applied: ' . $failure->getMessage(), + context: ['file' => __FILE__, 'line' => __LINE__, 'run' => (string)$task->getRunUuid()] + ); + }//end try + }//end recordHeartbeatRecovery() + /** * The task a node's resume slot points at, or null when it is gone. * diff --git a/lib/Service/Flow/Nodes/PortalTaskNode.php b/lib/Service/Flow/Nodes/PortalTaskNode.php index 9c628c9acc..abe7729271 100644 --- a/lib/Service/Flow/Nodes/PortalTaskNode.php +++ b/lib/Service/Flow/Nodes/PortalTaskNode.php @@ -76,6 +76,7 @@ use OCA\OpenRegister\Service\Flow\FlowItems; use OCA\OpenRegister\Service\Flow\FlowNodeResumeState; use OCA\OpenRegister\Service\Flow\FlowRunContext; +use OCA\OpenRegister\Service\Flow\FlowRunService; use OCA\OpenRegister\Service\Flow\FlowSuspension; use OCA\OpenRegister\Service\Flow\FlowTaskBridge; use OCA\OpenRegister\Service\Flow\IFlowNode; @@ -267,6 +268,7 @@ public function validateConfig(array $config): void { * the case names nobody, or a re-ask has no reason. * * @spec openspec/changes/flow-portal-task/specs/flow-portal-task/spec.md#requirement-a-portal-task-step-creates-one-external-task-and-suspends-the-run + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-heartbeat-wake-re-reads-the-awaited-task-and-applies-a-terminal-outcome */ public function execute(array $items, array $config, array $context): array { if ($items === []) { @@ -300,6 +302,14 @@ public function execute(array $items, array $config, array $context): array { } if ($resume->get(key: PortalTaskConfig::SLOT_PASSED_AT, default: null) === null) { + // A terminal read with no signal in hand means the completion's + // wake never arrived — refused or lost — and the heartbeat is what + // recovered it. Recorded before the outcome is applied, which is + // identical on both paths; the user-task node makes the same call. + if (array_key_exists(FlowRunService::SIGNAL_CONTEXT_KEY, $context) === false) { + $this->bridge->recordHeartbeatRecovery(task: $task); + } + // The first pass over a terminal task: the answer travels on. Marked // ONCE, so the next firing of this node in this run is a re-entry. $resume->set(key: PortalTaskConfig::SLOT_PASSED_AT, value: (new DateTime())->format('c')); diff --git a/lib/Service/Flow/Nodes/UserTaskNode.php b/lib/Service/Flow/Nodes/UserTaskNode.php index d7a309d962..7b79f74aed 100644 --- a/lib/Service/Flow/Nodes/UserTaskNode.php +++ b/lib/Service/Flow/Nodes/UserTaskNode.php @@ -65,6 +65,7 @@ use OCA\OpenRegister\Service\Flow\FlowItems; use OCA\OpenRegister\Service\Flow\FlowNodeResumeState; use OCA\OpenRegister\Service\Flow\FlowRunContext; +use OCA\OpenRegister\Service\Flow\FlowRunService; use OCA\OpenRegister\Service\Flow\FlowStop; use OCA\OpenRegister\Service\Flow\FlowSuspension; use OCA\OpenRegister\Service\Flow\FlowTaskBridge; @@ -270,6 +271,7 @@ public function validateConfig(array $config): void { * @throws RuntimeException When the node has no resume slot, or its task is gone. * * @spec openspec/changes/flow-user-task-node/specs/flow-user-task-node/spec.md#requirement-a-user-task-step-creates-exactly-one-task-and-suspends-the-run + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-heartbeat-wake-re-reads-the-awaited-task-and-applies-a-terminal-outcome */ public function execute(array $items, array $config, array $context): array { if ($items === []) { @@ -309,6 +311,15 @@ public function execute(array $items, array $config, array $context): array { throw $this->suspension(config: $config, items: $items); } + // A terminal read with no signal in hand means the completion's wake + // never arrived — refused by the assignee guard, or lost — and the + // heartbeat is what recovered it. Recorded on the task's audit, + // attributed to its completer, BEFORE the outcome is applied below: + // the applying is identical on both paths, which is the contract. + if (array_key_exists(FlowRunService::SIGNAL_CONTEXT_KEY, $context) === false) { + $this->bridge->recordHeartbeatRecovery(task: $task); + } + $bag = FlowTaskBridge::outcomeBagFor(task: $task); if ($bag['rejected'] === true && ($config['failOnReject'] ?? false) === true) { diff --git a/openspec/changes/flow-heartbeat-recovery/design.md b/openspec/changes/flow-heartbeat-recovery/design.md new file mode 100644 index 0000000000..2861336d40 --- /dev/null +++ b/openspec/changes/flow-heartbeat-recovery/design.md @@ -0,0 +1,97 @@ +# Design: flow-heartbeat-recovery + +## Context + +Measured before deciding anything: + +- **The node re-read has always existed.** `UserTaskNode::execute()` + (`lib/Service/Flow/Nodes/UserTaskNode.php`) re-enters on every wake, reads + `taskOrNull()` and applies the outcome bag when the task is terminal. + `PortalTaskNode` mirrors it. The single-stream walk and the stream walk + both re-dispatch a parked node on resume (`FlowStreamWalk::begin()`: + "Suspended streams become eligible again"). +- **The slot is the node's only memory.** The task uuid lives in the node's + per-node resume slot (#3325's scoping) and nowhere else the node can + reach. Lose the slot and the node MUST create a new task — its own + idempotency guard reads the slot to decide. +- **`persistResult()` was the only writer that lost it.** + `FlowResumeState::storableWhen(suspended:)` returned null for every + non-suspended pass end. `advanceStream()` finalising `queued` (sibling + work enabled) is a routine pass end for a run with parallel human + branches, and a claim refusal produces `queued` too. + +## Decisions + +### D-1: Fix the state, not the walk + +The heartbeat is made honest by making the state it reads durable, not by +adding a recovery sweep to `FlowRunWorker`. A worker-side "re-read every +suspended run's tasks" would be a second delivery mechanism with its own +addressing rules, racing the node's own re-read. With the slot intact, the +existing wake (`findDue()` → `advance()` → `execute()` → node re-entry) does +everything the defect report asks: re-read, apply, advance. + +`storableWhen()` keeps slots for every status outside `FlowRun::TERMINAL`. +Terminal runs still drop them, for the original reason: anything still held +belongs to a node the run never came back to, and the dispatcher has already +cleared every node that returned. + +### D-2: The recovery is audited on the task, attributed to the completer + +The guarded signal seam records a refusal +(`FlowRunSignalService::auditRefusal()`); without a matching entry the trail +ends there and reads as though the answer never reached the run. When a node +reads its task terminal with NO signal in the walk's context, the wake was a +heartbeat, not the completion's signal — `context['signal']` is set by +`signal()` and survives into the woken walk, so its absence is the +discriminator. The node then calls +`FlowTaskBridge::recordHeartbeatRecovery()`, which appends a +`heartbeat-recovered` audit entry on the task via `TaskService::record()`, +actor = `completedBy` — the fact recorded is THAT PERSON's answer arriving +late, not the cron job acting. Best-effort: a failure to write the entry is +logged and swallowed, because it must never turn a recovered run back into a +wedged one. + +### D-3: The symmetric cases need no new mechanism + +Stated explicitly, as the defect report asks: + +- **A task completed while the run was not yet suspended (the race).** + `signal()` returns null for a run that is `running` or `queued`, so the + completion's wake is lost. The run then parks with a non-null heartbeat + (`🔴 THE HEARTBEAT IS NEVER NULL`, UserTaskNode), and the next wake + re-reads the task — with the slot now durable, the race costs at most one + heartbeat period of latency. No pre-suspension re-check is added: the node + cannot read an answer before it has parked on the question, and the + heartbeat already bounds the wait. +- **A task whose sequence concluded (`TaskSequenceService`).** A sequence + drives every per-task transition through `TaskService`'s verbs, so the + task named by the node's slot reaches its terminal state on the same row + the heartbeat re-reads. Terminality is a property of that row + (`isInTerminalState()`); the re-read covers sequence-concluded tasks with + no sequence-specific handling. + +### D-4: Wrong-slot isolation is preserved by construction + +Each node reads only the slot the dispatcher scoped to it +(`FlowNodeResumeState`), so a heartbeat wake recovers exactly the nodes +whose OWN tasks are terminal; a sibling parked on an open task re-suspends +with its slot untouched. `FlowHeartbeatRecoveryTest::testOnlyTheNodeWhoseTaskEndedRecovers` +pins it. + +## Risks + +- Keeping slots on `queued`/`running` stores per-node state a little longer + than before. That state is exactly what a parked node needs on its next + wake; nodes that returned were already cleared by the dispatcher, so no + stale cursor can leak into a later pass. +- `recordHeartbeatRecovery()` re-dispatches `TaskTerminalEvent`, because + every `TaskService::record()` on a terminal task does. The re-entrancy is + closed by an existing guard rather than by a new one, and the chain is + short enough to state in full: the listener calls + `FlowTaskBridge::continueRun()`, which calls `FlowRunService::signal()`, + which returns null for any run that is not `suspended`. At the moment the + recovery is recorded the run row says `running` — `execute()` sets and + persists that before the walk begins — so the signal is refused and no + second walk starts. The recovery is written from inside that walk, and the + walk finishes normally. diff --git a/openspec/changes/flow-heartbeat-recovery/proposal.md b/openspec/changes/flow-heartbeat-recovery/proposal.md new file mode 100644 index 0000000000..801540cceb --- /dev/null +++ b/openspec/changes/flow-heartbeat-recovery/proposal.md @@ -0,0 +1,76 @@ +--- +kind: code +depends_on: [flow-user-task-node, flow-parallel-streams] +--- + +# Proposal: flow-heartbeat-recovery + +## Summary + +Make the user-task heartbeat honest: a suspended run whose completion signal +was refused or lost must recover on its next heartbeat wake instead of +re-suspending forever. The recovery mechanism already exists — the node +re-reads its task on every re-entry — but the state it depends on did not +survive: a pass that ends `queued` dropped every parked node's resume slot, +losing the uuid of the task the node was waiting on. This change keeps the +slots for every run that can still advance, and records a heartbeat-recovered +delivery on the task's audit so the trail no longer ends at the refusal. + +## Why + +**Observed on the acceptance rig, wedged forever.** A UserTask's completion +signal was refused (`[FlowRunSignalService] Refused a signal: the actor is +not the awaiting step's assignee` — the assignee group did not exist at +signal time). The suspended run's 30-minute heartbeat then fired, re-suspended +for another 30 minutes, and never advanced: `resume_at` rolled 08:07 → 08:37 +→ … while the task sat `completed` and the group had long been created. The +heartbeat exists precisely to recover a missed wake; it recovered nothing. + +**The node was never the problem.** `UserTaskNode::execute()` has always +re-read its task through `FlowTaskBridge::taskOrNull()` on every re-entry and +applied the outcome when the task is terminal — the unit suite proves it, and +a run driven through the real engine, dispatcher and stream commit path +recovers correctly (`FlowHeartbeatRecoveryTest`). What wedges is upstream: + +**`persistResult()` dropped every parked node's resume slot whenever a pass +ended anything but `suspended`.** `FlowResumeState::storableWhen(suspended:)` +read NOT-SUSPENDED as "nothing left to continue from", which conflates it +with TERMINAL. A pass legitimately ends `queued` while a node parked in an +EARLIER pass still waits: the in-request advance of a sibling branch +(`FlowTaskBridge::continueRun()` → `advanceStream()`) finalises `queued` +whenever other enabled work remains, and a claim refused on contention does +the same. The parked user-task node then lost its `taskUuid` slot; its next +wake found an empty slot and — exactly as its own guard demands — created a +NEW task rather than re-reading the original. From that moment: + +- the completion of the ORIGINAL task could never address the node's slot, + and its signal was refused against the new slot's recorded assignee — the + refusal observed on the rig; +- every heartbeat re-read the NEW, open task and re-suspended, rolling + `resume_at` forever; +- a duplicate task sat in somebody's inbox. + +`FlowHeartbeatRecoveryTest::testAnInRequestAdvanceKeepsTheSiblingNodesParkedSlot` +reproduces the drop red on the unfixed code. + +## What changes + +1. **Slots survive every live pass end.** `FlowResumeState::storableWhen()` + now keeps the per-node slots for any non-terminal status (`suspended`, + `queued`, `running`) and drops them only when the run is terminal. + `persistResult()` derives that from `FlowRun::TERMINAL`. +2. **A recovered delivery is recorded.** When `UserTaskNode` or + `PortalTaskNode` reads its task terminal WITHOUT a signal in hand (no + `context['signal']` — the completion's wake never arrived), it records + `heartbeat-recovered` on the task's audit through the new + `FlowTaskBridge::recordHeartbeatRecovery()`, attributed to the task's + `completedBy`. Best-effort: an audit failure never un-recovers the run. +3. **Nothing else.** No second delivery mechanism, no new sweep, no new + column: the heartbeat wake, the node re-read and the outcome application + are exactly the paths that already existed. + +## Out of scope + +Runs already wedged before this fix (slot lost, duplicate task created) +cannot be recovered retroactively: the original task's uuid is gone from the +run. They end at the abandoned-signal reaper or by manual retry, as today. diff --git a/openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md b/openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md new file mode 100644 index 0000000000..44cea2bff4 --- /dev/null +++ b/openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md @@ -0,0 +1,111 @@ +## Purpose + +A run suspended on a task must recover a missed completion signal on its +next heartbeat wake: the awaited task is re-read, a terminal outcome is +applied exactly as the signal path would have applied it, and the recovery +is recorded. A missed wake costs latency, never the run. + +## ADDED Requirements + +### Requirement: A live run keeps every parked node's resume slot + +The engine SHALL persist every node's resume slot across any pass end from +which the run can still advance — `suspended`, `queued` and `running` +alike. A pass that ends `queued` (an in-request advance of one branch while +a sibling has enabled work, a claim refused on contention) MUST NOT cost a +node parked in an earlier pass its stored progress; for a task-waiting node +that progress includes the uuid of the task it is waiting on, and losing it +forces a duplicate task and strands the original's completion. + +The engine SHALL drop the slots only when the run reaches a terminal +status: a finished run has nowhere to continue from, and the dispatcher has +already cleared every node that returned. + +#### Scenario: An in-request advance of one branch keeps the sibling's slot + +- **GIVEN** a run suspended on two parallel user-task nodes, each holding + its task's uuid in its own resume slot +- **WHEN** one task's completion advances its branch in-request and the pass + ends `queued` because the sibling branch still has enabled work +- **THEN** the sibling node's resume slot MUST still hold its task's uuid +- **AND** the sibling's next wake MUST re-park on that same task, never + create a second one + +#### Scenario: A terminal run stores no slots + +- **GIVEN** a run whose walk ends in a terminal status +- **THEN** no resume slots are persisted on the run + +### Requirement: A heartbeat wake re-reads the awaited task and applies a terminal outcome + +On every wake of a suspended run — heartbeat or signal — a task-waiting +node SHALL re-read the task named by its own resume slot. When that task is +terminal (completed, terminated, disabled), the node SHALL apply its +outcome exactly as the signal path would have: the same outcome bag under +`json.` on every item, the same advance of the run. When the +task is still open, the node SHALL suspend again on its heartbeat without +touching its slot. + +Recovery SHALL respect per-node slot addressing: only a node whose OWN task +is terminal advances; a sibling parked on an open task re-suspends with its +slot intact. + +The heartbeat is the recovery bound for the missed-signal cases, and no +second delivery mechanism SHALL be added for them: a completion that raced +the suspension (the run was not yet suspended when the signal was +attempted) and a task concluded by a task sequence both leave the task row +terminal, which the re-read observes within one heartbeat period. + +#### Scenario: A refused signal is recovered on the next heartbeat + +- **GIVEN** a run suspended on a user task whose completion signal was + refused, so the run never heard about the completion +- **WHEN** the run's heartbeat (`resume_at`) fires +- **THEN** the run MUST advance with the task's outcome on its items, + attributed to the task's completer +- **AND** no new task is created + +#### Scenario: A still-open task re-suspends unchanged + +- **GIVEN** a run suspended on a user task that is still open +- **WHEN** the heartbeat fires +- **THEN** the run suspends again on the same task, with the node's slot + (task uuid, askedAt) unchanged + +#### Scenario: Only the addressed node's slot recovers + +- **GIVEN** a run suspended on two user-task nodes, of which only one task + is terminal +- **WHEN** the heartbeat fires +- **THEN** the node whose task ended applies its outcome and advances its + branch +- **AND** the sibling re-suspends with its own slot intact + +### Requirement: A heartbeat-recovered delivery is recorded on the task's audit + +When a node applies a terminal task's outcome on a wake that carried no +signal — the completion's wake was refused or lost, and the heartbeat is +what recovered it — the engine SHALL record a `heartbeat-recovered` entry +on the task's audit trail, attributed to the task's `completedBy`. The +guarded signal seam already records the refusal; this entry is the other +half of that trail, so a recovered answer never reads as one that vanished. + +Recording SHALL be best-effort: a failure to write the audit entry MUST NOT +fail the recovery itself. + +A completion that arrived on its signal is the ordinary path and SHALL NOT +be recorded as a recovery. + +#### Scenario: The recovery is audited to the completer + +- **GIVEN** a suspended run whose awaited task was completed by a performer + while the completion signal never reached the run +- **WHEN** the heartbeat applies the outcome +- **THEN** the task's audit trail holds a `heartbeat-recovered` entry naming + that performer as actor + +#### Scenario: A signal-delivered completion records no recovery + +- **GIVEN** a suspended run woken by its task's completion signal +- **WHEN** the node applies the outcome +- **THEN** no `heartbeat-recovered` entry is written diff --git a/openspec/changes/flow-heartbeat-recovery/tasks.md b/openspec/changes/flow-heartbeat-recovery/tasks.md new file mode 100644 index 0000000000..bfe33b02ff --- /dev/null +++ b/openspec/changes/flow-heartbeat-recovery/tasks.md @@ -0,0 +1,37 @@ +# Tasks: flow-heartbeat-recovery + +## 1. Keep the slots alive + +- [x] 1.1 `FlowResumeState::storableWhen()` keeps the per-node slots for any + non-terminal status and drops them only on a terminal one; parameter + renamed `suspended` → `live` so the call site states the rule. +- [x] 1.2 `FlowRunService::persistResult()` derives liveness from + `FlowRun::TERMINAL` and passes it through, with a comment naming the + wedge the old `suspended`-only rule caused. + +## 2. Record the recovery + +- [x] 2.1 `FlowTaskBridge::recordHeartbeatRecovery()`: append a + `heartbeat-recovered` entry on the task's audit via + `TaskService::record()`, actor = the task's `completedBy`, reason + naming the run; log the recovery; swallow and log an audit failure. +- [x] 2.2 `UserTaskNode::execute()`: on a terminal read with no + `context['signal']`, call `recordHeartbeatRecovery()` before applying + the outcome, which stays identical on both paths. +- [x] 2.3 `PortalTaskNode::execute()`: the same call on its first pass over + a terminal task — the two nodes share the wedge. + +## 3. Prove it + +- [x] 3.1 `tests/Unit/Service/Flow/FlowHeartbeatRecoveryTest.php` — the + wedge reproduction driven through the real engine, dispatcher, node, + stream walk, claims and commit over in-memory mappers: + the in-request advance keeps the sibling's slot (RED on the unfixed + `storableWhen`), the heartbeat recovers a refused signal with + attribution, a still-open task re-parks unchanged, and only the + addressed node's slot recovers. +- [x] 3.2 `UserTaskNodeTest` — a heartbeat-recovered completion is audited + to the completer; a signal-delivered completion records no recovery. +- [x] 3.3 `PortalTaskNodeTest` — the same pair for the portal node. +- [x] 3.4 `FlowResumeStateTest` — `storableWhen()` keeps slots while live, + drops them on terminal, stores nothing when empty. diff --git a/tests/Unit/Service/Flow/FlowHeartbeatRecoveryTest.php b/tests/Unit/Service/Flow/FlowHeartbeatRecoveryTest.php new file mode 100644 index 0000000000..17ed44d8a7 --- /dev/null +++ b/tests/Unit/Service/Flow/FlowHeartbeatRecoveryTest.php @@ -0,0 +1,514 @@ + + * SPDX-License-Identifier: EUPL-1.2 + * + * @category Test + * @package OCA\OpenRegister\Tests\Unit\Service\Flow + * + * @author Conduction Development Team + * @copyright 2026 Conduction B.V. + * @license EUPL-1.2 https://joinup.ec.europa.eu/collection/eupl/eupl-text-eupl-12 + * + * @link https://OpenRegister.app + * + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md + */ + +declare(strict_types=1); + +namespace OCA\OpenRegister\Tests\Unit\Service\Flow; + +use OCA\OpenRegister\Db\FlowClaim; +use OCA\OpenRegister\Db\FlowClaimMapper; +use OCA\OpenRegister\Db\FlowRun; +use OCA\OpenRegister\Db\FlowRunMapper; +use OCA\OpenRegister\Db\FlowRunStep; +use OCA\OpenRegister\Db\FlowRunStepMapper; +use OCA\OpenRegister\Db\FlowStream; +use OCA\OpenRegister\Db\FlowStreamMapper; +use OCA\OpenRegister\Db\Task; +use OCA\OpenRegister\Service\Flow\FlowDefinitionBuilder; +use OCA\OpenRegister\Service\Flow\FlowEngine; +use OCA\OpenRegister\Service\Flow\FlowItems; +use OCA\OpenRegister\Service\Flow\FlowNodeRegistry; +use OCA\OpenRegister\Service\Flow\FlowPlaceClaims; +use OCA\OpenRegister\Service\Flow\FlowRunCommit; +use OCA\OpenRegister\Service\Flow\FlowRunService; +use OCA\OpenRegister\Service\Flow\FlowTaskBridge; +use OCA\OpenRegister\Service\Flow\IFlowNode; +use OCA\OpenRegister\Service\Flow\Nodes\UserTaskNode; +use OCA\OpenRegister\Service\Flow\RegisterFlowNodesEvent; +use OCA\OpenRegister\Service\Flow\Timer\FlowTimerService; +use OCA\OpenRegister\Service\Task\TaskForm; +use OCA\OpenRegister\Service\Task\TaskFormReader; +use OCP\EventDispatcher\Event; +use OCP\EventDispatcher\IEventDispatcher; +use OCP\IDBConnection; +use OCP\IL10N; +use OCP\IURLGenerator; +use PHPUnit\Framework\TestCase; +use Psr\Container\ContainerInterface; +use Psr\Log\LoggerInterface; +use Psr\Log\NullLogger; + +/** A subject carrying nothing — the marking lives on the run. */ +class HeartbeatSubject { +} + +/** A step that passes items through, to split the graph. */ +class HeartbeatPassNode implements IFlowNode { + + public function getId(): string { + return 'test.pass'; + } + + public function getDisplayName(): string { + return 'Pass'; + } + + public function getDescription(): string { + return 'Passes items through.'; + } + + public function getIcon(): string { + return 'i.svg'; + } + + public function isAvailableForScope(int $scope): bool { + return true; + } + + public function validateConfig(array $config): void { + } + + public function execute(array $items, array $config, array $context): array { + return $items; + } +}//end class + +/** + * The wedge, reproduced and recovered. + */ +class FlowHeartbeatRecoveryTest extends TestCase { + use PublishedVersionDouble; + + private FlowRunService $service; + + /** + * The tasks the bridge "persisted", by uuid. + * + * @var array + */ + private array $tasks = []; + + /** + * Node ids handed to createTask, in call order. Growing past one entry + * per user-task node is the duplicate-task defect. + * + * @var array + */ + private array $created = []; + + /** + * Task uuids handed to recordHeartbeatRecovery, in call order. + * + * @var array + */ + private array $recovered = []; + + /** + * The "database": one run row, its streams, its claims. + * + * @var FlowRun|null + */ + private ?FlowRun $row = null; + + /** @var array */ + private array $streams = []; + + /** @var array */ + private array $claims = []; + + protected function setUp(): void { + parent::setUp(); + $mapper = $this->createMock(FlowRunMapper::class); + $mapper->method('insert')->willReturnCallback(function (FlowRun $run): FlowRun { + $this->row = $run; + return $run; + }); + $mapper->method('update')->willReturnCallback(function (FlowRun $run): FlowRun { + $this->row = $run; + return $run; + }); + $mapper->method('lockByUuid')->willReturnCallback(fn (): FlowRun => $this->row); + + $bridge = $this->createMock(FlowTaskBridge::class); + $bridge->method('createTask')->willReturnCallback(function (array $data, string $runUuid, string $nodeId, ?string $actor): Task { + $this->created[] = $nodeId; + $uuid = sprintf('t-%s-%d', $nodeId, count($this->created)); + $task = new Task(); + $task->setUuid($uuid); + $task->setState(Task::STATE_ACTIVE); + $task->setAssignee('alice'); + $task->setRunUuid($runUuid); + $task->setNodeId($nodeId); + $this->tasks[$uuid] = $task; + + return $task; + }); + $bridge->method('taskOrNull')->willReturnCallback(fn (string $uuid): ?Task => ($this->tasks[$uuid] ?? null)); + $bridge->method('recordHeartbeatRecovery')->willReturnCallback(function (Task $task): void { + $this->recovered[] = (string)$task->getUuid(); + }); + + $l10n = $this->createMock(IL10N::class); + $l10n->method('t')->willReturnArgument(0); + $forms = $this->createMock(TaskFormReader::class); + $forms->method('fromConfig')->willReturn(new TaskForm(kind: null)); + + $node = new UserTaskNode( + $bridge, + $l10n, + $this->createMock(IURLGenerator::class), + $forms, + $this->createMock(FlowTimerService::class) + ); + + $pass = new HeartbeatPassNode(); + $dispatcher = $this->createMock(IEventDispatcher::class); + $dispatcher->method('dispatchTyped')->willReturnCallback( + static function (Event $event) use ($node, $pass): void { + if ($event instanceof RegisterFlowNodesEvent) { + $event->registerNode($node); + $event->registerNode($pass); + } + } + ); + + $registry = new FlowNodeRegistry($dispatcher, $this->createMock(LoggerInterface::class)); + $engine = new FlowEngine(new FlowDefinitionBuilder(), $this->createMock(LoggerInterface::class)); + + $container = $this->createMock(ContainerInterface::class); + $versions = $this->publishedVersionMapper(); + $pin = $this->pinReturning(); + $container->method('get')->willReturnCallback( + function (string $id) use ($versions, $pin): object { + if ($id === \OCA\OpenRegister\Db\FlowVersionMapper::class) { + return $versions; + } + + if ($id === \OCA\OpenRegister\Service\Flow\FlowDefinitionPin::class) { + return $pin; + } + + throw new \RuntimeException('not available'); + } + ); + + $db = $this->createMock(IDBConnection::class); + $db->method('inTransaction')->willReturn(false); + + $streamMapper = $this->createMock(FlowStreamMapper::class); + $streamMapper->method('findByRun')->willReturnCallback(function (): array { + $list = array_values($this->streams); + usort($list, static fn (FlowStream $a, FlowStream $b): int => strcmp((string)$a->getOrdinalPath(), (string)$b->getOrdinalPath())); + return $list; + }); + $streamMapper->method('findByRunAndStream')->willReturnCallback(fn (string $runUuid, string $streamId): ?FlowStream => ($this->streams[$streamId] ?? null)); + $streamMapper->method('insert')->willReturnCallback(function (FlowStream $stream): FlowStream { + $this->streams[(string)$stream->getStreamId()] = $stream; + return $stream; + }); + $streamMapper->method('update')->willReturnCallback(function (FlowStream $stream): FlowStream { + $this->streams[(string)$stream->getStreamId()] = $stream; + return $stream; + }); + $streamMapper->method('allocateNextSequence')->willReturnCallback(function (string $runUuid, string $streamId): int { + $stream = ($this->streams[$streamId] ?? null); + if ($stream === null) { + return 0; + } + + $next = (int)$stream->getNextSequence(); + $stream->setNextSequence($next + 1); + return $next; + }); + + $claimMapper = $this->createMock(FlowClaimMapper::class); + $claimMapper->method('countHeldForRun')->willReturn(0); + $claimMapper->method('countHeldByOwner')->willReturn(0); + $claimMapper->method('insertOrRefuse')->willReturnCallback(function (FlowClaim $claim): bool { + $this->claims[] = $claim; + return true; + }); + $claimMapper->method('findByRun')->willReturnCallback(fn (): array => array_values($this->claims)); + $claimMapper->method('release')->willReturnCallback(function (string $runUuid, array $places): int { + $before = count($this->claims); + $this->claims = array_values(array_filter($this->claims, static fn (FlowClaim $c): bool => in_array($c->getPlace(), $places, true) === false)); + return ($before - count($this->claims)); + }); + $claimMapper->method('releaseByOwner')->willReturnCallback(function (string $runUuid, string $owner): int { + $before = count($this->claims); + $this->claims = array_values(array_filter($this->claims, static fn (FlowClaim $c): bool => $c->getOwner() !== $owner)); + return ($before - count($this->claims)); + }); + + $stepMapper = $this->createMock(FlowRunStepMapper::class); + $stepMapper->method('highestSequence')->willReturn(0); + $stepMapper->method('insert')->willReturnCallback(static fn (FlowRunStep $step): FlowRunStep => $step); + + $commit = new FlowRunCommit( + db: $db, + runs: $mapper, + streams: $streamMapper, + claims: $claimMapper, + steps: $stepMapper, + logger: new NullLogger() + ); + + $this->service = new FlowRunService( + $mapper, + $this->createMock(\OCA\OpenRegister\Db\FlowStateMapper::class), + $engine, + $registry, + $this->createMock(LoggerInterface::class), + $container, + null, + null, + $streamMapper, + new FlowPlaceClaims(claims: $claimMapper, db: $db, logger: new NullLogger()), + $commit + ); + }//end setUp() + + /** + * A split into two parallel user-task branches — the shape whose sibling + * completion ends a pass `queued` and used to drop the other slot. + * + * @return array The flow document. + */ + private function flow(): array { + return [ + 'id' => 'f1', + 'nodes' => [ + ['id' => 'start', 'type' => 'test.pass'], + [ + 'id' => 'askA', + 'type' => 'openregister.user-task', + 'config' => ['title' => 'Approve A', 'assignee' => 'alice', 'heartbeatMinutes' => 30], + ], + [ + 'id' => 'askB', + 'type' => 'openregister.user-task', + 'config' => ['title' => 'Approve B', 'assignee' => 'alice', 'heartbeatMinutes' => 30, 'outcomeKey' => 'taskB'], + ], + ], + 'edges' => [ + ['id' => 'e1', 'from' => 'start', 'to' => 'askA'], + ['id' => 'e2', 'from' => 'start', 'to' => 'askB'], + ], + ]; + }//end flow() + + /** + * Park both branches on their freshly created tasks. + * + * @return FlowRun The suspended run. + */ + private function suspendedOnBothTasks(): FlowRun { + $run = $this->service->queue('f1', user: 'alice'); + $run = $this->service->execute( + $run, + $this->flow(), + new HeartbeatSubject(), + seedItems: [FlowItems::item(json: ['name' => 'Case 7'])] + ); + + $this->assertSame(FlowRun::STATUS_SUSPENDED, $run->getStatus()); + $this->assertSame(['askA', 'askB'], $this->created); + + return $run; + }//end suspendedOnBothTasks() + + /** + * Mark a task terminal, as its completion verb would have left it. + * + * @param string $uuid The task. + * @param string $completedBy Who answered. + * + * @return void + */ + private function complete(string $uuid, string $completedBy): void { + $this->tasks[$uuid]->setState(Task::STATE_COMPLETED); + $this->tasks[$uuid]->setIsTerminal(true); + $this->tasks[$uuid]->setOutcome('approved'); + $this->tasks[$uuid]->setCompletedBy($completedBy); + }//end complete() + + /** + * The live stream whose token stands on a place. + * + * @param string $place The place. + * + * @return string The stream id. + */ + private function streamOn(string $place): string { + foreach ($this->streams as $stream) { + if ((string)$stream->getPlace() === $place && $stream->isTerminal() === false) { + return (string)$stream->getStreamId(); + } + } + + $this->fail(sprintf('No live stream stands on place "%s".', $place)); + }//end streamOn() + + /** + * 🔴 THE WEDGE'S ROOT CAUSE, proven red before the fix: an in-request + * advance of one branch ends the pass `queued` while the sibling branch + * still has enabled work, and that pass end used to DROP the sibling's + * resume slot — the uuid of the task it was waiting on. From there the + * sibling asked again on its next wake, the original task's completion + * signal was refused against the new slot's assignee, and the run rolled + * its heartbeat forever. + * + * @return void + * + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-live-run-keeps-every-parked-nodes-resume-slot + */ + public function testAnInRequestAdvanceKeepsTheSiblingNodesParkedSlot(): void { + $run = $this->suspendedOnBothTasks(); + $slots = ($run->getContext()['resumeState'] ?? []); + $taskB = (string)($slots['askB']['taskUuid'] ?? ''); + $this->assertNotSame('', $taskB); + + // Task A completes; its completion signals the run and spends the + // node's advance budget in-request, as FlowTaskBridge::continueRun() + // does. Branch B still has enabled work, so this pass ends `queued`. + $this->complete(uuid: (string)$slots['askA']['taskUuid'], completedBy: 'bob'); + $woken = $this->service->signal($run, []); + $this->assertNotNull($woken); + $run = $this->service->advanceStream($woken, $this->flow(), new HeartbeatSubject(), $this->streamOn('askA'), 'all'); + + $this->assertSame(FlowRun::STATUS_QUEUED, $run->getStatus()); + $kept = ($run->getContext()['resumeState'] ?? []); + $this->assertSame( + $taskB, + (string)($kept['askB']['taskUuid'] ?? ''), + 'a queued pass end must keep the sibling\'s parked slot, or the heartbeat loses the task it is waiting on' + ); + + // A signal-delivered completion is not a heartbeat recovery. + $this->assertSame([], $this->recovered); + + // The worker's next pass re-parks branch B on the SAME task — never a + // duplicate in somebody's inbox. + $run = $this->service->execute($run, $this->flow(), new HeartbeatSubject()); + $this->assertSame(FlowRun::STATUS_SUSPENDED, $run->getStatus()); + $this->assertSame(['askA', 'askB'], $this->created, 'a wake must never create a second task for a parked node'); + $this->assertSame($taskB, (string)($run->getContext()['resumeState']['askB']['taskUuid'] ?? '')); + }//end testAnInRequestAdvanceKeepsTheSiblingNodesParkedSlot() + + /** + * The heartbeat's whole reason to exist: a completion whose signal was + * refused or lost is recovered on the next wake — the node re-reads its + * task, applies the outcome exactly as the signal path would have (same + * bag under `json.`, same advance), and the recovery is + * recorded on the task's audit attributed to its completer. + * + * @return void + * + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-heartbeat-wake-re-reads-the-awaited-task-and-applies-a-terminal-outcome + */ + public function testTheHeartbeatRecoversACompletionWhoseSignalWasRefused(): void { + $run = $this->suspendedOnBothTasks(); + $slots = ($run->getContext()['resumeState'] ?? []); + $taskA = (string)$slots['askA']['taskUuid']; + $taskB = (string)$slots['askB']['taskUuid']; + + // Both tasks complete, but NO signal ever reaches the run — the + // observed case: the assignee guard refused the delivery. + $this->complete(uuid: $taskA, completedBy: 'bob'); + $this->complete(uuid: $taskB, completedBy: 'carol'); + + // The heartbeat fires: findDue() → advance() → execute(). + $run = $this->service->execute($run, $this->flow(), new HeartbeatSubject()); + + $this->assertSame(FlowRun::STATUS_COMPLETED, $run->getStatus()); + $bagA = ($run->getItems()[0]['json']['task'] ?? null); + $bagB = ($run->getItems()[0]['json']['taskB'] ?? null); + $this->assertSame('approved', ($bagA['outcome'] ?? ($bagB['outcome'] ?? null))); + $this->assertSame(['askA', 'askB'], $this->created, 'recovery must never create a task'); + $this->assertEqualsCanonicalizing([$taskA, $taskB], $this->recovered, 'each recovered delivery is audited, attributed to its completer'); + }//end testTheHeartbeatRecoversACompletionWhoseSignalWasRefused() + + /** + * A heartbeat that finds the task still open is a re-suspend, not an + * answer: same task, same slot, no audit entry, still suspended. + * + * @return void + * + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-heartbeat-wake-re-reads-the-awaited-task-and-applies-a-terminal-outcome + */ + public function testAHeartbeatWakeWithTheTasksStillOpenParksAgainOnTheSameTasks(): void { + $run = $this->suspendedOnBothTasks(); + $before = ($run->getContext()['resumeState'] ?? []); + + $run = $this->service->execute($run, $this->flow(), new HeartbeatSubject()); + + $this->assertSame(FlowRun::STATUS_SUSPENDED, $run->getStatus()); + $this->assertSame(['askA', 'askB'], $this->created); + $this->assertSame([], $this->recovered); + $after = ($run->getContext()['resumeState'] ?? []); + $this->assertSame($before['askA']['taskUuid'], $after['askA']['taskUuid']); + $this->assertSame($before['askB']['taskUuid'], $after['askB']['taskUuid']); + $this->assertSame($before['askA']['askedAt'], $after['askA']['askedAt'], 'a heartbeat must not restamp askedAt'); + }//end testAHeartbeatWakeWithTheTasksStillOpenParksAgainOnTheSameTasks() + + /** + * Per-node slot addressing holds through a recovery: only the node whose + * task ended advances; its sibling re-parks on its own task, slot intact. + * + * @return void + * + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-heartbeat-wake-re-reads-the-awaited-task-and-applies-a-terminal-outcome + */ + public function testOnlyTheNodeWhoseTaskEndedRecovers(): void { + $run = $this->suspendedOnBothTasks(); + $slots = ($run->getContext()['resumeState'] ?? []); + $taskA = (string)$slots['askA']['taskUuid']; + $taskB = (string)$slots['askB']['taskUuid']; + + // Only task A completed, and its signal never arrived. + $this->complete(uuid: $taskA, completedBy: 'bob'); + + $run = $this->service->execute($run, $this->flow(), new HeartbeatSubject()); + + $this->assertSame(FlowRun::STATUS_SUSPENDED, $run->getStatus(), 'branch B still waits'); + $this->assertSame([$taskA], $this->recovered, 'only the addressed node\'s slot recovers'); + $this->assertSame(['askA', 'askB'], $this->created); + $kept = ($run->getContext()['resumeState'] ?? []); + $this->assertArrayNotHasKey('askA', $kept, 'a node that answered has nothing left to remember'); + $this->assertSame($taskB, (string)($kept['askB']['taskUuid'] ?? ''), 'the waiting sibling keeps its own task'); + }//end testOnlyTheNodeWhoseTaskEndedRecovers() +}//end class diff --git a/tests/Unit/Service/Flow/FlowResumeStateTest.php b/tests/Unit/Service/Flow/FlowResumeStateTest.php index 70d05a202a..3c0f893da4 100644 --- a/tests/Unit/Service/Flow/FlowResumeStateTest.php +++ b/tests/Unit/Service/Flow/FlowResumeStateTest.php @@ -156,6 +156,45 @@ public function testFromArrayPassesAnExistingStateThrough(): void { $this->assertSame($state, FlowResumeState::fromArray($state)); } + /** + * Slots survive every pass end the run can still advance from — a pass + * that ends `queued` (an in-request advance whose sibling has enabled + * work, a claim refused on contention) must not cost a parked node the + * uuid of the task it is waiting on. Dropping it there was the heartbeat + * wedge: the node asked again, and the original task's completion could + * never address the slot again. + * + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-live-run-keeps-every-parked-nodes-resume-slot + */ + public function testSlotsAreStorableWhileTheRunIsLive(): void { + $state = new FlowResumeState(); + $state->forNode(nodeId: 'ask')->set(key: 'taskUuid', value: 't-1'); + + $this->assertSame(['ask' => ['taskUuid' => 't-1']], $state->storableWhen(live: true)); + } + + /** + * A terminal run drops its slots: anything still held belongs to a node + * the run never came back to, and keeping it would put a stale cursor in + * front of anyone reading the finished run. + * + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-live-run-keeps-every-parked-nodes-resume-slot + */ + public function testATerminalRunDropsItsSlots(): void { + $state = new FlowResumeState(); + $state->forNode(nodeId: 'ask')->set(key: 'taskUuid', value: 't-1'); + + $this->assertNull($state->storableWhen(live: false)); + } + + /** + * Nothing held is nothing stored, live or not — an empty bag must not + * write an empty key into every run's context. + */ + public function testAnEmptyStateStoresNothing(): void { + $this->assertNull((new FlowResumeState())->storableWhen(live: true)); + } + /** * The scoped view is what a node is handed, and it must not be able to * name another node's slot: there is no API on it that takes a node id. diff --git a/tests/Unit/Service/Flow/PortalTaskNodeTest.php b/tests/Unit/Service/Flow/PortalTaskNodeTest.php index 7d4a9ac8ae..36df0d9936 100644 --- a/tests/Unit/Service/Flow/PortalTaskNodeTest.php +++ b/tests/Unit/Service/Flow/PortalTaskNodeTest.php @@ -34,6 +34,7 @@ use OCA\OpenRegister\Service\Flow\FlowNodeResumeState; use OCA\OpenRegister\Service\Flow\FlowResumeState; use OCA\OpenRegister\Service\Flow\FlowRunContext; +use OCA\OpenRegister\Service\Flow\FlowRunService; use OCA\OpenRegister\Service\Flow\FlowSuspension; use OCA\OpenRegister\Service\Flow\FlowTaskBridge; use OCA\OpenRegister\Service\Flow\Nodes\PortalTaskConfig; @@ -60,6 +61,7 @@ * @uses \OCA\OpenRegister\Service\Flow\FlowAdvanceBudget * @uses \OCA\OpenRegister\Service\Flow\FlowItems * @uses \OCA\OpenRegister\Service\Flow\FlowTaskBridge + * @uses \OCA\OpenRegister\Service\Flow\FlowRunService * @uses \OCA\OpenRegister\Service\Flow\FlowValueTemplate * @uses \OCA\OpenRegister\Service\Task\TaskState * @uses \OCA\OpenRegister\Db\PortalTaskDelivery @@ -467,6 +469,50 @@ public function testACompletedTaskPlacesTheAnswerOnEveryItemAndMarksThePass(): v $this->assertNotNull($state->read(nodeId: 'ask')[PortalTaskConfig::SLOT_PASSED_AT], 'the pass is marked so the next firing is a re-entry'); }//end testACompletedTaskPlacesTheAnswerOnEveryItemAndMarksThePass() + /** + * A terminal read with no signal in hand is the heartbeat recovering a + * missed wake, and the recovery lands on the task's audit — the same call + * the user-task node makes, because the two share the wedge. + * + * @return void + * + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-heartbeat-recovered-delivery-is-recorded-on-the-tasks-audit + */ + public function testAHeartbeatRecoveredAnswerIsAuditedOnTheTask(): void { + $state = new FlowResumeState(); + $state->write(nodeId: 'ask', values: [FlowTaskBridge::SLOT_TASK_UUID => 't-1', PortalTaskConfig::SLOT_CYCLE => 1]); + $task = $this->task(state: Task::STATE_COMPLETED); + $task->setOutcome('submitted'); + $task->setCompletedBy('party:bsn-1'); + $this->bridge->method('taskOrNull')->willReturn($task); + $this->bridge->expects($this->once())->method('recordHeartbeatRecovery')->with($task); + + $this->node->execute($this->items(), $this->config(), $this->context($state)); + }//end testAHeartbeatRecoveredAnswerIsAuditedOnTheTask() + + /** + * An answer that arrived on its signal is the ordinary path: no recovery + * entry lands on the task's audit. + * + * @return void + * + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-heartbeat-recovered-delivery-is-recorded-on-the-tasks-audit + */ + public function testASignalDeliveredAnswerRecordsNoHeartbeatRecovery(): void { + $state = new FlowResumeState(); + $state->write(nodeId: 'ask', values: [FlowTaskBridge::SLOT_TASK_UUID => 't-1', PortalTaskConfig::SLOT_CYCLE => 1]); + $task = $this->task(state: Task::STATE_COMPLETED); + $task->setOutcome('submitted'); + $this->bridge->method('taskOrNull')->willReturn($task); + $this->bridge->expects($this->never())->method('recordHeartbeatRecovery'); + + $this->node->execute( + $this->items(), + $this->config(), + $this->context($state, extra: [FlowRunService::SIGNAL_CONTEXT_KEY => []]) + ); + }//end testASignalDeliveredAnswerRecordsNoHeartbeatRecovery() + /** * An expiry-terminated task continues the run distinguishably from an answer. * diff --git a/tests/Unit/Service/Flow/UserTaskNodeTest.php b/tests/Unit/Service/Flow/UserTaskNodeTest.php index 6260e97d44..3ef0647533 100644 --- a/tests/Unit/Service/Flow/UserTaskNodeTest.php +++ b/tests/Unit/Service/Flow/UserTaskNodeTest.php @@ -461,6 +461,50 @@ public function testACompletedTaskContinuesWithTheOutcomeOnEveryItem(): void { } }//end testACompletedTaskContinuesWithTheOutcomeOnEveryItem() + /** + * A terminal read with no signal in hand is the heartbeat recovering a + * missed wake — the completion's signal was refused or lost — and that + * recovery is recorded on the task's audit, attributed to its completer. + * + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-heartbeat-recovered-delivery-is-recorded-on-the-tasks-audit + */ + public function testAHeartbeatRecoveredCompletionIsAuditedOnTheTask(): void { + $state = new FlowResumeState(); + $state->forNode(nodeId: 'ask')->set(key: FlowTaskBridge::SLOT_TASK_UUID, value: 't-1'); + $done = $this->task(state: Task::STATE_COMPLETED); + $done->setOutcome('approved'); + $done->setCompletedBy('bob'); + $this->bridge->method('taskOrNull')->willReturn($done); + $this->bridge->expects($this->once())->method('recordHeartbeatRecovery')->with($done); + + $out = $this->node->execute($this->items(), $this->config(), $this->context($state)); + + $this->assertSame('approved', $out[0][FlowItems::JSON]['task']['outcome'], 'the recovery applies the outcome exactly as the signal path would'); + }//end testAHeartbeatRecoveredCompletionIsAuditedOnTheTask() + + /** + * A completion whose signal DID arrive is the ordinary path, not a + * recovery: nothing extra lands on the task's audit. + * + * @spec openspec/changes/flow-heartbeat-recovery/specs/flow-heartbeat-recovery/spec.md#requirement-a-heartbeat-recovered-delivery-is-recorded-on-the-tasks-audit + */ + public function testASignalDeliveredCompletionRecordsNoHeartbeatRecovery(): void { + $state = new FlowResumeState(); + $state->forNode(nodeId: 'ask')->set(key: FlowTaskBridge::SLOT_TASK_UUID, value: 't-1'); + $done = $this->task(state: Task::STATE_COMPLETED); + $done->setOutcome('approved'); + $this->bridge->method('taskOrNull')->willReturn($done); + $this->bridge->expects($this->never())->method('recordHeartbeatRecovery'); + + $out = $this->node->execute( + $this->items(), + $this->config(), + $this->context($state, extra: [FlowRunService::SIGNAL_CONTEXT_KEY => []]) + ); + + $this->assertSame('approved', $out[0][FlowItems::JSON]['task']['outcome']); + }//end testASignalDeliveredCompletionRecordsNoHeartbeatRecovery() + /** * A delegated completion names both identities: the deputy who acted and * the person they acted for. A four-eyes rule cannot be enforced without