feat: close the executor side-channel gaps (#590, #591, #593) - #604
Merged
Conversation
A frame type only the far side knows was a parse error, and a reader loop can only drop the connection over one — abandoning every in-flight job to the reaper. Headers declare their own payload length, so an unrecognised type is now skipped and logged instead. A known type that will not parse stays fatal.
The counter existed but only surfaced in a warning at session end, so nothing asserted that a flood sheds rather than blocks or grows. The test gates the writer instead of racing the drain, which makes both halves of the contract deterministic: progress coalesces, the oldest logs are shed.
The detached stand-in warned and dropped progress and task logs, so a task reaching the queue directly reported nothing while the same call through the job context worked. Both now reach the scheduler.
publish was the only route to a task log, so a plain log line had nowhere to go — on an executor or in a worker. Extra blobs now share one encoder that falls back to the value's string form rather than failing the task on a circular reference.
📝 WalkthroughWalkthroughThe PR adds length-delimited unknown-frame skipping to the worker protocol. Rust readers preserve sessions across version skew. Node executors route progress and task logs through scheduler sinks, with safe metadata encoding and backpressure tests. ChangesExecutor side-channel reporting
Estimated code review effort: 4 (Complex) | ~45 minutes Possibly related issues
Possibly related PRs
Suggested labels: Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches📝 Generate docstrings
Comment |
![coderabbitai[bot]](https://avatars.github.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
crates/taskito-core/tests/rust/executor_tests.rs (1)
281-293: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the
Defaultderive onGate.
Defaultproducesopen: false, which is a shut gate.opened()is the only constructor in use. A laterGate::default()would park every writer and hang the test with no obvious cause.♻️ Proposed change
-#[derive(Default)] struct Gate { open: Mutex<bool>, changed: Condvar, }🤖 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 `@crates/taskito-core/tests/rust/executor_tests.rs` around lines 281 - 293, Remove the #[derive(Default)] attribute from the Gate struct, keeping opened() as its sole constructor. Do not alter the open mutex initialization or Condvar behavior.
🤖 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 `@sdks/node/src/detached.ts`:
- Around line 37-48: Replace the process-wide sink in
sdks/node/src/detached.ts#L37-L48 with executor- or detached-queue-scoped
registration and disposal; update detached queue write routing in
sdks/node/src/detached.ts#L92-L110 to use that queue’s sink, register only the
current executor’s sink in sdks/node/src/executor.ts#L155-L159, and dispose only
that executor’s registration in sdks/node/src/executor.ts#L229-L231 so multiple
executors remain isolated.
---
Nitpick comments:
In `@crates/taskito-core/tests/rust/executor_tests.rs`:
- Around line 281-293: Remove the #[derive(Default)] attribute from the Gate
struct, keeping opened() as its sole constructor. Do not alter the open mutex
initialization or Condvar behavior.
🪄 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: 9146b171-c9dc-448d-a250-a1778b56efee
📒 Files selected for processing (14)
crates/taskito-core/BINDING_CONTRACT.mdcrates/taskito-core/Cargo.tomlcrates/taskito-core/src/worker/executor.rscrates/taskito-core/src/worker/mod.rscrates/taskito-core/src/worker/protocol.rscrates/taskito-core/src/worker/remote.rscrates/taskito-core/tests/rust/executor_tests.rsdocs/content/docs/shared/guides/operations/executor.mdxsdks/node/src/context.tssdks/node/src/detached.tssdks/node/src/executor.tssdks/node/src/index.tssdks/node/src/task-callback.tssdks/node/test/worker/executorAttach.test.ts
🔗 Linked repositories identified
CodeRabbit considers these linked repositories for cross-repo context during reviews:
ByteVeda/taskito(manual)
A module-level sink let a second attach take the first executor's writes, and either one stopping silenced the other.
pratyush618
changed the title
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
sdks/node/src/executor.ts (1)
164-185: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winClear the sink when executor startup fails.
If
resources.acquireWorker()orworker.startedthrows after Line 164, this path shuts downnativebut leavessinkinstalled. NoExecutoris returned, soteardown()cannot clear it.Clear this sink after shutdown.
clearSink(queue, sink)preserves a replacement sink if another attach installed one while shutdown was pending.Proposed fix
} catch (error) { // The session is live by now and no caller holds an `Executor` to stop // it, so a throwing resource factory or `worker.started` listener would // leak the attach until the process exits. await native.shutdown().catch((failure) => { log.debug(() => "releasing the attach after a failed start failed", failure); }); + clearSink(queue, sink); throw error; }🤖 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/executor.ts` around lines 164 - 185, Update the startup-failure catch block after resources.acquireWorker() and the worker.started emission to call clearSink(queue, sink) after native.shutdown() completes. Preserve clearSink’s replacement-sink behavior and keep the existing shutdown failure logging and original error propagation unchanged.
🤖 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.
Outside diff comments:
In `@sdks/node/src/executor.ts`:
- Around line 164-185: Update the startup-failure catch block after
resources.acquireWorker() and the worker.started emission to call
clearSink(queue, sink) after native.shutdown() completes. Preserve clearSink’s
replacement-sink behavior and keep the existing shutdown failure logging and
original error propagation unchanged.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 615fa6a3-24cd-41e0-a396-ab49f2c8e67d
📒 Files selected for processing (3)
sdks/node/src/detached.tssdks/node/src/executor.tssdks/node/test/worker/detachedSink.test.ts
🔗 Linked repositories identified
CodeRabbit considers these linked repositories for cross-repo context during reviews:
ByteVeda/taskito(manual)
kartikeya-27
approved these changes
Aug 2, 2026
This was referenced Aug 2, 2026
Merged
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
2 participants
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.
Three gaps left by the side-channel work (#589 and its children). #592 and #594 were already complete.
#591 — an unknown frame tore the attach down
The scheduler's reader used the strict
FrameReader::read, so a frame type it had no variant for surfaced as a parse error, and the only thing a reader loop can do with one isbreak→abandon(&executor): a live peer deregistered and its in-flight jobs left to the reaper.Structural rather than an oversight — the payload length sits inside the header, so a header that will not deserialize gives no skip length.
FrameReader::read_or_skipparses in two stages: the typed parse first (a known frame pays nothing for the tolerance), and only on failure aFramePreamblecarryingtypeplus apayload_lenthat aliases the two legacy names. It skips the declared bytes and returnsIncoming::Unknown, so the stream stays aligned on the next frame.Frame::is_known_typekeeps that narrow: a tag this build does know whose header will not parse is still an error, because skipping a malformedjobwould silently lose a dispatch. Both readers use the tolerant form — scheduler and executor — each warning once per type per connection so a newer peer cannot flood the log.BINDING_CONTRACT.md's frame table had gone stale (noprogress/task_log, no capability list). Updated, along with the forward rule this depends on: a frame type added from now on must declare its payload length aspayload_len.The issue also called for progress to be clamped 0–100. Left as a drop: the in-process storage path rejects out-of-range, so dropping matches it, and clamping 5000 → 100 would show a job as complete.
#590 — the shed counter was never asserted
dropped_logsexisted but only surfaced in a warning at session end, so nothing proved a flood sheds rather than blocks or grows.ExecutorSideChannel::dropped_task_logs()exposes it.The test fills the queue deterministically instead of racing the drain: a test-local transport whose write half parks on a gate. With the writer held shut the result loop can hold at most one line outside the queue, so every push past the capacity must shed and every progress report must land on the same map entry. It asserts the flood never parks the task, that at least
FLOOD - CAPACITY - 1lines were shed, that the newest line still arrives (drop-oldest), and that 500 progress reports coalesce to at most two frames ending on the newest value.#593 — Node's queue-level shim still dropped
detached.tswarned and droppedupdateProgress/writeTaskLog; only the task-callback seam had been wired, so a task reaching the queue directly reported nothing while the same call through the job context worked. It now installs a sink on attach and clears it on teardown, matching the Python shell.The task-callback overrides stay.
deps.queueis only the detached stand-in when the process is ataskito executor; one started from a process that does have storage would otherwise write progress into its own database, naming a job that lives in the scheduler's.Also adds
currentJob().log(message, level?, extra?)—publishwas the only route to a task log, so a plain log line had nowhere to go at all.TaskLogWriteLevelexcludesresult, which ispublish's. Both share one encoder that falls back to the value's string form rather than failing a task over a circular reference.Verification
cargo test --workspacegreen;cargo fmt --checkandclippy --all-targets --all-features -- -D warningsclean; default,postgresandredisall checkCloses #590
Closes #591
Closes #593
Summary by CodeRabbit
New Features
Bug Fixes
Documentation