fix(flows): settle dedup nodes from the graph the run actually executed (#5268 item 2) - #5413
fix(flows): settle dedup nodes from the graph the run actually executed (#5268 item 2)#5413Mustaqeem66 wants to merge 5 commits into
Conversation
Moves the `mod tests` block out of `flows/bus.rs` into a sibling
`bus_tests.rs`, using the `#[cfg(test)] #[path = "..."] mod tests;`
convention already used throughout this crate (`subconscious/factory.rs`,
`skills/ops.rs`, `config/schema/channels.rs`, `platform/about_app/catalog.rs`,
and ~170 other sites).
Pure move: the test bodies are unchanged apart from being dedented one
level now that they are no longer nested inside an inline `mod tests { .. }`.
Also adds the tests for the tinyhumansai#5268 item 2 fix (settle dedup state from the
graph the run actually executed).
Closes tinyhumansai#5268 (item 2). `DedupCommitSubscriber` settled a finished run's `dedup` nodes by reading the flow's CURRENT saved graph. A flow edited while a run was still in flight was therefore mis-settled: a `dedup` node the run had written `tentative` keys under, but which was deleted or renamed before `FlowRunFinished` fired, was no longer found — so those keys were neither committed nor released, and the items silently reprocessed on the flow's next run. The converse also held: a node added after the run started was settled by a run that never executed it. `DedupCommitSubscriber` now also handles `DomainEvent::FlowRunStarted` (already published by `flows::ops::flows_run{,_detached}` right after the `flow_runs` row insert, and already tagged "cron" so the existing domain filter admits it) and pins that run's dedup node ids into the existing per-flow `flow_state` KV table under `dedup::run_nodes:<run_id>`. Settlement prefers that snapshot and clears it once the run is settled. - No schema migration: reuses the `flow_state` table the dedup keys already live in. The `dedup::` double-colon prefix cannot collide with the engine's `dedup:<node_id>:<committed|tentative>` keys. - Zero added I/O for the common case: a graph with no `dedup` node writes no snapshot at all, so an absent key stays an unambiguous "no snapshot". - Falls back to the saved graph when no snapshot exists (runs in flight across the upgrade, resumed runs, or a failed snapshot write), so the degraded path is exactly the historical behaviour. Also moves the test module to `bus_tests.rs` per the crate-wide `#[cfg(test)] #[path = "..."] mod tests;` convention.
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughThe flow event bus now captures dedup-node IDs at run start, uses each snapshot during completion, serializes settlement per flow, and removes snapshots afterward. Tests moved to ChangesRun-scoped dedup settlement
Estimated code review effort: 4 (Complex) | ~45 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
🧹 Nitpick comments (4)
src/openhuman/flows/bus.rs (3)
791-795: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winThe comment overstates what the in-lock delete prevents.
A duplicate
FlowRunFinishedthat arrives after this delete does not stop at "no snapshot".dedup_node_idsfalls back to the flow's saved graph and settles again. The delete therefore changes which graph a duplicate settles against; it does not prevent a second settlement.The behavior is still safe, because
commitunions into a set andreleasedeletes an already-absent key. Reword the comment to state that, so a future reader does not rely on a guarantee this code does not provide.♻️ Proposed comment
- // Still inside the per-flow lock: the snapshot is this run's alone, so - // no other run contends for it, but keeping the delete here means a - // settled run never leaves a readable snapshot behind for a - // late-arriving duplicate `FlowRunFinished` to settle a second time. + // Still inside the per-flow lock: the snapshot is this run's alone, so + // no other run contends for it. Deleting here keeps the row from + // outliving the run it describes. A duplicate `FlowRunFinished` is + // still handled — it falls back to the saved graph — and settling + // twice is harmless: `commit` unions into a set and `release` deletes + // an already-absent key.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/flows/bus.rs` around lines 791 - 795, Rewrite the comment immediately above self.clear_run_snapshot in the per-flow lock to remove the claim that deleting the snapshot prevents duplicate settlement. State that late duplicate FlowRunFinished events may fall back to the saved flow graph, while commit’s set-union behavior and release’s idempotent deletion keep repeated settlement safe.
757-763: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueSnapshot rows leak when a run never reaches settlement.
clear_run_snapshotruns only at line 795. Two paths skip it:
- The process dies between
FlowRunStartedandFlowRunFinished, so no settlement ever happens.node_idsis empty at line 760, sohandle_finishedreturns before line 795. This happens when the snapshot is unusable and the saved graph also holds no dedup node.Each skipped path leaves one
flow_staterow keyed by arun_idthat is never reused. The rows are small, and correctness is unaffected. Over a long-lived install with frequent crashes, they accumulate with no reaper.Consider clearing the snapshot on the early-return path as well, and tracking a periodic sweep of
dedup::run_nodes:*rows for runs that are no longer inflow_runs.♻️ Proposed change for the early-return path
if node_ids.is_empty() { tracing::trace!(target: "flows", %flow_id, %run_id, %status, "[dedup-commit] no dedup nodes in this flow — nothing to settle"); + // Nothing to settle, but a snapshot row may still exist (an + // unusable snapshot plus a dedup-free saved graph). Drop it so + // the row does not outlive the run it describes. + self.clear_run_snapshot(&namespace, flow_id, run_id); return; }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/flows/bus.rs` around lines 757 - 763, Update handle_finished so the node_ids.is_empty() early-return path also calls clear_run_snapshot for the current namespace and run_id before returning. Additionally, add periodic cleanup for dedup::run_nodes:* snapshot rows whose run IDs no longer exist in flow_runs, reusing the existing snapshot and flow-state cleanup mechanisms.
730-739: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a diagnostic for the malformed or empty snapshot branch.
The read-error branch logs a warning. The malformed branch (
as_array()returnsNone) and the empty-ids branch return silently. Both branches change the settlement source from the run snapshot to the saved graph. Without a log line, that switch is not observable in the field.♻️ Proposed diagnostics
- let ids: Vec<String> = value - .as_array()? - .iter() + let Some(array) = value.as_array() else { + tracing::warn!( + target: "flows", %namespace, %run_id, + "[dedup-commit] this run's dedup snapshot is not an array — falling back to the \ + flow's saved graph" + ); + return None; + }; + let ids: Vec<String> = array + .iter() .filter_map(Value::as_str) .map(str::to_string) .collect(); if ids.is_empty() { + tracing::warn!( + target: "flows", %namespace, %run_id, + "[dedup-commit] this run's dedup snapshot held no usable node ids — falling back \ + to the flow's saved graph" + ); return None; }As per coding guidelines: "New or changed flows must include verbose, grep-friendly diagnostics for entry/exit, branches, external calls, retries, state transitions, and errors".
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/flows/bus.rs` around lines 730 - 739, Add grep-friendly warning diagnostics to the snapshot parsing flow around the ids extraction: explicitly log when the snapshot value is not an array before returning None, and when the parsed ids collection is empty before returning None. Include enough context to identify the snapshot-to-saved-graph settlement fallback, while leaving valid non-empty snapshots unchanged.Source: Coding guidelines
src/openhuman/flows/bus_tests.rs (1)
1097-1108: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert snapshot cleanup on the failure path.
dedup_commit_clears_the_run_snapshot_after_settlingcovers onlystatus = "completed". No test proves the snapshot is cleared when settlement takes thereleasebranch.handle_finishedclears the snapshot after the loop for both branches, so add the assertion here to pin that behavior.💚 Proposed assertion
assert!( store::kv_get(&config, &namespace, "dedup:dd:committed") .unwrap() .is_none(), "a failed run must never commit anything" ); + assert!( + store::kv_get(&config, &namespace, &run_snapshot_key("run-1")) + .unwrap() + .is_none(), + "a failed run must also drop its own run-scoped snapshot" + ); }As per coding guidelines: "Cover at least 80% of changed lines with Vitest and Rust coverage before merge."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/openhuman/flows/bus_tests.rs` around lines 1097 - 1108, Add a failure-path assertion in dedup_commit_clears_the_run_snapshot_after_settling, after the release-branch settlement, verifying the run snapshot is cleared while retaining the existing tentative-key and committed-key checks.Source: Coding guidelines
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Nitpick comments:
In `@src/openhuman/flows/bus_tests.rs`:
- Around line 1097-1108: Add a failure-path assertion in
dedup_commit_clears_the_run_snapshot_after_settling, after the release-branch
settlement, verifying the run snapshot is cleared while retaining the existing
tentative-key and committed-key checks.
In `@src/openhuman/flows/bus.rs`:
- Around line 791-795: Rewrite the comment immediately above
self.clear_run_snapshot in the per-flow lock to remove the claim that deleting
the snapshot prevents duplicate settlement. State that late duplicate
FlowRunFinished events may fall back to the saved flow graph, while commit’s
set-union behavior and release’s idempotent deletion keep repeated settlement
safe.
- Around line 757-763: Update handle_finished so the node_ids.is_empty()
early-return path also calls clear_run_snapshot for the current namespace and
run_id before returning. Additionally, add periodic cleanup for
dedup::run_nodes:* snapshot rows whose run IDs no longer exist in flow_runs,
reusing the existing snapshot and flow-state cleanup mechanisms.
- Around line 730-739: Add grep-friendly warning diagnostics to the snapshot
parsing flow around the ids extraction: explicitly log when the snapshot value
is not an array before returning None, and when the parsed ids collection is
empty before returning None. Include enough context to identify the
snapshot-to-saved-graph settlement fallback, while leaving valid non-empty
snapshots unchanged.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: af6a3d77-e942-4142-aaf6-2e9079b20304
📒 Files selected for processing (2)
src/openhuman/flows/bus.rssrc/openhuman/flows/bus_tests.rs
|
| Filename | Overview |
|---|---|
| src/openhuman/flows/bus.rs | Adds started-event snapshotting, snapshot-based dedup settlement, fallback behavior, and cleanup. |
| src/openhuman/flows/bus_tests.rs | Moves existing bus tests into a sibling module and adds coverage for snapshot settlement, fallback, cleanup, and graph edits. |
Sequence Diagram
sequenceDiagram
participant Ops as Flow operations
participant Bus as Event bus
participant Dedup as Dedup subscriber
participant State as Flow state
Ops->>Bus: FlowRunStarted(flow_id, run_id)
Bus->>Dedup: Dispatch started event
Dedup->>State: Store dedup::run_nodes:run_id
Ops->>Bus: FlowRunFinished(flow_id, run_id, status)
Bus->>Dedup: Dispatch finished event
Dedup->>State: Load run-node snapshot
Dedup->>State: Commit or release node state
Dedup->>State: Delete run-node snapshot
Reviews (3): Last reviewed commit: "style: apply cargo fmt to the touched fl..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| let ids: Vec<String> = value | ||
| .as_array()? | ||
| .iter() | ||
| .filter_map(Value::as_str) | ||
| .map(str::to_string) | ||
| .collect(); |
There was a problem hiding this comment.
A mixed array silently drops non-string entries while treating the remaining IDs as a complete snapshot, so omitted nodes are not settled and the fallback intended for malformed snapshots is bypassed. Validate every element and reject the entire array when any entry is not a string.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: e2e3cf5da6
ℹ️ About Codex in GitHub
Codex has been enabled to automatically review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
When you sign up for Codex through ChatGPT, Codex can also answer questions or update the PR, like "@codex address that feedback".
| DomainEvent::FlowRunStarted { flow_id, run_id } => { | ||
| match store::get_flow(&self.config, flow_id) { | ||
| Ok(Some(flow)) => { | ||
| snapshot_run_dedup_nodes(&self.config, flow_id, run_id, &flow); |
There was a problem hiding this comment.
When a flow is edited immediately after a run starts, this FlowRunStarted handler can still snapshot the edited graph rather than the graph already loaded by prepare_flow_run: publish_global only enqueues the event and the subscriber runs later on its own task, so publication is not a synchronous barrier before edits or engine execution. In that timing window the deleted/renamed dedup node is still missed (or a newly-added one is captured), so the new settlement logic falls back into the exact mid-run edit corruption it is intended to fix; write the snapshot from the prepared Flow on the run-start path instead of reloading by flow_id here.
Useful? React with 👍 / 👎.
CI's `cargo fmt --all -- --check` gate rejected the previous revision. Three formatting-sensitive constructs are reworked, each matched against a formatter-clean precedent already in the tree rather than guessed at: * `dedup_node_ids_in` broke a method chain as `flow.graph` / `.nodes` / …, merging the chain parent with its first field access. The same expression elsewhere in this file is written with the parent alone on its own line, so the receiver is now bound to a `nodes` local and the chain starts from it — the shape `flows/ops.rs` already uses at the same indent level. * `clear_run_snapshot` had an `if let Err(e) = store::kv_delete(...)` line sitting exactly on the 100-column limit. The key is hoisted into a local, which shortens the line and matches how `snapshotted_dedup_node_ids` immediately above already builds the same key. * `handle` had been rewritten from the original `if let` into a `match` with a nested `match` inside a block arm, which changed formatting that was previously known-good for no behavioural benefit. The `FlowRunFinished` block is restored verbatim and the new `FlowRunStarted` handling is a short `if let` above it, delegating to a new `snapshot_run_nodes` method that holds the graph load. No behaviour change: the same snapshot is written at run start, read at settlement, and cleared afterwards. Pure formatting and extraction. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
|
Want your agent to iterate on Greptile's feedback? Try greploops. |
1 participant
Summary
DedupCommitSubscribernow settles a finished run'sdedupnodes against the graph that run executed, not against whatever the flow has since been edited into.Scoped deliberately to item 2 of #5268. Item 1 (the engine-side in-run read-modify-write on
tentative) is explicitly blocked on a CAS-capableStateStoreand is not touched here.Problem
DedupCommitSubscriber::handle_finishedderived the set ofdedupnode ids by loading the flow's current saved graph atFlowRunFinishedtime.StateStoreexposes no prefix-scan, so the graph was the only way to recover whichdedup::*keys exist — but the graph read is not the graph the run ran.If a flow is edited while a run is in flight:
dedupnode the run wrotetentativekeys under, then deleted or renamed before the run finished, is no longer in the graph. Its keys are therefore neither committed nor released — they sittentativeforever. Since the node only ever consultscommitted, those items are treated as unseen and silently reprocess on the flow's next run. That is a direct violation of the exactly-once contract intinyflows::nodes::control_flow::dedup's module docs.dedupnode added after the run started was settled by a run that never executed it — a no-op today, but it is settlement acting on a node outside the run's actual scope.Both are silent. Nothing logs, nothing fails, the run row reads
completed.Solution
Pin the run's dedup nodes at run start, settle against that pin, then drop it.
DedupCommitSubscribernow also handlesDomainEvent::FlowRunStarted. That variant already exists and is already published byflows::ops::flows_run{,_detached}immediately after theflow_runsrow insert — before the engine executes a single node — so the graph read back in the handler is the graph the run is starting with. It also already shares a match arm withFlowRunFinishedinDomainEvent::domain(), so it is tagged"cron"and the subscriber's existingdomains() -> Some(&["cron"])filter admits it unchanged.snapshot_run_dedup_nodeswrites those node ids into the existing per-flowflow_stateKV table: namespaceflow:<flow_id>, keydedup::run_nodes:<run_id>, value a JSON string array.dedup_node_idsprefers that snapshot and falls back to the saved graph.clear_run_snapshotdeletes the snapshot after settling, inside the existing per-flow commit lock.Design notes worth reviewing:
tinyflows::caps::build_capabilities'sstate_namespace). Note this is notflow_namespace(), which is the flow's memory namespace — a different string. The new helper is namedflow_state_namespaceto make that distinction unmissable.dedup::double-colon prefix cannot collide with the engine'sdedup:<node_id>:*keys: reaching that shape would require a node id beginning with:, which the graph schema does not admit. There is a test asserting exactly this againstdedup_node::tentative_key/committed_key.dedupnode writes no snapshot at all, so the overwhelming majority of runs cost one graph scan and nothing else — and an absent key stays an unambiguous "no snapshot".flows_resumere-enters an existing run without a fresh run-start), and a failed snapshot write. An unreadable, non-array, or empty snapshot is reported as absent rather than as "settle nothing" — settling nothing would strand tentative keys, which is the very bug being fixed.tracing::warn!-ed and swallowed. By the time these events are observed the run has already settled its ownflow_runsrow, so nothing here may retroactively affect run status.Test module extraction — please flag if unwanted
bus.rsgrew past a comfortable size, so this PR also moves its#[cfg(test)] mod testsinto a siblingsrc/openhuman/flows/bus_tests.rsvia:This is the crate's own established convention — the pattern appears at ~170 sites, including plain (non-
mod.rs) source files such assubconscious/factory.rs → factory_tests.rsandskills/ops.rs → ops_tests.rs. The pre-existing test bodies are unchanged apart from one level of dedent (verified by a round-trip check: re-joining the two files reproduces the combined file byte-for-byte). If you would rather keep a bug-fix diff free of this, say so and I will fold the tests back intobus.rs— it is a mechanical revert.Impact
dedupnode: unchanged, and no new I/O.dedupnodes that are never edited mid-run: unchanged behaviour, one extra small KV write + delete per run.Related
Closes #5268 (item 2 only — item 1 remains open and engine-blocked).
Context: #5263 (the two-sided dedup contract), #5265 (the per-flow commit lock this builds on).
Submission Checklist
.envcontent touchedcargo fmt --all -- --check— cannot be run in the authoring environment (no Rust toolchain); CI is the authority. See Validation Blocked.cargo clippy -p openhuman -- -D warnings— cannot be run in the authoring environment (no Rust toolchain); CI is the authority. See Validation Blocked.cargo test -p openhuman flows::bus— cannot be run in the authoring environment (no Rust toolchain); CI is the authority. See Validation Blocked.AI Authored PR Metadata
Validation Run: None. No command was executed against this change.
Validation Blocked: This PR was authored in an environment with no Rust toolchain, no
git, and no outbound network.cargo fmt,cargo clippy, andcargo testcould not be run, and I will not claim otherwise. CI is the authority on this diff.What was done instead, mechanically:
-D warningsclippy gate, since a single unused item fails the build:snapshot_run_dedup_nodes1 def / 11 refs,run_dedup_snapshot_key1 / 10,dedup_node_ids_in1 / 4,flow_state_namespace1 / 4,snapshotted_dedup_node_ids1 / 1,clear_run_snapshot1 / 1. No dead code; all new free functions are private.Update — the first CI run caught exactly what you would expect it to.
Rust Qualityfailed oncargo fmt --all -- --check. My line-width checking was necessary but not sufficient: rustfmt also has opinions about where a construct breaks, which a width check cannot see. Since I cannot read the CI log from here, I re-derived the formatter's behaviour empirically from code already in the tree (which is formatter-clean by definition) and corrected three constructs against that evidence:dedup_node_ids_inbroke a chain asflow.graph/.nodes/ …, merging the chain parent with its first field access. The same expression elsewhere in this file is written with the parent alone on its own line. Fixed by binding the receiver to a local and starting the chain there — the shapeflows/ops.rsalready uses at the same indent.clear_run_snapshothad a line sitting exactly on the 100-column limit. Key hoisted into a local.handlehad been rewritten from the originalif letinto amatchwith a nestedmatchin a block arm — changing formatting that was already known-good, for no behavioural benefit. TheFlowRunFinishedblock is now restored verbatim, and the newFlowRunStartedhandling is a shortif letabove it delegating to asnapshot_run_nodesmethod.That last change is the one worth a reviewer's eye: it is a pure restructure with no behavioural difference, made specifically to stop guessing at formatter behaviour and instead reuse a construct the formatter has already accepted in this exact file.
Concretely: the logic has been reasoned about carefully and the file is structurally sound, but it has never been compiled. Please continue to treat a CI failure here as expected-and-my-fault rather than surprising, and I will turn fixes around promptly.
Behavior Changes: Yes — described under Impact. Settlement now consults a per-run snapshot before the saved graph.
Parity Contract: N/A — no cross-platform or frontend/backend parity surface is touched.
Duplicate / Superseded PR Handling: I checked the open PR list before starting; #5268 is unassigned with no open PR against it. If one has landed since, close this in favour of it.
Coverage note:
docs/TEST-COVERAGE-MATRIX.mdis unchanged — this is a behaviour fix within an already-covered module, not a new surface.Tests added (15)
Snapshot-based settlement:
..._settles_a_node_deleted_from_the_flow_mid_run— the headline bug..._settles_renamed_node_end_to_end_through_events— fullFlowRunStarted→FlowRunFinishedpath..._prefers_the_snapshot_over_a_node_added_after_the_run_started..._releases_..._on_failure— the non-success path still releases..._clears_the_run_snapshot_after_settlingFallback correctness:
..._falls_back_..._no_snapshot..._falls_back_..._malformed_snapshotHelpers and invariants:
flow_state_namespace_matches_the_engine_state_namespace— pins the namespace tobuild_capabilities, so a future rename there cannot silently desync settlementrun_dedup_snapshot_key_cannot_collide_with_a_dedup_nodes_own_keysdedup_node_ids_in_returns_every_dedup_node_in_graph_orderPlus three tests covering the new
FlowRunStartedarm (no dedup node ⇒ no write; flow missing ⇒ no panic; load error ⇒ no panic) and two regression tests around the existing commit/release paths.Happy to split the test-module extraction into its own PR, revert it entirely, or rework the storage location if
flow_stateis the wrong home for this. Feedback welcome on all of it.Summary by CodeRabbit
Bug Fixes
Tests