feat(core): executor side-channel for progress, logs and middleware toggles - #599
Conversation
Optional fields and a hello_ack capability list rather than a protocol bump, so a scheduler and its executors upgrade independently.
Queued off the reader thread so a database write never delays a result, and dropped unless the executor is actually running the job it names.
Queued and coalesced, so a task in a progress loop never parks on the socket, and silent when the scheduler advertised no support.
|
Warning Review limit reached
Next review available in: 13 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (5)
📝 WalkthroughWalkthroughThe worker protocol now negotiates side-channel capabilities and carries progress, task logs, metadata, and disabled middleware. Executors buffer and send reports. Remote dispatchers validate and forward reports to storage. ChangesExecutor side-channel
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Executor
participant RemoteDispatcher
participant Storage
Executor->>RemoteDispatcher: Negotiate side-channel capability
RemoteDispatcher->>Executor: Dispatch job with metadata and disabled middleware
Executor->>RemoteDispatcher: Send progress and task-log frames
RemoteDispatcher->>Storage: Persist progress and task logs
Executor->>RemoteDispatcher: Send job result
Possibly related issues
Possibly related PRs
Suggested labels: Suggested reviewers: 🚥 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.github.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
crates/taskito-core/src/worker/executor.rs (1)
899-916: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winRelease the toggle entry when a job is declined.
accept_jobcallsshared.remember_togglesbefore it can calldecline.declineframes aFailuredirectly and never callsforget_toggles, so a declined job with a non-empty disable list leaves an entry inShared::togglesfor the life of the session. The doc comment ontogglesstates that entries are released when the job reports, which does not hold on this path.Under sustained saturation the map grows once per declined job.
🐛 Proposed fix
fn decline(shared: &Arc<Shared>, job: &Job, reason: &str) { log::warn!( "[taskito] executor {} declining job {}: {reason}", shared.executor_id, job.id ); + // The job reports here rather than through `send_result`, so its toggle + // entry has to be released on this path too. + shared.forget_toggles(&job.id); let (frame, payload) = ExecutorMessage::from_job_result(JobResult::Failure {🤖 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/src/worker/executor.rs` around lines 899 - 916, Update decline to release the job’s toggle entry before sending its failure result by calling the existing Shared::forget_toggles mechanism for the declined job. Preserve the current warning, failure payload, and retry behavior while ensuring declined jobs with disable lists do not remain in Shared::toggles.
🧹 Nitpick comments (2)
crates/taskito-core/src/worker/remote.rs (2)
186-208: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueConsider tying the drain thread to
Shared's lifetime.
SideChannelPump::startspawns the drain thread inRemoteDispatcher::new, but onlyShared::runstops it throughstop_side_channel. A dispatcher that is built and then dropped withoutrunleaves the thread alive for the life of the process. The new capability tests incrates/taskito-core/tests/rust/remote_tests.rsdo exactly that.A
Dropimplementation onSharedthat callsstop_side_channelmakes the teardown self-contained and keepsstop_side_channelidempotent, since it already takes the handle out of theMutex.🤖 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/src/worker/remote.rs` around lines 186 - 208, Implement Drop for Shared so dropping it calls stop_side_channel, ensuring the SideChannelPump drain thread is stopped even when RemoteDispatcher::run is never invoked. Reuse the existing idempotent stop_side_channel method and its side_channel_drain handle rather than adding separate teardown logic.
986-1001: 🗄️ Data Integrity & Integration | 🔵 Trivial | 💤 Low valueUse
extra_lento tell an emptyextrafrom an absent one.
handle_framediscardsextra_len(extra_len: _), soapply_task_loginfers absence from an empty payload. The protocol keepsSome(0)andNonedistinct, andcrates/taskito-core/src/worker/protocol.rshas a test nameda_log_without_extra_is_distinct_from_one_with_empty_extrathat pins that distinction. Here both collapse toNone, so a storedextraof""becomesNULL.Pass
extra_lenthrough and branch on it to keep the wire contract and the stored row in agreement.🤖 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/src/worker/remote.rs` around lines 986 - 1001, Update handle_frame and apply_task_log to preserve the distinction between absent extra data and an explicitly empty extra value: pass extra_len through instead of discarding it, use it to determine None versus Some(""), and only decode the payload when present. Keep the existing invalid UTF-8 warning and dropping behavior for non-empty invalid blobs.
🤖 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-core/src/worker/remote.rs`:
- Around line 793-812: Update the dispatch flow around dispatch_to and its place
caller so disabled_middleware is not executed synchronously on the shared Tokio
runtime. Resolve the toggle list through the pump’s dedicated thread and cache
it in the dispatcher, or use tokio::task::spawn_blocking at the place call site,
then pass the resolved list to write_job_with while preserving the existing
behavior.
In `@crates/taskito-core/src/worker/side_channel.rs`:
- Around line 115-119: Validate the progress value in update_progress before
calling storage.update_progress, rejecting or clamping values outside the
documented 0..=100 range so attached SDK requests match in-process behavior.
Preserve the existing storage error warning for valid values.
---
Outside diff comments:
In `@crates/taskito-core/src/worker/executor.rs`:
- Around line 899-916: Update decline to release the job’s toggle entry before
sending its failure result by calling the existing Shared::forget_toggles
mechanism for the declined job. Preserve the current warning, failure payload,
and retry behavior while ensuring declined jobs with disable lists do not remain
in Shared::toggles.
---
Nitpick comments:
In `@crates/taskito-core/src/worker/remote.rs`:
- Around line 186-208: Implement Drop for Shared so dropping it calls
stop_side_channel, ensuring the SideChannelPump drain thread is stopped even
when RemoteDispatcher::run is never invoked. Reuse the existing idempotent
stop_side_channel method and its side_channel_drain handle rather than adding
separate teardown logic.
- Around line 986-1001: Update handle_frame and apply_task_log to preserve the
distinction between absent extra data and an explicitly empty extra value: pass
extra_len through instead of discarding it, use it to determine None versus
Some(""), and only decode the payload when present. Keep the existing invalid
UTF-8 warning and dropping behavior for non-empty invalid blobs.
🪄 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 Plus
Run ID: 55279bca-261c-4f87-9974-5a041b0c7590
📒 Files selected for processing (11)
crates/taskito-core/src/lib.rscrates/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/src/worker/side_channel.rscrates/taskito-core/tests/rust/executor_tests.rscrates/taskito-core/tests/rust/remote_tests.rscrates/taskito-python/src/prefork/child.rscrates/taskito-server/src/runtime/mod.rscrates/taskito-server/tests/attach_e2e.rs
🔗 Linked repositories identified
CodeRabbit considers these linked repositories for cross-repo context during reviews:
ByteVeda/taskito(manual)
|
All five findings were valid against the current code and all five are fixed, one commit each.
Toggle resolution. Declined jobs. Drain thread. Empty vs absent The two runtime findings have regression tests that were checked to fail with the fix reverted — the starvation test panics with "a parked settings read starved the runtime", and the drain test reports a strong count of 2.
|
2 participants
Part 1 of #589: the wire protocol and the scheduler-side plumbing that lets an attached executor report progress, task logs and published partials, and receive its middleware toggles, without holding a database connection of its own. The SDK sides (Python, Node, Java) follow in their own PRs on top of this one.
What changes
Protocol (
worker/protocol.rs)SchedulerMessage::HelloAckgrows acapabilities: Vec<String>list —#[serde(default)], so a legacy ack still parses as "no capabilities".PROTOCOL_VERSIONis untouched on purpose: bumping it forces scheduler and executors to upgrade in lockstep, which is the coupling attached executors exist to remove.CAP_SIDE_CHANNELis the one capability defined here.SchedulerMessage::Jobcarriesdisabled_middlewareandmetadataalongside the dispatch. These are reads the executor needs at dispatch time and the scheduler is already reading task config when it dispatches, so they ride along instead of costing a round trip. Both are additive and defaulted, so an older executor ignores them.ExecutorMessage::ProgressandExecutorMessage::TaskLogare the two new fire-and-forget frames.publishis aTaskLogat levelresult, so one frame covers streamed partials too.Scheduler side (
worker/side_channel.rs,worker/remote.rs)SideChannelis the narrow traitRemoteDispatcherneeds:update_progress,write_task_log,disabled_middleware. Every method is infallible — a task that only wanted to report progress must not fail because the database was briefly unhappy. A trait rather than anArc<dyn Storage>so the dispatcher is testable against a fake, and so the settings key toggles live under is spelled once.StorageSideChannelis the implementation a real deployment installs. Toggle lists are cached with a 5s TTL, matching the SDK worker caches (_MW_CHAIN_TTLin the Python shell): a dashboard toggle is rare, a dispatch is hot.RemoteDispatcherhandles the new frames inhandle_framein arms placed beforeinto_job_result— they are not results, and taking the in-flight entry would break the exactly-once accounting that entry exists for. Applying them is queued off the reader thread so a database write never delays a result, and a frame naming a job this executor is not running is dropped.Executor side (
worker/executor.rs)ExecutorSideChannelis what a task body reports through. Tasks never touch the sharedMutex<FrameWriter>directly; everything goes through a bounded outbound queue drained by the existing result thread, so a task in a progress loop can neither park on the socket nor flood the scheduler. Progress coalesces (idempotent-latest, only the newest value per job need survive a backlog); logs cannot coalesce, so the queue is bounded and drops oldest with a counter. Silent entirely when the scheduler advertised noside_channelcapability.Wiring
taskito-serverinstalls aStorageSideChannelover the backend it already holds — this process has the connection an executor deliberately does not.Tests
New coverage across
protocol.rs,side_channel.rs,tests/rust/executor_tests.rs,tests/rust/remote_tests.rsandtaskito-server/tests/attach_e2e.rs:extrastays distinct from one with emptyextracargo test --workspace,cargo clippy --workspace --all-targets,cargo fmt --check, andcargo checkunder--features postgresand--features redisare all clean.Refs #589.
Summary by CodeRabbit