feat(core): shared worker protocol and RemoteDispatcher - #583
Conversation
The module became a directory; both references still pointed at the old file.
One JSON header line then the payload's raw bytes, so the same format serves a pipe and a socket and the blob stays the wire-envelope bytes. Closes #547.
Replaces the base64-in-JSON frames and the one-way ready signal with raw payload bytes and a versioned hello/hello_ack, so a mismatched TASKITO_PYTHON interpreter fails closed instead of silently.
The registry reports the supplied pool_type so it never claims 'native' for a pool that isn't.
split() hands back the read/write halves plus a Connection control, because the read timeout and the close both have to outlive the split.
Dispatches only task names an executor advertised in hello; a dropped executor leaves its in-flight jobs to the scheduler's reaper. Closes #548.
|
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 (8)
📝 WalkthroughWalkthroughThe PR adds a shared worker frame protocol with binary payload support, transport abstractions, remote executor dispatch, configurable worker dispatchers, and Python prefork integration. It adds protocol, transport, remote-dispatch, and binary-payload tests. ChangesWorker execution infrastructure
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related issues
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
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 |
✅ Action performedFull review finished. |
![coderabbitai[bot]](https://avatars.github.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 7
🧹 Nitpick comments (5)
crates/taskito-core/src/worker/runner.rs (1)
115-131: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winWarn at spawn when handlers are registered alongside a custom dispatcher.
The doc comment states that registered handlers are unused with a custom dispatcher.
spawnthen dropsregistrysilently, because only theunwrap_or_elsebranch consumes it. A caller who uses both.register(...)and.dispatcher(...)gets tasks that never run and no diagnostic. A log line at spawn makes the misconfiguration visible.If
TaskRegistryexposes a length or emptiness accessor, add this inspawnafter the dispatcher is resolved:if pool_type != "native" && !registry.is_empty() { log::warn!( "[taskito] {} handler(s) are registered but a custom '{pool_type}' dispatcher is \ configured; the registered handlers will not run", registry.len(), ); }Note that
registryis moved intoNativeDispatcher::newin the default branch, so the check must read it before theunwrap_or_elseruns, or the branch must be restructured to an explicitmatch.🤖 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/runner.rs` around lines 115 - 131, Update spawn’s dispatcher-resolution logic to warn when a non-native custom dispatcher is configured alongside registered handlers. Check registry emptiness and length before registry is moved into NativeDispatcher::new, using an explicit match if needed, and emit the specified warning with the pool type and handler count.crates/taskito-core/src/worker/transport.rs (2)
174-199: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDocument that
closeon a memory transport is one-directional.
Connection::closeon a socket transport callsshutdown(Both), so both peers observe the teardown. OnMemoryTransportit callsclose_readeronincomingonly. The peer end keeps its reader blocked and never sees EOF. For an embedded in-process executor this means it will not learn that the scheduler closed the connection. The current dispatcher use indrain_and_closeonly needs the local reader to return, so behavior is correct today. Consider closingoutgoingas well, or state the limitation in the doc comment.🤖 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/transport.rs` around lines 174 - 199, Document in the `MemoryTransport` implementation that `Connection::close` only closes the local `incoming` reader, unlike socket shutdown, so the peer may remain blocked without observing EOF. Keep the current `drain_and_close` behavior unchanged unless the implementation is explicitly updated to close `outgoing` as well.
314-379: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a test for
Connection::closeunblocking a parked reader.The tests cover byte transfer, EOF on writer drop, and read timeouts. They do not cover
Connection::close. That path is the mechanismRemoteDispatcher::drain_and_closedepends on to bound shutdown. A direct unit test here would catch a regression inChannel::close_readerwithout relying on the dispatcher-level test.💚 Suggested test
#[test] fn closing_the_connection_unblocks_a_parked_reader() { let (a, b) = MemoryTransport::pair(); let (_a_read, _a_write, _a_conn) = Box::new(a).split().expect("split a"); let (mut b_read, _b_write, b_conn) = Box::new(b).split().expect("split b"); let reader = std::thread::spawn(move || { let mut buf = [0u8; 1]; b_read.read(&mut buf) }); std::thread::sleep(Duration::from_millis(20)); b_conn.close(); assert_eq!(reader.join().expect("reader thread").expect("read"), 0); }🤖 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/transport.rs` around lines 314 - 379, Add a unit test alongside the existing MemoryTransport tests that splits a pair, parks a reader thread on the read half, calls the corresponding Connection::close method, and asserts the reader unblocks with a zero-byte read. Use the split connection handle and a short synchronization delay, preserving the existing test style.crates/taskito-core/src/worker/remote.rs (1)
515-589: 🩺 Stability & Availability | 🔵 Trivial
idle_msis recorded but never enforced.
last_seen_msis updated on every frame and surfaced asAttachedExecutor::idle_ms, but nothing in the dispatcher acts on it. A half-open connection (peer host lost, no FIN) keeps the executor registered with its slots advertised.placewill keep selecting it, and eachwrite_jobsucceeds into the kernel buffer until TCP eventually gives up. Jobs dispatched in that window are only recovered by the scheduler's stale-job sweep.A follow-up liveness reaper — deregister and
connection.close()any executor whoseidle_msexceeds a configured multiple of the heartbeat interval — would close this gap. TCP keepalive onTcpTransportwould also shorten detection.🤖 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 515 - 589, Add liveness enforcement for executors tracked by the dispatcher: periodically inspect each executor’s last-seen time and, when its idle duration exceeds the configured heartbeat-based threshold, deregister it and close its connection. Integrate this reaper with the existing executor lifecycle and ensure stale executors no longer remain eligible for job placement.crates/taskito-core/BINDING_CONTRACT.md (1)
99-138: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a language to the frame-example fence.
The frame illustration in the fenced code block has no language identifier, unlike the equivalent example in
worker/protocol.rs's module doc, which uses ```text. This triggers a markdownlint MD040 warning.📝 Proposed fix
-``` +```text {"type":"job","id":"018f…","task_name":"resize","payload_len":7,…}\n <7 raw bytes></details> Aside from this, the frame table and rules match the `SchedulerMessage`/`ExecutorMessage` definitions and limits in `worker/protocol.rs` field-for-field. <details> <summary>🤖 Prompt for AI Agents</summary>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/BINDING_CONTRACT.mdaround lines 99 - 138, Update the
frame illustration fenced code block in the Worker frame protocol section of
BINDING_CONTRACT.md to include the text language identifier, matching the
equivalent worker/protocol.rs example; leave the example contents unchanged.</details> <!-- cr-comment:v1:42475594243691fca336cf16 --> _Source: Linters/SAST tools_ </blockquote></details> </blockquote></details> <details> <summary>🤖 Prompt for all review comments with AI agents</summary>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/lib.rs:
- Around line 49-51: Re-export Capacity from the crate root alongside
AttachedExecutor, RemoteConfig, and RemoteDispatcher in the public export list,
so consumers can name the return type of RemoteDispatcher::capacity() without
accessing the worker module directly.In
@crates/taskito-core/src/worker/protocol.rs:
- Around line 369-381: Update read_header_line to distinguish an unterminated
header caused by EOF from one that consumed MAX_HEADER_BYTES: return
ProtocolError::HeaderTooLarge only when the byte count reaches the cap, and map
a shorter partial header to ProtocolError::Io. Add a regression test alongside
truncated_payload_is_an_error_not_a_clean_eof covering a source closed after a
partial header without a newline and asserting the Io variant.In
@crates/taskito-core/src/worker/remote.rs:
- Around line 452-479: Update dispatch_to so the blocking writer.write_job call
cannot indefinitely occupy a Tokio runtime worker: either configure and apply a
bounded socket write timeout through Connection, move each executor’s writes
behind a bounded per-executor writer queue/thread, or execute the write via
tokio::task::spawn_blocking with the existing failure cleanup preserved. Ensure
dispatch and the cancel router remain responsive when an executor stops reading.- Around line 366-376: Update attach to reject new connections once shutdown has
started by adding a shutdown-related AttachError variant and checking
self.shutdown while holding the executors lock immediately before inserting the
executor. Preserve normal attachment behavior before shutdown, and ensure the
check is inside the same critical section as the registry insert to prevent
attachments racing with drain_and_close.- Around line 696-699: Make drain_and_close asynchronous and replace the
blocking std::thread::sleep polling with an awaited async delay while retaining
the deadline and Executor::is_busy conditions. Update its call site in the async
run method to await drain_and_close, leaving the bounded handle.join calls
unchanged.- Around line 310-312: Update the reader attachment flow around spawn_reader and
self.readers so it removes finished JoinHandles before pushing the newly spawned
handle. Retain active handles, then notify capacity changes as before, ensuring
repeated reconnects do not grow self.readers indefinitely.In
@crates/taskito-python/src/prefork/child.rs:
- Around line 90-121: Update the handshake failure paths in spawn_child to
explicitly terminate and reap the Child process before returning Err, including
read/decode failures, non-Hello messages, acknowledgment write failures, and
protocol-version mismatches. Ensure cleanup occurs for every error branch while
preserving the existing error messages and successful return path.
Nitpick comments:
In@crates/taskito-core/BINDING_CONTRACT.md:
- Around line 99-138: Update the frame illustration fenced code block in the
Worker frame protocol section of BINDING_CONTRACT.md to include the text
language identifier, matching the equivalent worker/protocol.rs example; leave
the example contents unchanged.In
@crates/taskito-core/src/worker/remote.rs:
- Around line 515-589: Add liveness enforcement for executors tracked by the
dispatcher: periodically inspect each executor’s last-seen time and, when its
idle duration exceeds the configured heartbeat-based threshold, deregister it
and close its connection. Integrate this reaper with the existing executor
lifecycle and ensure stale executors no longer remain eligible for job
placement.In
@crates/taskito-core/src/worker/runner.rs:
- Around line 115-131: Update spawn’s dispatcher-resolution logic to warn when a
non-native custom dispatcher is configured alongside registered handlers. Check
registry emptiness and length before registry is moved into
NativeDispatcher::new, using an explicit match if needed, and emit the specified
warning with the pool type and handler count.In
@crates/taskito-core/src/worker/transport.rs:
- Around line 174-199: Document in the
MemoryTransportimplementation that
Connection::closeonly closes the localincomingreader, unlike socket
shutdown, so the peer may remain blocked without observing EOF. Keep the current
drain_and_closebehavior unchanged unless the implementation is explicitly
updated to closeoutgoingas well.- Around line 314-379: Add a unit test alongside the existing MemoryTransport
tests that splits a pair, parks a reader thread on the read half, calls the
corresponding Connection::close method, and asserts the reader unblocks with a
zero-byte read. Use the split connection handle and a short synchronization
delay, preserving the existing test style.</details> <details> <summary>🪄 Autofix (Beta)</summary> Fix all unresolved CodeRabbit comments on this PR: - [ ] <!-- {"checkboxId": "4b0d0e0a-96d7-4f10-b296-3a18ea78f0b9"} --> Push a commit to this branch (recommended) - [ ] <!-- {"checkboxId": "ff5b1114-7d8c-49e6-8ac1-43f82af23a33"} --> Create a new PR with the fixes </details> --- <details> <summary>ℹ️ Review info</summary> <details> <summary>⚙️ Run configuration</summary> **Configuration used**: Organization UI **Review profile**: CHILL **Plan**: Pro Plus **Run ID**: `eaad64c7-6191-48a2-8020-6f11c3fb0fbb` </details> <details> <summary>📥 Commits</summary> Reviewing files that changed from the base of the PR and between 5508015902c6ab4ae31d6a90bff8c58ed2c3f274 and 4e1882c5fb42d02be8a8c9a510ef1f47233355e7. </details> <details> <summary>⛔ Files ignored due to path filters (1)</summary> * `Cargo.lock` is excluded by `!**/*.lock` </details> <details> <summary>📒 Files selected for processing (19)</summary> * `crates/taskito-core/BINDING_CONTRACT.md` * `crates/taskito-core/src/lib.rs` * `crates/taskito-core/src/worker/mod.rs` * `crates/taskito-core/src/worker/protocol.rs` * `crates/taskito-core/src/worker/remote.rs` * `crates/taskito-core/src/worker/remote_tests.rs` * `crates/taskito-core/src/worker/runner.rs` * `crates/taskito-core/src/worker/transport.rs` * `crates/taskito-python/Cargo.toml` * `crates/taskito-python/src/lib.rs` * `crates/taskito-python/src/prefork/child.rs` * `crates/taskito-python/src/prefork/mod.rs` * `crates/taskito-python/src/prefork/protocol.rs` * `docs/content/docs/python/guides/advanced-execution/prefork.mdx` * `sdks/python/taskito/_taskito.pyi` * `sdks/python/taskito/prefork/child.py` * `sdks/python/taskito/worker_protocol.py` * `sdks/python/tests/worker/prefork_apps/timeout_app.py` * `sdks/python/tests/worker/test_prefork.py` </details> <details> <summary>🔗 Linked repositories identified</summary> CodeRabbit considers these linked repositories for cross-repo context during reviews: - `ByteVeda/taskito` _(manual)_ </details> <details> <summary>💤 Files with no reviewable changes (2)</summary> * crates/taskito-python/src/prefork/protocol.rs * crates/taskito-python/Cargo.toml </details> </details> <!-- This is an auto-generated comment by CodeRabbit for review status -->
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/runner.rs (1)
167-189: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick winWire remote cancel notifications through the configured dispatcher.
Worker::spawn()stores the user-provided dispatcher butWorkerHandle::shutdown()is the only code path that callsdispatcher.shutdown(). Jobs running underRemoteDispatcherneednotify_cancel(job_id)delivered through that same dispatcher; otherwise only the storage cancel flag can terminate a blocked/long-running remote worker instead of using the executor-side cancel channel.🤖 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/runner.rs` around lines 167 - 189, Update Worker::spawn and WorkerHandle::shutdown so the configured WorkerDispatcher’s notify_cancel(job_id) is invoked for remote cancellation, using the same dispatcher instance stored during spawn rather than creating or using a separate dispatcher. Preserve the existing storage cancellation behavior and dispatcher shutdown flow.
♻️ Duplicate comments (1)
crates/taskito-python/src/prefork/child.rs (1)
90-121: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winKill the child process on every handshake failure path.
Every
Errreturn inspawn_child(read failure, non-hello frame, ack write failure, version mismatch) dropsprocesswithout callingkill/wait.Child::dropdoes not terminate the process, so the child becomes an orphan (a zombie on Unix). On the version-mismatch branch,is_alive()never reports the process as dead, so the restart logic inmod.rsnever replaces this slot.Call
ChildProcess::kill_and_reap-equivalent cleanup (process.kill()+process.wait()) on each error branch before returningErr.This was previously flagged as unresolved in a review comment on this exact code segment.
🤖 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-python/src/prefork/child.rs` around lines 90 - 121, Update the handshake error paths in spawn_child to terminate and reap the child process before returning Err. Apply process.kill() followed by process.wait() for read failures, non-hello frames, acknowledgement write failures, and protocol-version mismatches, while preserving the existing error messages and successful return path.
🧹 Nitpick comments (4)
crates/taskito-core/BINDING_CONTRACT.md (1)
108-111: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueAdd a language to the fenced block.
markdownlint reports MD040 for this fence. The module doc in
crates/taskito-core/src/worker/protocol.rsuses ```text for the same example. Use the same tag here.♻️ Proposed fix
-``` +```text {"type":"job","id":"018f…","task_name":"resize","payload_len":7,…}\n <7 raw bytes></details> <details> <summary>🤖 Prompt for AI Agents</summary>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/BINDING_CONTRACT.mdaround lines 108 - 111, Update the
fenced example in BINDING_CONTRACT.md to use the text language tag, matching the
existing protocol documentation convention and satisfying markdownlint MD040.</details> <!-- cr-comment:v1:9efc93069e8d48693f03fc07 --> _Source: Linters/SAST tools_ </blockquote></details> <details> <summary>crates/taskito-core/src/worker/remote_tests.rs (2)</summary><blockquote> `165-166`: _📐 Maintainability & Code Quality_ | _🔵 Trivial_ | _💤 Low value_ **Correct the `with_running` doc comment.** The comment says "current-thread runtime", but the helper builds a multi-thread runtime with two worker threads. The tests depend on that, because the body calls `blocking_send` and blocking frame reads while `run` progresses. <details> <summary>♻️ Proposed fix</summary> ```diff -/// Run `body` with the dispatcher's `run` loop live on a current-thread runtime. +/// Run `body` with the dispatcher's `run` loop live on a multi-thread runtime. +/// The body blocks on channel sends and frame reads, so `run` needs its own +/// worker thread to make progress.🤖 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_tests.rs` around lines 165 - 166, Update the doc comment for with_running to state that it runs the dispatcher loop on a multi-thread runtime with two worker threads, preserving the documented behavior needed for blocking sends and frame reads.
74-108: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winA regression makes these tests hang instead of fail.
FakeExecutor::readblocks with no timeout, becauseattach_with_versiondiscards the executor end'sConnectionat line 52.expect_job,expect_hello_ack, andexpect_shutdowntherefore block forever if the dispatcher stops writing the expected frame. The suite then hits the CI job timeout with no failing assertion. Keep theConnectionand set a read timeout bounded bySETTLE, so a missing frame surfaces as a failed expectation.♻️ Proposed refactor
struct FakeExecutor { reader: FrameReader<ReadHalf>, writer: FrameWriter<WriteHalf>, + connection: Connection, }- let (read, write, _timeout) = Box::new(executor_end).split().expect("split executor end"); + let (read, write, connection) = Box::new(executor_end).split().expect("split executor end"); + connection + .set_read_timeout(Some(SETTLE)) + .expect("bound executor reads"); let mut executor = Self { reader: FrameReader::new(read), writer: FrameWriter::new(write), + connection, };Import
Connectionalongside the other transport items on line 13.🤖 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_tests.rs` around lines 74 - 108, Update attach_with_version and FakeExecutor to retain the executor-side Connection returned during setup, import Connection with the existing transport types, and configure its read timeout to the SETTLE duration. Ensure FakeExecutor::read uses this timed connection so expect_job, expect_hello_ack, and expect_shutdown fail promptly when frames are missing rather than blocking indefinitely.crates/taskito-core/src/worker/transport.rs (1)
252-288: 🩺 Stability & Availability | 🔵 Trivial | 💤 Low valueThe read timeout is sampled once per
readcall, so a cleared timeout does not wake a parked reader.
readcopiesread_timeoutat line 257 and then parks with that value. IfConnection::set_read_timeout(None)runs while a reader is parked withSome(limit), that reader still returnsWouldBlockatlimit. The currentattachsequence incrates/taskito-core/src/worker/remote.rsclears the timeout only after the handshake read returns, so this is not reachable today. Re-reading the timeout inside the loop would make the transport robust against a future caller that clears it concurrently.♻️ Proposed refactor
- let timeout = *self.channel.read_timeout.lock().unwrap_or_else(recover); let mut state = self.channel.state.lock().unwrap_or_else(recover); while state.buffer.is_empty() { // Before the peer splits, `writer_open` is false but no data can // have been lost — treating that as EOF would race the handshake. if state.writer_ever_opened && !state.writer_open { return Ok(0); } + // Re-read each pass so a cleared timeout takes effect on a parked + // reader instead of expiring it. + let timeout = *self.channel.read_timeout.lock().unwrap_or_else(recover); state = match timeout {🤖 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/transport.rs` around lines 252 - 288, Update ChannelReader::read to re-read self.channel.read_timeout on each loop iteration before waiting, rather than caching it once per read call. This must let a concurrently cleared timeout switch the parked reader to an unbounded wait, while preserving the existing timed WouldBlock behavior when a timeout remains configured.
🤖 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 386-410: Update place to check self.shutdown within each
placement-loop iteration and fail the job retryably when shutdown is active,
allowing the reaper and retry policy to handle it instead of waiting for
placement_timeout. Also ensure shutdown() wakes the capacity_changed waiter so a
task parked in changed.await observes the flag immediately.
In `@sdks/python/taskito/worker_protocol.py`:
- Around line 40-48: Update declared_payload_len to validate payload_len and
result_len as non-negative integers before returning them, rejecting invalid
header values rather than passing them to read operations. Ensure read_frame and
write_frame both use this validation path before reading or writing payload
data, while preserving zero for headers without a declared length.
---
Outside diff comments:
In `@crates/taskito-core/src/worker/runner.rs`:
- Around line 167-189: Update Worker::spawn and WorkerHandle::shutdown so the
configured WorkerDispatcher’s notify_cancel(job_id) is invoked for remote
cancellation, using the same dispatcher instance stored during spawn rather than
creating or using a separate dispatcher. Preserve the existing storage
cancellation behavior and dispatcher shutdown flow.
---
Duplicate comments:
In `@crates/taskito-python/src/prefork/child.rs`:
- Around line 90-121: Update the handshake error paths in spawn_child to
terminate and reap the child process before returning Err. Apply process.kill()
followed by process.wait() for read failures, non-hello frames, acknowledgement
write failures, and protocol-version mismatches, while preserving the existing
error messages and successful return path.
---
Nitpick comments:
In `@crates/taskito-core/BINDING_CONTRACT.md`:
- Around line 108-111: Update the fenced example in BINDING_CONTRACT.md to use
the text language tag, matching the existing protocol documentation convention
and satisfying markdownlint MD040.
In `@crates/taskito-core/src/worker/remote_tests.rs`:
- Around line 165-166: Update the doc comment for with_running to state that it
runs the dispatcher loop on a multi-thread runtime with two worker threads,
preserving the documented behavior needed for blocking sends and frame reads.
- Around line 74-108: Update attach_with_version and FakeExecutor to retain the
executor-side Connection returned during setup, import Connection with the
existing transport types, and configure its read timeout to the SETTLE duration.
Ensure FakeExecutor::read uses this timed connection so expect_job,
expect_hello_ack, and expect_shutdown fail promptly when frames are missing
rather than blocking indefinitely.
In `@crates/taskito-core/src/worker/transport.rs`:
- Around line 252-288: Update ChannelReader::read to re-read
self.channel.read_timeout on each loop iteration before waiting, rather than
caching it once per read call. This must let a concurrently cleared timeout
switch the parked reader to an unbounded wait, while preserving the existing
timed WouldBlock behavior when a timeout remains configured.
🪄 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: c090e563-84ca-4b6a-9b27-9ee3d96223c0
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (19)
crates/taskito-core/BINDING_CONTRACT.mdcrates/taskito-core/src/lib.rscrates/taskito-core/src/worker/mod.rscrates/taskito-core/src/worker/protocol.rscrates/taskito-core/src/worker/remote.rscrates/taskito-core/src/worker/remote_tests.rscrates/taskito-core/src/worker/runner.rscrates/taskito-core/src/worker/transport.rscrates/taskito-python/Cargo.tomlcrates/taskito-python/src/lib.rscrates/taskito-python/src/prefork/child.rscrates/taskito-python/src/prefork/mod.rscrates/taskito-python/src/prefork/protocol.rsdocs/content/docs/python/guides/advanced-execution/prefork.mdxsdks/python/taskito/_taskito.pyisdks/python/taskito/prefork/child.pysdks/python/taskito/worker_protocol.pysdks/python/tests/worker/prefork_apps/timeout_app.pysdks/python/tests/worker/test_prefork.py
🔗 Linked repositories identified
CodeRabbit considers these linked repositories for cross-repo context during reviews:
ByteVeda/taskito(manual)
💤 Files with no reviewable changes (2)
- crates/taskito-python/src/prefork/protocol.rs
- crates/taskito-python/Cargo.toml
RemoteDispatcher::capacity() returns it, so a root-level consumer could not name its own return type.
A peer closing mid-header reported HeaderTooLarge, contradicting the Eof contract and mislabelling a plain disconnect as an oversized frame.
Child::drop does not terminate the process, so a version-mismatched child survived as an orphan the restart path never notices.
A negative length reached stream.read(), which drains the connection to EOF rather than reading a payload.
The vector only shrank at shutdown, so a reconnecting executor grew it for the life of the process.
An executor attaching after the registry drained was never sent a shutdown and never joined, leaking its reader thread.
A job parked on placement ignored the flag, so shutdown blocked for a full placement_timeout before the drain even began.
run shares a runtime with the scheduler task, so a blocking sleep starved it for up to the whole drain budget.
An executor that stops reading fills the send buffer; the write had no timeout and could park a runtime thread indefinitely.
2 participants
First two phases of Executor Attach (#546): the shared wire format, and the dispatcher that speaks it over a socket. Pure Rust in
taskito-coreplus the prefork migration — no server binary, no listener, no SDK executors, no dashboard.Closes #547. Closes #548.
#547 — shared worker protocol
crates/taskito-core/src/worker/protocol.rsnow owns the frames;crates/taskito-python/src/prefork/protocol.rsis gone and prefork runs on the shared module.A frame is a JSON header line followed by exactly the raw payload bytes it declares, replacing base64-in-JSON. Over a socket that base64 was a permanent 33% size tax plus encode/decode on every payload and result; raw bytes also mean the blob on the wire is the CBOR wire-envelope bytes the contract describes.
base64drops out ofcrates/taskito-python/Cargo.tomlentirely.Prefork's one-way
readysignal is replaced by the samehello/hello_ackan attached executor uses, so there is one handshake rather than two.protocol_versionrides on both frames — the issue's table put it only on the ack, but one-sided validation lets a mismatched peer stay attached until it happens to notice. The ack is sent even when rejecting, so both ends can log both versions. This is not hypothetical:TASKITO_PYTHONlets the prefork child run from a different interpreter, so a version-mismatched install is reachable.Python side gains
sdks/python/taskito/worker_protocol.py— transport-agnostic, so the socket executor reuses it — reading the version from_taskito.WORKER_PROTOCOL_VERSIONrather than mirroring a literal.#548 — RemoteDispatcher + Transport
transport.rs(UDS, TCP, in-memory) andremote.rs(RemoteDispatcher+ executor registry). Binding and accepting stay with the caller —attach()takes an already-connected transport — so the listener and its auth defaults land in a later phase without this one guessing at them.Jobs go only to executors that advertised the task name in
hello. Free-slot counts are the only limiter, withcapacity()exposed so the server phase can sizeSchedulerConfig::max_in_flightfrom it instead of running a parallel limiter. A job nobody advertises fails retryably afterplacement_timeout— it reschedules under the normal retry policy and surfaces the misconfiguration rather than stalling silently.An executor that drops mid-job gets no synthesised result: it may have run, so recovery stays with the scheduler's reaper, matching prefork's existing behaviour.
Worker::dispatcher(pool_type, …)is the one change outside the new modules — it makes the reaper path testable end to end, and the server phase needs it regardless. The registry reports the suppliedpool_typeso it never claimsnativefor a pool that isn't.Two bugs the design invites
Both were found by noticing every remote test took exactly 10.00s — the default
handshake_timeout. Both are fixed, and each regression test was verified to fail without its fix.Transport::splitnow returns aConnectioncontrol that outlives the split;attachbounds the hello read, then clears it.Connection::close()after a boundedshutdown_drainnow bounds it.Separately,
place()pins and enables theNotify::notified()future before checking capacity:notify_waiters()only wakes already-registered waiters, so subscribing lazily loses a slot freed in the check-then-await window.Other SDKs
None needed. Node and Java never use
taskito_core::Worker, have no prefork or out-of-process pool, and reference nothing in the new modules. Theirexecutorsubcommands are separate phases and will build against the contract section added here.Verification
cargo test --workspacegreen; every commit compiles standalonecargo clippy --workspace --all-targetsclean;--features postgres|redis|native-asyncall checkscripts/version.mjs --checkpassesNone-returning taskSummary by CodeRabbit
New Features
Improvements
Documentation