feat(nodes): add dedup node kind (exactly-once, commit-on-success filter half) - #25
Conversation
Adds the 14th node kind, `dedup`: a control-flow filter that drops an
item whose per-item `key` expression is already in the host-managed
`StateStore` committed set, and otherwise passes it through while
staging the key in a tentative set for the host to commit/release
based on run outcome (PR 2, in the OpenHuman host crate).
- model/node_kind.rs: NodeKind::Dedup, wire value "dedup"
- catalog.rs: NODE_KINDS (13->14) + node_contracts() entry
- validate.rs: require non-empty `key` on dedup nodes
- nodes/control_flow/dedup.rs: the executor, StateStore key layout
(dedup:<node_id>:{committed,tentative}), and null/duplicate-key
handling, with unit tests
- nodes/control_flow/mod.rs, nodes/mod.rs: wire the new module and
executor_for dispatch arm
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughAdds a ChangesDedup node
Estimated code review effort: 4 (Complex) | ~45 minutes Possibly related issues
Poem
Sequence Diagram(s)sequenceDiagram
participant InputBatch
participant DedupNode
participant StateStore
participant NodeOutput
InputBatch->>DedupNode: items and config.key expression
DedupNode->>StateStore: load committed keys
DedupNode->>DedupNode: resolve keys and filter duplicates
DedupNode->>StateStore: persist tentative keys once per batch
DedupNode->>NodeOutput: emit passed items and diagnostics
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
Comment |
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
|
| Filename | Overview |
|---|---|
| src/nodes/control_flow/dedup.rs | New DedupNode executor: commit-on-success exactly-once filter with clear StateStore contract, fail-open null key handling, within-run duplicate detection, and 9 test cases covering all spec behaviors |
| src/validate.rs | Adds static config validation for dedup nodes requiring a non-empty string key field; correctly distinguishes static-config missing-key (authoring error) from run-time null-resolve (fail-open) |
| src/catalog.rs | Adds dedup contract with required key field, fail-open documentation, and linear port spec; NODE_KINDS bumped 13 to 14 with corresponding test updates |
| src/model/node_kind.rs | Adds Dedup variant to NodeKind enum with doc comment pointing to module; serde wire name dedup tested |
| src/nodes/mod.rs | Adds NodeKind::Dedup dispatch to executor; all_kinds() helper updated for exhaustive coverage |
| src/nodes/control_flow/mod.rs | Exposes dedup submodule and re-exports DedupNode; module doc updated to note the StateStore exception to the pure-control-flow rule |
Reviews (2): Last reviewed commit: "docs(dedup): document concurrency contra..." | Re-trigger Greptile
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 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.
Inline comments:
In `@src/nodes/control_flow/dedup.rs`:
- Around line 131-157: Prevent lost updates in the load_key_set/store_key_set
flow used for tentative_key(node_id) by serializing per-key read-modify-write
operations or introducing a StateStore compare-and-swap/atomic update primitive.
Ensure overlapping workflow runs union disjoint keys without last-writer-wins
overwrites, while preserving the existing sorted JSON-array representation.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 9330f813-aea3-46f7-9725-f566f10ab256
📒 Files selected for processing (6)
src/catalog.rssrc/model/node_kind.rssrc/nodes/control_flow/dedup.rssrc/nodes/control_flow/mod.rssrc/nodes/mod.rssrc/validate.rs
| async fn load_key_set(ctx: &NodeContext<'_>, key: &str) -> Result<HashSet<String>> { | ||
| let stored = ctx.caps.state.load(key).await?; | ||
| let set: HashSet<String> = stored | ||
| .as_ref() | ||
| .and_then(Value::as_array) | ||
| .map(|arr| { | ||
| arr.iter() | ||
| .filter_map(Value::as_str) | ||
| .map(str::to_string) | ||
| .collect() | ||
| }) | ||
| .unwrap_or_default(); | ||
| tracing::debug!(key, set_len = set.len(), "{LOG_PREFIX} loaded key set"); | ||
| Ok(set) | ||
| } | ||
|
|
||
| /// Persists `set` under `key` as a JSON array of strings, sorted for a stable, | ||
| /// diffable on-disk representation (membership is exact-match either way, so | ||
| /// sort order carries no semantic meaning — it only makes manual inspection | ||
| /// and test assertions predictable). | ||
| async fn store_key_set(ctx: &NodeContext<'_>, key: &str, set: &HashSet<String>) -> Result<()> { | ||
| let mut keys: Vec<String> = set.iter().cloned().collect(); | ||
| keys.sort_unstable(); | ||
| let value = Value::Array(keys.into_iter().map(Value::String).collect()); | ||
| tracing::debug!(key, set_len = set.len(), "{LOG_PREFIX} storing key set"); | ||
| ctx.caps.state.store(key, value).await | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Locate the StateStore trait definition to check for atomic update/CAS primitives
# and any documentation about concurrent-execution guarantees for the same node/flow.
rg -n -B2 -A20 'trait StateStore' --type=rs
rg -n -B2 -A5 'fn (load|store|update|compare_and_swap|cas)\(' --type=rs -g '*caps*'
rg -n -i 'concurren|serial(ize|ized)|same (node|flow) instance' --type=rs -g '*engine*' -g '*caps*'Repository: tinyhumansai/tinyflows
Length of output: 253
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Map the repository and find the relevant files.
git ls-files | rg '^(src/|.*caps.*|.*engine.*|.*state.*|.*workflow.*)' | head -n 200
printf '\n--- StateStore candidates ---\n'
rg -n 'trait StateStore|struct .*StateStore|enum .*StateStore|fn (load|store|update|compare_and_swap|cas)\(' src
printf '\n--- concurrency keywords ---\n'
rg -n -i 'concurren|serial(ize|ized)|same (node|flow) instance|overlap|race|atomic' srcRepository: tinyhumansai/tinyflows
Length of output: 16116
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Find the relevant source files first.
fd -a 'dedup.rs|caps|state' src
# Narrow in on the state-store trait and implementation surface.
rg -n 'trait StateStore|fn (load|store|update|compare_and_swap|cas)\(' src
# Search for any documented execution/concurrency guarantees.
rg -n -i 'concurren|serial(ize|ized)|same (node|flow) instance|overlap|race|atomic' srcRepository: tinyhumansai/tinyflows
Length of output: 14917
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf 'PWD: '; pwd
printf '\nTop-level files:\n'
git ls-files | head -n 200Repository: tinyhumansai/tinyflows
Length of output: 3832
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Inspect the files most likely to define state storage and execution semantics.
fd -a 'dedup.rs|caps.rs|state.rs|engine.rs|mod.rs' src
printf '\n--- trait/impl surface ---\n'
rg -n 'trait StateStore|struct .*StateStore|impl .*StateStore|fn (load|store|update|compare_and_swap|cas)\(' src
printf '\n--- execution/concurrency docs ---\n'
rg -n -i 'concurren|serial(ize|ized)|same (node|flow) instance|overlap|race|atomic|lock' srcRepository: tinyhumansai/tinyflows
Length of output: 21819
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- src/nodes/control_flow/dedup.rs ---'
sed -n '1,280p' src/nodes/control_flow/dedup.rs
printf '\n%s\n' '--- src/caps/mod.rs (StateStore section) ---'
sed -n '180,240p' src/caps/mod.rs
printf '\n%s\n' '--- src/engine.rs concurrency notes ---'
sed -n '1,220p' src/engine.rs
printf '\n%s\n' '--- src/engine.rs relevant concurrency excerpts ---'
sed -n '620,760p' src/engine.rs
printf '\n%s\n' '--- src/engine.rs branch race notes ---'
sed -n '1230,1270p' src/engine.rsRepository: tinyhumansai/tinyflows
Length of output: 35387
Serialize the tentative merge or add CAS support The load_key_set → union → store_key_set path on tentative_key(node_id) can lose disjoint keys when two runs of the same workflow overlap; StateStore only exposes load/store, so this is a last-writer-wins race. Consider a host-side per-key serialization guarantee or an atomic update primitive.
🤖 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/nodes/control_flow/dedup.rs` around lines 131 - 157, Prevent lost updates
in the load_key_set/store_key_set flow used for tentative_key(node_id) by
serializing per-key read-modify-write operations or introducing a StateStore
compare-and-swap/atomic update primitive. Ensure overlapping workflow runs union
disjoint keys without last-writer-wins overwrites, while preserving the existing
sorted JSON-array representation.
…key canonicalization Review fixes (tinyflows#25): - CodeRabbit (Major): document that the load→merge→store on tentative/committed is not atomic, so overlapping runs of the SAME flow can lose disjoint keys — bounded + safe-direction (under-dedup / occasional duplicate, never a wrong drop). Exactly-once holds under one-active-run-per-flow; full CAS deferred. - Greptile (P2): note non-scalar keys canonicalize via JSON text deterministically but a scalar id is recommended.
Review comments addressed — pushed in
|
|
Tip For best results, initiate chat on the files or code changes. 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
echo '--- commit summary ---'
git show --stat --oneline --decorate 0ce4fbc
echo
echo '--- changed dedup module documentation/context ---'
git show --format=fuller --find-renames 0ce4fbc -- src/nodes/control_flow/dedup.rs
echo
echo '--- current dedup module outline ---'
ast-grep outline src/nodes/control_flow/dedup.rs --items all
echo
echo '--- linked host PR metadata and changed files (if public/available) ---'
gh pr view 5265 --repo tinyhumansai/openhuman --json number,title,state,url,body,files 2>&1 || trueLength of output: 8765
One accuracy issue remains in the concurrency documentation: the residual tentative race can cause the same key to pass and be processed by two overlapping runs themselves, not solely “again on a later run.” The outcome is still safe-direction under-dedup (duplicate processing, never an incorrect drop), but the wording should cover both concurrent and subsequent duplicate processing. The original finding therefore remains open pending that small clarification. 🐇🔍 ✏️ Learnings added
|
|
Both addressed cleanly. Concurrency (major): The module doc is now the right place for this — the failure-mode analysis (lost tentative → re-process, never wrong-drop) is precise, and documenting the Structured keys (P2): Agreed that deterministic-but-smelly is better than a surprising hard error here. The "Key canonicalization" doc section is sufficient — authors who end up here while debugging will find the explanation. Looks good to me. |
1 participant
What
Adds the 14th node kind,
dedup— a first-class exactly-once filter. Config{ "key": "=item.id" }. Passes through items whose key hasn't been committed in a prior run and stages them tentative; the host (OpenHuman, tinyhumansai/openhuman#5263) commits tentative→committed on run success / releases on failure (commit-on-success). Motivated by a live test proving dedup-via-memory-node can't work (semantic recall ≠ exact keyed membership).StateStore contract (host commit half depends on this)
Per dedup node id:
dedup:<id>:committed(host writes on success; node reads) anddedup:<id>:tentative(node writes on pass; host reads on completion). Each is a JSON array set (exact-match). The node NEVER mutatescommitted.Behavior (test-pinned)
Fail-open on null/empty key (item passes, unrecorded — never silently drop); non-string keys canonicalized; same key twice in a run passes once; two dedup nodes keep independent sets; stale tentative preserved across runs. Single
mainoutput (engine emits one port perexecute).NODE_KINDS13→14, additive —CURRENT_SCHEMA_VERSIONunchanged.Tests
Node executor (9), validate (4), catalog (2).
cargo test --lib418 pass; clippy + fmt clean.Summary by CodeRabbit
dedupcontrol-flow node that performs exactly-once filtering using durable persisted state.keyexpressions, suppresses within-batch duplicates, and skips items when the resolved key isnull, missing, or empty (passes through without recording).dedupand its state-commit behavior.dedup.