feat(python): report executor progress, logs and toggles over the side-channel - #600
Conversation
The scheduler drops a frame naming a job it no longer holds, so a result that overtook them lost the task's final progress and last log lines.
|
Warning Review limit reached
Next review available in: 24 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 (6)
📝 WalkthroughWalkthroughThe executor now forwards progress and task-log frames before job results. Prefork workers relay these frames through a scheduler side channel, propagate disabled middleware settings, and preserve fallback behavior when side-channel support is unavailable. ChangesExecutor result ordering
Detached worker forwarding
Prefork executor wiring
Protocol and integration validation
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant Task
participant PreforkChild
participant PreforkPool
participant Scheduler
Task->>PreforkChild: report progress or task log
PreforkChild->>PreforkPool: send side-channel frame
PreforkPool->>Scheduler: relay progress or task log
PreforkChild->>PreforkPool: send job result
PreforkPool->>Scheduler: relay 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: 1
🧹 Nitpick comments (2)
crates/taskito-python/src/prefork/mod.rs (1)
309-317: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueClone the handle out of the slot before calling into it.
dispatch_jobholds theside_channelmutex while it callsrelay.disabled_middleware(&job.id), which takes the core executor's ownsharedlock.relay_side_channelat Line 461 does the opposite: it clones the handle, releases the slot guard, then calls the relay. Aligningdispatch_jobwith that pattern shortens the critical section and keeps one lock-ordering rule in this module.♻️ Proposed refactor
- let disabled = side_channel - .lock() - .unwrap_or_else(|poisoned| poisoned.into_inner()) - .as_ref() - .map(|relay| relay.disabled_middleware(&job.id)) - .unwrap_or_default(); + let relay = side_channel + .lock() + .unwrap_or_else(|poisoned| poisoned.into_inner()) + .clone(); + let disabled = relay + .map(|relay| relay.disabled_middleware(&job.id)) + .unwrap_or_default();🤖 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/mod.rs` around lines 309 - 317, Update dispatch_job’s disabled-middleware lookup to clone the relay handle from the side_channel slot while holding the mutex, release the guard, and only then call disabled_middleware(&job.id). Match the existing relay_side_channel pattern at its visible implementation, preserving the default behavior when no relay is configured.sdks/python/tests/worker/test_executor_attach_server.py (1)
239-251: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winBound the inner waits by the remaining outer budget.
The retry loop uses
SETTLEas its own deadline, and each iteration callswait_for_status(..., timeout=SETTLE)andresult(timeout=SETTLE). A single slow iteration can therefore consume the whole outer budget on its own. Two consequences follow. The test can overrun its intendedSETTLEbound by a multiple. The loop can also get only one attempt, which defeats the purpose of retrying until the scheduler's dispatch cache expires.Passing the remaining time into the inner waits keeps the total bounded and guarantees several attempts.
♻️ Proposed refactor
deadline = time.monotonic() + SETTLE while time.monotonic() < deadline: + remaining = max(deadline - time.monotonic(), 0.1) job_id = enqueue(db_path, MIDDLEWARED) - wait_for_status(db_path, job_id, "complete") + wait_for_status(db_path, job_id, "complete", timeout=remaining) toggled = queue.get_job(job_id) assert toggled is not None - if toggled.result(timeout=SETTLE) == "": + if toggled.result(timeout=remaining) == "": return time.sleep(0.5)🤖 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/python/tests/worker/test_executor_attach_server.py` around lines 239 - 251, Update the retry loop after disable_middleware_for_task so each wait uses the remaining outer deadline rather than the full SETTLE value: compute the remaining budget before enqueue/status polling and before toggled.result, pass it to wait_for_status and result, and stop or fail when no budget remains. Preserve the retry behavior and ensure the total loop duration stays within SETTLE.
🤖 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/python/taskito/prefork/child.py`:
- Around line 119-124: Update the exception tuple in the static _send method to
catch OSError instead of BrokenPipeError, while preserving the existing
EOFError, ValueError, and ProtocolError handling and debug logging behavior.
---
Nitpick comments:
In `@crates/taskito-python/src/prefork/mod.rs`:
- Around line 309-317: Update dispatch_job’s disabled-middleware lookup to clone
the relay handle from the side_channel slot while holding the mutex, release the
guard, and only then call disabled_middleware(&job.id). Match the existing
relay_side_channel pattern at its visible implementation, preserving the default
behavior when no relay is configured.
In `@sdks/python/tests/worker/test_executor_attach_server.py`:
- Around line 239-251: Update the retry loop after disable_middleware_for_task
so each wait uses the remaining outer deadline rather than the full SETTLE
value: compute the remaining budget before enqueue/status polling and before
toggled.result, pass it to wait_for_status and result, and stop or fail when no
budget remains. Preserve the retry behavior and ensure the total loop duration
stays within SETTLE.
🪄 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: 66543270-228e-494e-9a11-b6c0c5d011bc
📒 Files selected for processing (11)
crates/taskito-core/src/worker/executor.rscrates/taskito-core/tests/rust/executor_tests.rscrates/taskito-python/src/executor.rscrates/taskito-python/src/prefork/mod.rssdks/python/taskito/detached.pysdks/python/taskito/mixins/decorators.pysdks/python/taskito/prefork/child.pysdks/python/taskito/worker_protocol.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)
|
All three findings were valid and all three are fixed, plus the CI failure and a real bug the review prompted me to go looking for.
Toggle retry. Each inner wait gets what is left of the outer deadline rather than a fresh CI. The progress bugRunning
Covered by Verification
|
2 participants
Part 2 of #589, on top of #599. The core protocol and the scheduler's half landed there; this wires the Python SDK to it, so a task running on an attached executor gets its progress bar, its task logs, its published partials and its dashboard middleware toggles back. Node and Java follow in their own PRs.
What changes
taskito/detached.py— where an executor's storage-shaped writes goDetachedNativepreviously warned once and droppedupdate_progressandwrite_task_log. It now forwards both to an installedExecutorSink, aProtocolwith exactly those two methods. Both are fire-and-forget: a task reporting progress must not be able to fail, or block, because of what is happening at the far end. With no sink installed — an app imported outsidetaskito executor, or a scheduler that advertised noside_channelcapability — the old warn-once degradation is unchanged.The module also holds the toggle list the scheduler attached to the current dispatch (
set_disabled_middleware/disabled_middleware). A prefork child runs one job at a time, so a single value is the whole story.taskito/prefork/child.py— the child's halfA
_ParentSinkframes progress and task logs to the parent over the existing child protocol, and installs itself once the child is running under an executor.crates/taskito-python/src/prefork/— the parent's halfPreforkPoolgains aset_side_channel, called once the attach completes — the earliest the handle exists. Reader threads start before that, so the slot is anArc<Mutex<Option<_>>>rather than owned state. Frames from a child are relayed to the scheduler; until the slot is filled, and always for an in-process worker whose children hold real storage and write for themselves, they are dropped.dispatch_jobreads the toggle list off the side channel and carries it on to the child viawrite_job_with, since the task body — and so the middleware chain — runs there.taskito/mixins/decorators.py— honouring the togglesThe middleware chain resolver takes a different route when detached: it has no settings store to read, so it uses the list the dispatch carried. That is per job rather than per task name, so it also bypasses the
_MW_CHAIN_TTLcache — there is nothing to save when the list arrived with the work. The disable-filtering logic moved to a shared_chain_withoutso both routes key onmiddleware_keyidentically.taskito/worker_protocol.pydeclared_payload_lenunderstandstask_log'sextra_len, so a published partial's blob is read off the wire rather than desyncing the stream.One core fix,
a6c8ba0fix(core): send queued side-channel work before a result. The executor's result loop drained results to empty before considering anything else. The scheduler drops a side-channel frame naming a job it no longer holds, so a result that overtook the queues silently lost the task's final progress and its last log lines — the two it is most likely to care about. Results now flush the side-channel queues first, through a sharedflush_side_channelthat the teardown tail reuses. Bounded work: both queues are capped and each entry is one small frame on an already-open socket.Covered by
a_result_never_overtakes_what_the_task_already_reported.Tests
test_progress_and_logs_reach_a_scheduler_that_advertised_the_side_channeltest_a_middleware_disabled_on_the_dispatch_does_not_runtest_progress_and_logs_reach_storage_through_a_real_schedulertest_a_dashboard_toggle_reaches_an_attached_executorThe last two run against a real standalone scheduler and assert on the rows that land in storage, not on the frames.
cargo test --workspace,cargo clippy --workspace --all-targets,cargo fmt --check,cargo checkunder--features postgresand--features redis,ruff check,mypy(194 files) and the Python suite (1379 passed, 14 skipped) are all clean.Refs #589.
Summary by CodeRabbit
New Features
Bug Fixes