feat(sdks): taskito executor subcommand for Python, Node and Java - #595
Merged
kartikeya-27 merged 46 commits intoAug 1, 2026
Merged
Conversation
|
Warning Review limit reached
Next review available in: 50 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 (7)
📝 WalkthroughWalkthroughThe PR adds a shared scheduler-attachment executor to the Rust core and exposes it through Node.js, Python, and Java SDKs. It adds detached storage and cancellation handling, lifecycle controls, CLIs, Java handler discovery, protocol conversion, and end-to-end tests. ChangesCore executor attachment
Node.js executor
Python executor
Java executor
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related issues
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 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 |
pratyush618
added
deployment
Contributor
Author
stromanni
force-pushed
the
feat/546-executor-attach-sdks
branch
2 times, most recently
from
August 1, 2026 04:16
9c5d725 to
7cdfec9
Compare
The attach protocol only had scheduler-side conversions. An executor needs the other direction: a Job rebuilt from a dispatch frame, and a result frame built from a JobResult.
ExecutorClient is Worker::spawn with storage swapped for a socket: it dials a scheduler, then feeds the same WorkerDispatcher every SDK already implements. Needs no Storage, so an executor image carries app code and no database credentials. Draining announces zero capacity by heartbeat before disconnecting, which the scheduler already honours, so in-flight work reports instead of waiting for a reap.
Every SDK executor needs to turn TASKITO_ATTACH into a connection, and the grammar has to match what the listener binds. One parser here rather than three in three languages, where unix: support would drift.
A worker reads the storage cancel flag; an attached executor has no storage and learns from the scheduler's cancel frame instead. Both answer one question, so both ask it in one place.
Slots are prefork children, so the timeout watchdog, least-loaded dispatch and restart-on-crash all come from shipped code and child.py is untouched. start/wait/stop rather than one blocking run: a Python signal handler only runs when the main thread holds the GIL, so a blocking call would make the process deaf to SIGTERM.
Reads TASKITO_ATTACH and TASKITO_SLOTS, the contract shared with the other SDKs. The token is env-only: in argv it would show up in ps and shell history.
A socket speaking the frame protocol stands in for the scheduler so these run without a Rust build; the same assertions run against the real taskito-server when TASKITO_SERVER_BIN points at one.
ExecutorHandle::wait consumes the handle, which a shell that must also hold it to shut down cannot do. A cloneable session view lets an async runtime watch for the scheduler ending the session.
The executor runs the same task bodies as the worker, so the middleware, codec and resource-scope handling is shared rather than copied.
run and executor both resolve a Queue from a module path.
Attaching is async: dialling and the handshake block, so on the JS thread they would freeze the event loop until the scheduler answers. The dispatcher now takes its cancels from the shared source, so a detached executor can learn of one without reading storage.
The inverse of runWorker: the scheduler holds the database connection and dispatches over a socket. Same middleware, codecs and resources — only the transport differs.
Reads TASKITO_ATTACH and TASKITO_SLOTS, the contract shared with the other SDKs. The token is env-only: in argv it would show up in ps.
A socket speaking the frame protocol stands in for the scheduler so these run without a Rust build; the same assertions run against the real taskito-server when TASKITO_SERVER_BIN points at one.
stop() cannot unpark the frame reader, which is blocked on a read only the scheduler could satisfy. A shell that stopped and then waited would hang instead of draining.
The processor now generates a ServiceLoader provider per annotated class and lists it in META-INF/services, so an executor finds handlers with no user main. ServiceLoader cannot load HandlerRegistry itself: on the classpath a listed class must be a subtype with a public no-arg constructor, and the static provider() form only applies to modules. Hence the new interface. Classes with no accessible no-arg constructor are skipped with a note — only their own code knows how to build them.
stop() is documented as safe on a signal-handling path, but begin_drain wrote the zero-capacity heartbeat first — a scheduler that stopped reading blocked that write for the whole write timeout.
teardown joined unconditionally after the drain budget, so a task that ignores its cancel kept the pool's run alive and the process never exited. The test could not catch it: its blocking behaviour released itself.
The companion is written at package level and names the handler by simple name, so an Outer.Inner provider would not compile. Also corrects the service type in the Javadoc.
System.exit runs hooks while main is still inside Runtime.exit, so a registered hook waited out the whole drain budget for a join that could not complete, then called stop() on an already-closed control.
A throwaway CommandLine routed a bad TASKITO_SLOTS through the execution-exception handler, so the user saw a stack trace rather than a usage message. Zero and negative counts also passed, and the builder clamped them behind the banner.
awaitSession held the read lock for the whole park. A queued close() then blocked a concurrent stop() — the only call that could have ended the session. Waiters are counted instead, and close drains them before freeing.
Nothing awaits a signal listener's return value, so a rejected stop() surfaced as an unhandled rejection and aborted the process mid-drain.
runExecutor tracked nothing, so queue.shutdown() left an attached executor running and its scheduler session open.
Nothing read the pipes, so a scheduler that logs per job fills the buffer and blocks on its next write.
Contributor
|
@CodeRabbit full review |
✅ 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: 5
🧹 Nitpick comments (3)
crates/taskito-core/tests/rust/executor_tests.rs (1)
59-60: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAssert that the pool observed
shutdown().
TestPool::shutdownrecords intoself.shutdown, but no test reads that field.ExecutorHandle::stopcallsself.dispatcher.shutdown(), andspawn_readercalls it again when the session ends. Neither call is covered.Add an accessor and assert it in the drain test, so a regression that stops signalling the pool is caught.
💚 Proposed test addition
fn cancelled(&self) -> Vec<String> { self.cancels.lock().expect("cancels").clone() } + + /// Whether the executor asked the pool to stop accepting work. + fn was_shut_down(&self) -> bool { + self.shutdown.load(Ordering::SeqCst) + }Then in
a_drain_announces_zero_capacity_before_anything_else, or instop_finishes_in_flight_work_before_disconnectingafter the result arrives:attached.handle.stop(); + wait_until( + || attached.pool.was_shut_down(), + "stop() must tell the pool to stop accepting work", + );Also applies to: 161-163
🤖 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 59 - 60, Add a `TestPool` accessor for its recorded `shutdown` flag, then update the drain-related test (`a_drain_announces_zero_capacity_before_anything_else` or `stop_finishes_in_flight_work_before_disconnecting`) to assert the accessor is true after shutdown completes, covering the `ExecutorHandle::stop`/`spawn_reader` signaling paths.sdks/java/src/main/java/org/byteveda/taskito/worker/Executor.java (1)
113-138: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winMake
close()idempotent across threads.
closedis a plain field andclose()is not synchronized.Executoris public, so a caller can close it from a shutdown hook while the main thread also closes it. Both threads can then pass the guard, callresources.teardownWorker()twice, and emitWORKER_STOPPEDtwice.synchronizedonclose()plus avolatileread keeps the documented idempotence.♻️ Proposed refactor
- private boolean closed; + private volatile boolean closed;`@Override` - public void close() { + public synchronized void close() { if (closed) { return; }🤖 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/java/src/main/java/org/byteveda/taskito/worker/Executor.java` around lines 113 - 138, Make Executor.close() thread-safe and idempotent by synchronizing the close() method and declaring the closed field volatile. Preserve the existing early-return guard so concurrent callers allow only one shutdown sequence, including resources.teardownWorker() and WORKER_STOPPED emission.sdks/java/src/test/java/org/byteveda/taskito/worker/ExecutorAttachTest.java (1)
65-65: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueRemove the unused
resultsdeque.Nothing adds to
results.readFrame()returns each frame directly, so the!results.isEmpty()branch at lines 190-192 is unreachable. Deleting the field and the branch makes the polling loop easier to read.♻️ Proposed cleanup
- private final Deque<JsonNode> results = new ArrayDeque<>();JsonNode nextResult() throws IOException { long deadline = System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(SETTLE_MS); while (System.nanoTime() < deadline) { - if (!results.isEmpty()) { - return results.removeFirst(); - } JsonNode frame = readFrame();Drop the now-unused
ArrayDequeandDequeimports.Also applies to: 186-202
🤖 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/java/src/test/java/org/byteveda/taskito/worker/ExecutorAttachTest.java` at line 65, Remove the unused results field from ExecutorAttachTest, delete the unreachable results check in the polling loop within readFrame-related logic, and remove the now-unused ArrayDeque and Deque imports while preserving direct frame handling.
🤖 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/protocol.rs`:
- Around line 244-293: Update the SchedulerMessage::Job dispatch representation
and SchedulerMessage::into_job to carry and reconstruct created_at,
scheduled_at, priority, metadata, unique_key, and notes instead of defaulting
them. Populate these fields when converting a Job into a dispatch message, and
ensure the Node and Python converters expose the reconstructed values
consistently for detached and attached workers.
In `@crates/taskito-python/src/executor.rs`:
- Around line 142-154: Update Executor::stop and its contract to acknowledge
that concurrent calls may block behind wait’s with_handle mutex for up to
timeout_ms, or otherwise ensure stop uses a separately shared primitive that
does not depend on the polling handle lock. Keep the existing wait behavior and
CLI signal-handling flow unchanged.
In
`@sdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.java`:
- Around line 35-38: Update the class Javadoc near the generated companion
description to state that an accessible no-arg constructor causes generation of
a nested Provider class implementing HandlerRegistryProvider, which is
registered through META-INF/services; remove the inaccurate provider() method
wording and preserve the explanation of classpath discovery.
In `@sdks/java/src/main/java/org/byteveda/taskito/cli/Cli.java`:
- Around line 294-296: Update the token assignment in Cli.Executor to pass
TASKITO_ATTACH_TOKEN through a blankToNull helper before calling token, treating
null and blank values as absent while preserving nonblank tokens. Add the helper
as a private static nullable method alongside the existing Executor helpers,
matching the address normalization behavior.
In `@sdks/python/tests/worker/test_executor_attach_server.py`:
- Around line 134-138: Replace the fixed one-second attach waits with a shared
polling readiness helper that confirms the executor is attached before
enqueueing. Update sdks/python/tests/worker/test_executor_attach_server.py lines
134-138, 146-150, and 175-178 to call the helper before enqueueing ECHO, BOOM,
and SLOW respectively; preserve each test’s existing enqueue and status
assertions.
---
Nitpick comments:
In `@crates/taskito-core/tests/rust/executor_tests.rs`:
- Around line 59-60: Add a `TestPool` accessor for its recorded `shutdown` flag,
then update the drain-related test
(`a_drain_announces_zero_capacity_before_anything_else` or
`stop_finishes_in_flight_work_before_disconnecting`) to assert the accessor is
true after shutdown completes, covering the
`ExecutorHandle::stop`/`spawn_reader` signaling paths.
In `@sdks/java/src/main/java/org/byteveda/taskito/worker/Executor.java`:
- Around line 113-138: Make Executor.close() thread-safe and idempotent by
synchronizing the close() method and declaring the closed field volatile.
Preserve the existing early-return guard so concurrent callers allow only one
shutdown sequence, including resources.teardownWorker() and WORKER_STOPPED
emission.
In `@sdks/java/src/test/java/org/byteveda/taskito/worker/ExecutorAttachTest.java`:
- Line 65: Remove the unused results field from ExecutorAttachTest, delete the
unreachable results check in the polling loop within readFrame-related logic,
and remove the now-unused ArrayDeque and Deque imports while preserving direct
frame handling.
🪄 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: e32abdfc-cc4f-4a80-bcfe-8c0a07073426
📒 Files selected for processing (49)
crates/taskito-core/src/lib.rscrates/taskito-core/src/worker/cancel.rscrates/taskito-core/src/worker/dial.rscrates/taskito-core/src/worker/executor.rscrates/taskito-core/src/worker/mod.rscrates/taskito-core/src/worker/protocol.rscrates/taskito-core/tests/rust.rscrates/taskito-core/tests/rust/executor_tests.rscrates/taskito-core/tests/rust/remote_tests.rscrates/taskito-core/tests/rust/worker_tests.rscrates/taskito-java/src/dispatcher.rscrates/taskito-java/src/executor.rscrates/taskito-java/src/lib.rscrates/taskito-node/src/dispatcher.rscrates/taskito-node/src/executor.rscrates/taskito-node/src/lib.rscrates/taskito-python/src/executor.rscrates/taskito-python/src/lib.rssdks/java/processor/src/main/java/org/byteveda/taskito/processor/TaskHandlerProcessor.javasdks/java/src/main/java/org/byteveda/taskito/cli/Cli.javasdks/java/src/main/java/org/byteveda/taskito/internal/JniExecutorControl.javasdks/java/src/main/java/org/byteveda/taskito/internal/MiddlewareDisables.javasdks/java/src/main/java/org/byteveda/taskito/internal/NativeExecutor.javasdks/java/src/main/java/org/byteveda/taskito/worker/Executor.javasdks/java/src/main/java/org/byteveda/taskito/worker/HandlerRegistryProvider.javasdks/java/src/main/java/org/byteveda/taskito/worker/WorkerDispatchBridge.javasdks/java/src/test/java/org/byteveda/taskito/worker/ExecutorAttachTest.javasdks/node/src/cli/commands/executor.tssdks/node/src/cli/commands/index.tssdks/node/src/cli/commands/run.tssdks/node/src/cli/index.tssdks/node/src/cli/load-app.tssdks/node/src/detached.tssdks/node/src/executor.tssdks/node/src/index.tssdks/node/src/native.tssdks/node/src/queue.tssdks/node/src/task-callback.tssdks/node/src/worker.tssdks/node/test/worker/executorAttach.test.tssdks/node/test/worker/executorAttachServer.test.tssdks/python/taskito/_taskito.pyisdks/python/taskito/app.pysdks/python/taskito/cli.pysdks/python/taskito/detached.pysdks/python/tests/worker/executor_apps/__init__.pysdks/python/tests/worker/executor_apps/attach_app.pysdks/python/tests/worker/test_executor_attach.pysdks/python/tests/worker/test_executor_attach_server.py
🔗 Linked repositories identified
CodeRabbit considers these linked repositories for cross-repo context during reviews:
ByteVeda/taskito(manual)
kartikeya-27
approved these changes
Aug 1, 2026
This was referenced Aug 1, 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
3 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.
Implements the three SDK executor subcommands — the S5–S7 phases of #546.
Closes #551
Closes #552
Closes #553
One client in core, three thin CLIs
tasks/executor-attach.mdsays "only the concrete stream changes", and that turned out to be literally true, so it sets the shape:WorkerDispatcherwas already the seam every SDK pool implements, and its signature —run(job_rx, result_tx)+shutdown()+notify_cancel()— is exactly what an executor needs.Worker::dispatcher()even documented the mirror case. SoExecutorClientisWorker::spawnwith storage swapped for a socket, and each SDK's subcommand is a wrapper over the pool it already had.Writing that client three times would have tripled the subtle parts — handshake ordering, exactly-once result accounting, drain sequencing, cancel routing — with nothing gained. The JSON-header wire is unchanged, so a third-party SDK can still write an executor with its standard library alone.
The executor opens no storage
An executor imports the user's app to find its handlers, and that import used to construct a
Queue— which connected eagerly, putting the database credentials straight back into the app image that #546 exists to keep them out of.TASKITO_DETACHED_EXECUTOR=1is now set by each CLI before it imports the app, and the native queue is replaced by a stand-in. Python's prefork children inherit the variable, so they open nothing either; Java already took noQueueBackend. One rule decides every method:None/nullis what a queue with no such row returns anyway, and callers already handle it. Not optional:Queue.__init__reads settings (webhook subscriptions), so a stand-in that refused them could not build aQueueat all.set_progressmust not fail because it happens to be running detached.What that costs — progress, task logs, published partials, dashboard middleware toggles — is tracked in #589 with per-layer sub-issues; closing it needs new protocol frames.
Four things the tests caught that reading the docs did not
provider()form only applies to modules on the module path. The first attempt generatedprovider()and died withServiceConfigurationError: demo.GreeterTasks not a subtype. A newHandlerRegistryProviderinterface is now the service, and the processor generates a nested<Class>Tasks$Providerimplementing it. Owners with no accessible no-arg constructor are skipped with a compiler note rather than silently — only their own code knows how to build them.startExecutorblocked the Node event loop for the whole handshake timeout. Now async. Found because the in-process test deadlocked against its own fake scheduler.stop()hung forever.stop()cannot unpark the frame reader, which is parked on a read only the scheduler could satisfy, so the session never ended and any shell that stopped and then waited deadlocked. Fixed in core with a regression test.AbortSignala cooperative task watches is driven by a storage poll.JsExecutor.isCancelRequestednow exposes the state the cancel frames land in. The test asserts the result frame comes backcancelled, notfailure— proving both that the handler aborted and that the native side reclassified the throw.Verification
cargo test --workspaceclean,clippy --all-targets -D warningsclean,--features postgresand--features redischecktaskito-serverbinary; ruff + mypy cleanMETA-INF/services, with no usermainEach SDK has two test layers: a hermetic scheduler speaking the wire (so they run with no Rust build, in every CI job) and the same assertions against the real
taskito-serverbehindTASKITO_SERVER_BIN. That pairing is what keeps the fakes honest.Two environment notes, both reproduced on a clean tree and unrelated to this branch: Java's
:compileTestJavais skipped here, so its tests were run through the JUnit platform launcher directly; and 10test/dashboard/**Node files fail on a missing SPA build.Reviewing
The first three commits (
feat(server): authenticate the attach listener handshakeand the two storage fixes) are the #584 review follow-ups this branch was cut from, not part of this work — they are unmerged and unreviewed, so say the word if you would rather they went up as their own PR and this one rebased onto master.The remaining 25 split cleanly by layer: core (7), Python (5), Node (7), Java (6). S5/S6/S7 were meant to be separate PRs, so this can be split per SDK if you prefer that shape.
Summary by CodeRabbit
New Features
Bug Fixes