feat(node): bind mesh + workflows, locks, periodic, fan-out#270
Merged
Conversation
expandFanOut/checkFanOutCompletion/createDeferredJob/finalizeRunIfTerminal plus getWorkflowRunPlan/workflowNodeForJob/cascadeSkipPending; deferred-node submit; skipCascade on markWorkflowNodeResult; JsWorkflowNode gains fanOutCount + compensation fields.
WorkflowTracker reacts to the worker outcome callback, reconstructs the run plan from storage, expands fan-outs, aggregates fan-ins, and finalizes managed runs. Builder gains .fanOut()/.fanIn(); deferred-node args persisted via step_metadata.args_template.
Surface fanOutCount + compensation fields in the node contract; enrich /dag with per-node deps, live status, and job-id links so the SPA visualizer renders edges, colours, and links.
e2e fan-out/fan-in, child-failure fail-fast, empty fan-out, fan-out without collector; assert the enriched DAG shape in the dashboard workflow test.
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: defaults Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
✅ Files skipped from review due to trivial changes (1)
📝 WalkthroughWalkthroughThis PR adds five major Node.js SDK features via new Rust N-API bindings and TypeScript modules: workflow DAG orchestration (builder, manager, fan-out/fan-in tracker), distributed advisory locks with auto-renewal, cron-based periodic task registration, a mesh work-stealing overlay, and circuit-breaker wiring. It also adds a leveled logger, dashboard workflow API endpoints, Node SDK documentation pages, and a ChangesNode SDK Feature Parity
Sequence Diagram(s)sequenceDiagram
participant App as Application
participant WM as WorkflowManager
participant NQ as JsQueue (N-API / Rust)
participant WF as WorkflowStorage
participant Core as CoreStorage
participant W as Worker + WorkflowTracker
App->>WM: define("pipeline").step("a").fanOut("expand").fanIn("collect").build()
App->>WM: submit(spec)
WM->>NQ: submitWorkflow(dag, stepMeta, args)
NQ->>Core: enqueue_job("a", depends_on=[])
NQ->>WF: insert_node("a", job_id)
NQ->>WF: insert_node("expand", deferred=true)
NQ->>WF: insert_node("collect", deferred=true)
NQ->>WF: transition_run(Running)
NQ-->>WM: run_id
WM-->>App: WorkflowHandle
W->>Core: poll + execute job "a"
W->>W: onOutcome(outcome for "a")
W->>NQ: markWorkflowNodeResult("a", ok, result=[1,2,3])
W->>NQ: expandFanOut("expand", children=[1,2,3])
NQ->>Core: enqueue_job per child
NQ->>WF: insert child nodes
NQ-->>W: childJobIds
W->>Core: execute child jobs (parallel)
W->>NQ: checkFanOutCompletion("expand")
NQ-->>W: JsFanOutCompletion { ok: true, childJobIds }
W->>NQ: createDeferredJob("collect", args=[allResults])
NQ->>Core: enqueue_job("collect")
W->>Core: execute "collect"
W->>NQ: finalizeRunIfTerminal(run_id)
NQ->>WF: transition_run(Completed)
App->>WM: handle.wait()
WM-->>App: WorkflowRun { state: "completed" }
Estimated code review effort🎯 5 (Critical) | ⏱️ ~120 minutes Possibly related PRs
Suggested labels
Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 20
🧹 Nitpick comments (1)
sdks/node/src/workflows/tracker.ts (1)
221-266: ⚡ Quick winCache node statuses once per successor-evaluation pass.
allPredecessorsTerminal()currently rebuilds status state for each successor, causing repeated native fetches in the hot path.Proposed refactor
private evaluateSuccessors(runId: string, nodeName: string, plan: RunPlan): void { + const status = this.nodeStatuses(runId); for (const succ of plan.successors.get(nodeName) ?? []) { - if (!plan.deferred.has(succ) || !this.allPredecessorsTerminal(runId, succ, plan)) { + if (!plan.deferred.has(succ) || !this.allPredecessorsTerminal(succ, plan, status)) { continue; } @@ - private allPredecessorsTerminal(runId: string, node: string, plan: RunPlan): boolean { + private allPredecessorsTerminal( + node: string, + plan: RunPlan, + status: Map<string, string>, + ): boolean { const preds = plan.predecessors.get(node) ?? []; if (preds.length === 0) { return true; } - const status = this.nodeStatuses(runId); return preds.every((p) => TERMINAL_NODE_STATUS.has(status.get(p) ?? "")); }Also applies to: 299-301
🤖 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 `@sdks/node/src/workflows/tracker.ts` around lines 221 - 266, The evaluateSuccessors() method calls allPredecessorsTerminal() for each successor in the loop, which internally calls this.nodeStatuses(runId) repeatedly, causing unnecessary native fetches in the hot path. Cache the node statuses once at the beginning of evaluateSuccessors() by calling this.nodeStatuses(runId), then refactor allPredecessorsTerminal() to accept this cached status map as a parameter instead of fetching it internally, and update the call to allPredecessorsTerminal() to pass the cached status.
🤖 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 `@crates/taskito-node/src/config.rs`:
- Around line 95-98: The MeshWorkerConfig.port field is defined as u32 but is
converted to u16 downstream and port + 1 is computed for the steal traffic port.
Values above 65534 can wrap when converted to u16, causing unexpected port
bindings and breaking connectivity. Add validation in the MeshWorkerConfig to
constrain the port value to the range 0-65534, rejecting any values that would
cause wrapping when computing port + 1 within a u16 range.
In `@crates/taskito-node/src/convert/task_config.rs`:
- Around line 39-57: The circuit_breaker_config function accepts unchecked
numeric input from CircuitBreakerInput without validating that values are valid
(non-positive and non-finite values are not rejected). Add validation logic at
the beginning of the circuit_breaker_config function to check that all numeric
fields (threshold, window_ms, cooldown_ms, half_open_max_probes, and
half_open_success_rate) have valid bounds before constructing and returning the
CircuitBreakerConfig struct. Invalid numeric values should cause the function to
return an error or panic with a descriptive message to prevent invalid breaker
configurations from reaching runtime logic.
In `@crates/taskito-node/src/queue/locks.rs`:
- Around line 17-37: The acquire_lock and extend_lock methods accept ttl_ms
parameter without validating that it is positive, which can result in
non-functional locks. Add validation checks at the beginning of both the
acquire_lock method (starting at line 17) and the extend_lock method (starting
at line 33) to ensure ttl_ms is greater than zero, and return a descriptive
error if the validation fails, before passing the value to the storage layer.
In `@crates/taskito-node/src/queue/workflows.rs`:
- Around line 72-90: The issue is in the Some(existing) branch of the match
statement where existing.id is reused without validating that the stored
definition's dag_data and step_metadata match the incoming dag and step_meta
values. Add a validation check in the Some(existing) branch to compare the
existing definition's content with the new dag and step_meta. If they don't
match, create a new WorkflowDefinition with a new UUID and insert it instead of
reusing existing.id. Only reuse the existing.id when the definition content
matches exactly, preventing definition drift where a run could execute a
different DAG than what its definition_id references.
- Around line 466-485: The enqueue operation happens before
set_workflow_node_job, creating a data consistency issue where if the binding
fails, an orphaned job remains in the queue. After calling self.storage.enqueue
and getting the job result, store it in a variable. Then, when calling
wf.set_workflow_node_job, if it fails, first remove or dequeue the job that was
just created (you may need to add a rollback/delete method to remove the job
from storage) before returning the error. Only return Ok(job.id) after both
enqueue and set_workflow_node_job succeed without errors.
- Around line 415-443: The child jobs are enqueued and persisted to storage in
the loop before the workflow nodes are batch-inserted in the
create_workflow_nodes_batch call. If the batch insert fails, the function
returns an error but the child jobs remain persisted in storage as orphaned jobs
with no workflow node tracking. To fix this, implement a rollback mechanism by
catching the error from create_workflow_nodes_batch and deleting the previously
enqueued child jobs using their collected child_job_ids before returning the
error. This ensures that if the workflow node creation fails, no orphaned jobs
are left in the system.
In `@crates/taskito-node/src/worker.rs`:
- Around line 276-279: The build_mesh_config function silently truncates
input.port (a u32 value) when casting to u16, causing ports above 65535 to wrap
around unexpectedly. Change the function signature to return a Result type
instead of MeshConfig directly, and add validation before the saturating_add and
cast operations to ensure input.port is within the valid u16 range (0-65535). If
the port value is invalid, return an error rather than allowing the truncation
to occur.
- Around line 117-123: The gossip and steal-server tasks spawned in the second
spawn block are not being explicitly shut down when the worker terminates.
Instead of relying on the bridge to implicitly close these tasks, call
mesh_node.request_shutdown() before or during worker shutdown to explicitly
signal the tasks to stop. This ensures proper resource cleanup and prevents the
tasks from waiting indefinitely on the shutdown notification. Add the shutdown
call in the appropriate cleanup path or ensure the Arc<MeshNode> reference is
properly dropped to trigger task termination.
In `@docs/content/docs/node/workflows/fan-out.mdx`:
- Around line 1-7: The file is missing an import statement for the Mermaid
component that is used on lines 11 and 74. Currently, the import statement only
includes Callout from fumadocs-ui/components/callout. Update the import
statement to also include Mermaid component alongside Callout from the
appropriate import path (determine the correct source path for Mermaid similar
to how Callout is imported, or add a separate import statement for Mermaid if it
comes from a different source).
In `@NODE_SDK_TODO.md`:
- Line 250: Line 250 in NODE_SDK_TODO.md contains a fenced code block opening
(triple backticks) without a language identifier, which violates markdownlint
rule MD040. Add the language token "text" to the opening fence by changing ```
to ```text to specify the block contains plain text content and resolve the
linting warning.
- Line 7: Update the NODE_SDK_TODO.md file to remove stale information and
reconcile conflicting data. First, update line 7 to reflect the current status
that this work is now in a PR rather than claiming "no PRs yet". Second, resolve
the conflicting test count claims where line 42 states 48 tests but line 265
states 52 tests—either verify the correct count and update both references to
match, or remove the hardcoded numbers entirely and reference a single source of
truth for test metrics. This will ensure the tracker provides reliable
verification guidance.
In `@sdks/node/README.md`:
- Around line 162-166: The lock example at line 162 uses the `using` syntax
which is not compatible with Node.js 18.x (the minimum version targeted by this
README). Replace this example with a `try/finally` block pattern that works
across all supported Node versions, where you call lock.acquire() in the try
block, perform the critical section, and ensure lock.release() is called in the
finally block. Alternatively, you can keep the `using` syntax example but add a
clear note documenting that this syntax requires Node.js 20.4.0 or later and
provide a separate try/finally example for earlier versions.
In `@sdks/node/src/locks/lock.ts`:
- Around line 37-39: The ttlMs value in the Lock class constructor is assigned
without validation, which can allow invalid values like non-positive numbers or
NaN that degrade lock renewal semantics. Add validation logic in the constructor
after the ttlMs assignment (in the line `this.ttlMs = options.ttlMs ??
DEFAULT_TTL_MS;`) to check that ttlMs is a positive number and throw an
appropriate error if it is invalid. This validation should reject zero, negative
values, or NaN to ensure proper lock and timer scheduling behavior.
- Around line 49-51: The early return in the `acquire()` method trusts the local
`this.held` state without verifying that the remote lock is still valid and
hasn't expired. Add a check for lease expiration before returning true in the
condition where `this.held` is true. Verify that the lock lease has not expired
on the remote side, and only return true if both the local state indicates the
lock is held and the actual lease is still valid. This ensures the method
doesn't report success when the remote lock has already expired due to
`autoExtend: false` or delayed renewals.
In `@sdks/node/src/queue.ts`:
- Around line 92-101: The withLock method ignores the boolean return value from
lock.release() in the finally block, which indicates whether the lock was
successfully held until release. If the lock is lost during execution (e.g., due
to renewal failure), this should be detected and raise an error rather than
allowing the method to resolve successfully. Capture the return value from
lock.release() and throw an appropriate error if it returns false, ensuring the
"run while holding lock" contract is enforced.
In `@sdks/node/src/utils/logger.ts`:
- Around line 36-53: The levelFromEnv() function uses the in operator to check
if the environment variable value exists in SEVERITY, which incorrectly matches
inherited properties (e.g., "toString" would pass). Additionally, setLogLevel()
and setLogSink() accept their parameters without runtime validation, allowing
invalid values to be assigned to the config object, which breaks threshold
comparisons or causes logging to fail. Fix levelFromEnv() by using
Object.prototype.hasOwnProperty.call(SEVERITY, raw) instead of the in operator
to only match own properties. Then add runtime validation in setLogLevel() to
ensure the level parameter is a valid LogLevel value from SEVERITY, and in
setLogSink() to ensure the sink parameter is a valid function before assigning
to config.
In `@sdks/node/src/webhooks/deliverer.ts`:
- Around line 71-76: The log.warn statement in the webhook delivery failure
handling logs the raw webhook.url which may contain sensitive credentials or
tokens in query parameters or userinfo sections. Before logging, mask or
sanitize the URL to remove any sensitive information, then use the sanitized
version in the log message instead of the raw webhook.url value. This prevents
credential leaks into centralized logs while still providing enough URL
information for debugging purposes.
In `@sdks/node/src/workflows/builder.ts`:
- Around line 89-95: The build() method currently validates that all edge
predecessors are known steps, but it does not validate that when a fanIn node
specifies an after option, that target must be a fan-out node. If fanIn.after
points to a non-fan-out step, the tracker will never enqueue the fan-in node
since it only creates fan-ins from fan-out completion, causing potential hangs.
Add validation in the build() method alongside the existing unknown step check
to ensure that each fanIn node with an after property targets an actual fan-out
step, and throw an appropriate error if this constraint is violated.
In `@sdks/node/src/workflows/manager.ts`:
- Around line 135-137: Add validation guards for timeoutMs and pollMs before
they are used in the wait loop. Check that both timeoutMs and pollMs are valid
finite numbers (not NaN, not Infinity, and preferably positive values) to
prevent timeout behavior from being disabled or runtime degradation. This
validation should occur immediately after the assignment of these variables and
before the deadline calculation and wait loop logic, and should also be applied
to the similar code section referenced at lines 146-149.
In `@sdks/node/test/dashboard.test.ts`:
- Around line 163-164: Replace the hardcoded 80ms setTimeout delay with a proper
wait for the server to start listening. Instead of using `await new
Promise((resolve) => setTimeout(resolve, 80))`, create a promise that resolves
when the server emits a 'listening' event. This ensures the srv object is
actually ready and `srv.address()` will be available before attempting to read
the port, eliminating the race condition that causes flakiness in CI
environments.
---
Nitpick comments:
In `@sdks/node/src/workflows/tracker.ts`:
- Around line 221-266: The evaluateSuccessors() method calls
allPredecessorsTerminal() for each successor in the loop, which internally calls
this.nodeStatuses(runId) repeatedly, causing unnecessary native fetches in the
hot path. Cache the node statuses once at the beginning of evaluateSuccessors()
by calling this.nodeStatuses(runId), then refactor allPredecessorsTerminal() to
accept this cached status map as a parameter instead of fetching it internally,
and update the call to allPredecessorsTerminal() to pass the cached status.
🪄 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: defaults
Review profile: CHILL
Plan: Pro
Run ID: 54aa5e9a-6d00-43e1-9a13-be17d0275a4c
📒 Files selected for processing (49)
NODE_SDK_TODO.mdcrates/taskito-node/Cargo.tomlcrates/taskito-node/src/config.rscrates/taskito-node/src/convert/lock.rscrates/taskito-node/src/convert/mod.rscrates/taskito-node/src/convert/task_config.rscrates/taskito-node/src/convert/workflow.rscrates/taskito-node/src/queue/locks.rscrates/taskito-node/src/queue/mod.rscrates/taskito-node/src/queue/periodic.rscrates/taskito-node/src/queue/workflows.rscrates/taskito-node/src/worker.rsdocs/content/docs/meta.jsondocs/content/docs/node/index.mdxdocs/content/docs/node/meta.jsondocs/content/docs/node/workflows/fan-out.mdxdocs/content/docs/node/workflows/index.mdxdocs/content/docs/node/workflows/meta.jsonsdks/node/README.mdsdks/node/package.jsonsdks/node/src/dashboard/contract.tssdks/node/src/dashboard/handlers.tssdks/node/src/dashboard/routes.tssdks/node/src/dashboard/server.tssdks/node/src/errors.tssdks/node/src/index.tssdks/node/src/locks/index.tssdks/node/src/locks/lock.tssdks/node/src/locks/types.tssdks/node/src/native.tssdks/node/src/queue.tssdks/node/src/types.tssdks/node/src/utils/index.tssdks/node/src/utils/logger.tssdks/node/src/webhooks/deliverer.tssdks/node/src/worker.tssdks/node/src/workflows/builder.tssdks/node/src/workflows/index.tssdks/node/src/workflows/manager.tssdks/node/src/workflows/plan.tssdks/node/src/workflows/tracker.tssdks/node/src/workflows/types.tssdks/node/test/dashboard.test.tssdks/node/test/locks.test.tssdks/node/test/logger.test.tssdks/node/test/mesh.test.tssdks/node/test/periodic.test.tssdks/node/test/workflows-fanout.test.tssdks/node/test/workflows.test.ts
Collaborator
Author
|
Addressed all 20 CodeRabbit findings (61 vitest green; cargo+clippy+tsc+biome clean). Mesh — validate port CI note — the prior red on this PR was |
No Mermaid component exists in the fumadocs setup, so the import broke the docs build. Plain mermaid code fences need no import.
This was referenced Jun 21, 2026
This was referenced Jul 5, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
taskito-mesh(work-stealing overlay) andtaskito-workflows(DAG workflows) behind cargo features (mesh,workflows; enabled bybuild:native).withLock/usinghelpers.utils/.Verification
cargo check -p taskito-node --features postgres,redis,mesh,workflows✓napi build --release(postgres,redis,mesh,workflows) ✓ ·tsc✓ · biome ✓ · tsup ✓vitest run— 59 tests pass ✓Notes
server.ts(kept feat(node): dashboard server, events/middleware, webhooks #269 hardening + adopted this branch's leveled logger), README wording, andconvert/task_config.rs(kept rate-limit fail-fast + added circuit-breaker config).Summary by CodeRabbit
Release Notes
New Features
Documentation
Tests