fix(plugin): bound proposal queue deadlines - #1180
Conversation
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (1)
🚧 Files skipped from review as they are similar to previous changes (1)
📝 WalkthroughWalkthroughThe PR replaces native proposal dispatch with deadline-aware active and passive workers. It adds structured source telemetry, propagates it through Skippy query outcomes, emits privacy-safe attributes, and tests deadline recovery and late candidates. ChangesProposal deadline and telemetry flow
Estimated code review effort: 4 (Complex) | ~60 minutes Sequence Diagram(s)sequenceDiagram
participant LinearProposalIngress
participant PluginDriver
participant ActiveWorker
participant PassiveWorker
participant LinearProposalQuery
LinearProposalIngress->>PluginDriver: submit proposal query with deadline
PluginDriver->>ActiveWorker: enqueue proposal command
ActiveWorker->>PluginDriver: return proposal and source telemetry
PluginDriver->>PassiveWorker: enqueue report or discard callback
PluginDriver->>LinearProposalIngress: return LinearProposalSourceResponse
LinearProposalIngress->>LinearProposalQuery: preserve source telemetry
Possibly related PRs
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches 💡 1📝 Generate docstrings 💡
🧪 Generate unit tests (beta)
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 |
|
This pull request is currently a draft. Reviews will not take place until the PR is marked as ready for review. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 2
🧹 Nitpick comments (3)
crates/skippy-server/src/frontend/local_generation/linear_decode.rs (1)
187-199: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winThe source event is emitted on every decode step.
emit_linear_proposal_source_telemetryruns for everyNoProposal,DeadlineExceeded, andReadyoutcome.try_execute_linear_proposalruns once per decode step, so this emits one event per generated token per request. The comparable per-step proposal event at Lines 175-180 usesemit_debugand is gated bystate.emit_token_debug. The existingstage.openai_linear_proposal_lateevent usesemit, but it fires only on the rare late path.Consider gating the routine outcomes behind
emit_debug, and keeping unconditionalemitfor the deadline and queue-pressure outcomes only.🤖 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/skippy-server/src/frontend/local_generation/linear_decode.rs` around lines 187 - 199, Update emit_linear_proposal_source_telemetry so routine NoProposal and Ready outcomes use emit_debug gated by state.emit_token_debug, while DeadlineExceeded and queue-pressure outcomes continue using unconditional emit. Preserve the existing telemetry attributes and event name, and adjust try_execute_linear_proposal callers as needed to distinguish these outcome classes.crates/mesh-native-serving-plugin-host/src/lib.rs (2)
604-617: 🩺 Stability & Availability | 🔵 Trivial | ⚡ Quick winPassive commands can be deferred without bound.
pop_nextreturns a queued proposal ahead of everyReportandDiscardcommand. If proposals arrive continuously from several decode threads, the passive commands are never selected. They accumulate until the queue reachesPLUGIN_COMMAND_CAPACITY. After that,try_enqueuereportsFull, so proposals returnQueueFulland lifecycle delivery fails.Bound the deferral. One option is to select the oldest passive command when its
enqueued_atage exceeds a threshold. A second option is to select a passive command whenever the queue occupancy exceeds a watermark.🤖 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/mesh-native-serving-plugin-host/src/lib.rs` around lines 604 - 617, Update pop_next so passive commands (Report/Discard) cannot be deferred indefinitely behind proposals: select an eligible passive command once its enqueued_at age exceeds the chosen threshold or queue occupancy reaches a defined watermark, while preserving lifecycle-command priority and normal proposal ordering otherwise.
798-808: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winA plugin error is reported as
Abstained.The
Err(_)arm maps a failed plugin callback toLinearProposalSourceOutcome::Abstained. Telemetry then cannot separate a deliberate abstention from a callback failure.LinearProposalSourceOutcomeis#[non_exhaustive], so adding a dedicated variant is additive and does not break older consumers.Note that this telemetry value is also unreachable today for the error case:
NativeProposalIngress::proposereturnsErrat Line 489, andquery_linear_proposalreturns early without callingtake_proposal_telemetry. The stored value then leaks into the next query. That leak is covered by the comment on the trait contract incrates/skippy-server/src/frontend/linear_proposal.rs.Add a
SourceErrorvariant, or document thatAbstainedalso covers callback 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/mesh-native-serving-plugin-host/src/lib.rs` around lines 798 - 808, Distinguish plugin callback failures from deliberate abstentions in the outcome mapping around the result match: add the additive non-exhaustive LinearProposalSourceOutcome::SourceError variant and map Err(_) to it. Ensure the error path still records or consumes telemetry before NativeProposalIngress::propose/query_linear_proposal returns, preventing the failed result from leaking into the next query.
🤖 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/mesh-native-serving-plugin-host/src/lib.rs`:
- Around line 559-586: Add a shared stopped-state flag to PluginCommandQueue,
have plugin_worker set it before normal return and via a Drop guard during
unwinding, and update try_enqueue to reject commands once the worker has
stopped. Preserve the existing Full and Poisoned error handling while ensuring
enqueue reports the stopped-worker failure so lifecycle delivery does not
succeed without a running callback worker.
In `@crates/skippy-server/src/frontend/linear_proposal.rs`:
- Around line 290-299: Remove the calling-thread assumption from
take_proposal_telemetry by carrying telemetry with the individual proposal
result through propose into query_linear_proposal. Ensure failed propose calls
also return or clear telemetry so stale values cannot be consumed by a later
successful query, and update the method documentation to match the new
request-scoped behavior.
---
Nitpick comments:
In `@crates/mesh-native-serving-plugin-host/src/lib.rs`:
- Around line 604-617: Update pop_next so passive commands (Report/Discard)
cannot be deferred indefinitely behind proposals: select an eligible passive
command once its enqueued_at age exceeds the chosen threshold or queue occupancy
reaches a defined watermark, while preserving lifecycle-command priority and
normal proposal ordering otherwise.
- Around line 798-808: Distinguish plugin callback failures from deliberate
abstentions in the outcome mapping around the result match: add the additive
non-exhaustive LinearProposalSourceOutcome::SourceError variant and map Err(_)
to it. Ensure the error path still records or consumes telemetry before
NativeProposalIngress::propose/query_linear_proposal returns, preventing the
failed result from leaking into the next query.
In `@crates/skippy-server/src/frontend/local_generation/linear_decode.rs`:
- Around line 187-199: Update emit_linear_proposal_source_telemetry so routine
NoProposal and Ready outcomes use emit_debug gated by state.emit_token_debug,
while DeadlineExceeded and queue-pressure outcomes continue using unconditional
emit. Preserve the existing telemetry attributes and event name, and adjust
try_execute_linear_proposal callers as needed to distinguish these outcome
classes.
🪄 Autofix
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: cd439b5b-e53f-453c-a7f7-9aa92320fd41
📒 Files selected for processing (6)
crates/mesh-llm-host-runtime/src/runtime/survey.rscrates/mesh-native-serving-plugin-host/src/lib.rscrates/skippy-server/src/frontend.rscrates/skippy-server/src/frontend/linear_proposal.rscrates/skippy-server/src/frontend/local_generation/linear_decode.rsdocs/plugins/telemetry.md
c67f73e to
1ebeea1
Compare
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
crates/mesh-native-serving-plugin-host/src/lib.rs (1)
840-867: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winUse one completion timestamp for late-candidate handling.
Lines 840 and 843 read
Instant::now()independently. If the first read is before the deadline and the second is after it, telemetry reportsCandidateReturnedTooLate, butcandidate_was_lateremains false and Lines 859-867 forward the candidate. Capture one callback completion timestamp and derive both the outcome and proposal suppression from it.Proposed fix
- let candidate_was_late = - matches!(&result, Ok(Some(_))) && Instant::now() >= query.deadline; + let callback_finished_at = Instant::now(); + let candidate_was_late = + matches!(&result, Ok(Some(_))) && callback_finished_at >= query.deadline; let outcome = match &result { - Ok(Some(_)) if Instant::now() >= query.deadline => { + Ok(Some(_)) if candidate_was_late => {🤖 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/mesh-native-serving-plugin-host/src/lib.rs` around lines 840 - 867, Capture a single callback completion timestamp before the late-candidate checks, then derive both candidate_was_late and the outcome match from that shared timestamp. Use candidate_was_late to preserve suppression of late proposals in ProposalResponse and the existing discard behavior, ensuring telemetry classification and proposal forwarding cannot disagree.
🤖 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/mesh-native-serving-plugin-host/src/lib.rs`:
- Around line 525-529: Extract the deadline dispatch subsystem spanning
PluginCommandQueue and its associated queue, driver, worker, and shutdown logic
into a private, responsibility-named plugin_dispatch module. Move all related
symbols and implementation together, update internal references and imports, and
keep the module private without adding a crate-root re-export.
- Around line 780-790: Update PluginDriver::stop_worker to close each command
queue atomically with its Shutdown command, retrying or waiting when enqueue
fails because the queue is full. After receiving an acknowledged shutdown or
observing queue.stopped, always take and join the worker handle without checking
JoinHandle::is_finished, ensuring the worker fully exits before cleanup
continues.
- Around line 853-858: Ensure the late-candidate discard in the candidate
handling path is never lost when passive_queue.enqueue rejects it. Update the
logic around candidate_was_late and PluginCommand::Discard to reserve capacity
or retain and retry the terminal command until the passive worker accepts it,
and handle the enqueue result explicitly rather than ignoring it.
---
Outside diff comments:
In `@crates/mesh-native-serving-plugin-host/src/lib.rs`:
- Around line 840-867: Capture a single callback completion timestamp before the
late-candidate checks, then derive both candidate_was_late and the outcome match
from that shared timestamp. Use candidate_was_late to preserve suppression of
late proposals in ProposalResponse and the existing discard behavior, ensuring
telemetry classification and proposal forwarding cannot disagree.
🪄 Autofix
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: b769829c-f6e7-4eb9-a2f3-fd44c5bbc834
📒 Files selected for processing (7)
crates/mesh-native-serving-plugin-host/src/lib.rscrates/skippy-server/src/frontend.rscrates/skippy-server/src/frontend/linear_proposal.rscrates/skippy-server/src/frontend/local_generation/linear_decode.rscrates/skippy-server/src/lib.rscrates/skippy-server/src/serving_hooks.rsdocs/plugins/telemetry.md
🚧 Files skipped from review as they are similar to previous changes (3)
- docs/plugins/telemetry.md
- crates/skippy-server/src/frontend.rs
- crates/skippy-server/src/frontend/local_generation/linear_decode.rs
| struct PluginCommandQueue { | ||
| commands: Mutex<VecDeque<QueuedPluginCommand>>, | ||
| stopped: AtomicBool, | ||
| available: Condvar, | ||
| } |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy lift
Extract the deadline dispatch subsystem.
Lines 525-925 add a separable queue, driver, worker, and shutdown subsystem to a file that now reaches line 1,613. Move this subsystem to a private, responsibility-named module such as plugin_dispatch. Do not add a crate-root re-export.
As per coding guidelines, “When modifying a Rust source file over 1,000 lines, extract any separable responsibility into a semantically named module.”
🤖 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/mesh-native-serving-plugin-host/src/lib.rs` around lines 525 - 529,
Extract the deadline dispatch subsystem spanning PluginCommandQueue and its
associated queue, driver, worker, and shutdown logic into a private,
responsibility-named plugin_dispatch module. Move all related symbols and
implementation together, update internal references and imports, and keep the
module private without adding a crate-root re-export.
Source: Coding guidelines
| fn stop_worker(&self, queue: &Arc<PluginCommandQueue>, worker: &Mutex<Option<JoinHandle<()>>>) { | ||
| let (reply, response) = sync_channel(1); | ||
| let queued = self.sender.try_send(PluginCommand::Shutdown(reply)).is_ok(); | ||
| if queued | ||
| && response.recv_timeout(CLEAN_SHUTDOWN_TIMEOUT).is_ok() | ||
| && let Ok(worker) = self.worker.get_mut() | ||
| let clean_shutdown = queue.enqueue(PluginCommand::Shutdown(reply)).is_ok() | ||
| && response.recv_timeout(CLEAN_SHUTDOWN_TIMEOUT).is_ok(); | ||
| if (clean_shutdown || queue.stopped.load(Ordering::Acquire)) | ||
| && let Ok(mut worker) = worker.lock() | ||
| && worker.as_ref().is_some_and(JoinHandle::is_finished) | ||
| && let Some(worker) = worker.take() | ||
| { | ||
| let _ = worker.join(); | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | 🏗️ Heavy lift
Make worker shutdown and joining reliable.
A full queue makes queue.enqueue(Shutdown) fail, so that worker never receives a terminal command. After a successful shutdown reply, plugin_worker sends the reply before it breaks, so JoinHandle::is_finished() can still be false and this code also skips join. Either path leaves a worker holding Arc<ActivePlugin>, causes Arc::get_mut at Line 771 to fail, and skips the native plugin shutdown callback.
Close each queue atomically with its shutdown command, retry or wait when capacity is full, and join an acknowledged or stopped worker without an is_finished gate.
🤖 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/mesh-native-serving-plugin-host/src/lib.rs` around lines 780 - 790,
Update PluginDriver::stop_worker to close each command queue atomically with its
Shutdown command, retrying or waiting when enqueue fails because the queue is
full. After receiving an acknowledged shutdown or observing queue.stopped,
always take and join the worker handle without checking JoinHandle::is_finished,
ensuring the worker fully exits before cleanup continues.
| if candidate_was_late && let Ok(Some(proposal)) = &result { | ||
| let _ = passive_queue.enqueue(PluginCommand::Discard( | ||
| proposal.decision_id.as_bytes().to_vec(), | ||
| LinearProposalDiscardReason::DeadlineExceeded, | ||
| )); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy lift
Do not drop the terminal discard for a late candidate.
If the passive queue is full or stopped, passive_queue.enqueue(...) returns an error that this code discards. The late candidate is withheld from decode, so downstream cannot send a receipt or discard. The plugin then receives no terminal disposition for that decision ID.
Reserve capacity for deadline discards, or retain and retry terminal commands until the passive worker accepts them. Do not ignore this enqueue 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/mesh-native-serving-plugin-host/src/lib.rs` around lines 853 - 858,
Ensure the late-candidate discard in the candidate handling path is never lost
when passive_queue.enqueue rejects it. Update the logic around
candidate_was_late and PluginCommand::Discard to reserve capacity or retain and
retry the terminal command until the passive worker accepts it, and handle the
enqueue result explicitly rather than ignoring it.
|
Note GitHub couldn't provide a complete incremental comparison for this pull request, so CodeRabbit is performing a full review instead. This review may take a little longer. |
| let candidate_was_late = | ||
| matches!(&result, Ok(Some(_))) && Instant::now() >= query.deadline; | ||
| let outcome = match &result { | ||
| Ok(Some(_)) if Instant::now() >= query.deadline => { |
There was a problem hiding this comment.
Double Instant::now() in late-candidate path can disagree with telemetry.
Lines 891 and 893 each read Instant::now() independently. Between the two reads, time advances — it is possible for line 891 to see pre-deadline (so candidate_was_late = false) while line 893 sees post-deadline and classifies telemetry as CandidateReturnedTooLate. The proposal then gets forwarded (!candidate_was_late at line 908) but telemetry reports it was late.
Fix: Capture a single timestamp once:
let callback_finished_at = Instant::now();
let candidate_was_late = matches!(&result, Ok(Some(_))) && callback_finished_at >= query.deadline;
let outcome = match &result {
Ok(Some(_)) if candidate_was_late => LinearProposalSourceOutcome::CandidateReturnedTooLate,Also note line 897 has a third Instant::now() for the Ok(None) deadline case — same window.
| Err(_) => LinearProposalSourceOutcome::SourceError, | ||
| }; | ||
| if candidate_was_late && let Ok(Some(proposal)) = &result { | ||
| let _ = passive_queue.enqueue(PluginCommand::Discard( |
There was a problem hiding this comment.
'Passive Discard for late candidate is silently dropped when passive queue full.
The result of passive_queue.enqueue(...) on line 904 is let _ = ..., so if the passive queue hits capacity, the discard for a late candidate is silently lost. The proposal itself isn't forwarded (good), but its activation-frame resources are never cleaned up — no downstream report that this slot should be discarded. This leaks state in the decode pipeline until eviction happens naturally.
Consider either:
- Making
Discardterminal commands un-droppable (e.g., reserve one capacity slot, or retry with backoff) - At minimum logging a warning when it fails'
| self.driver.propose(query) | ||
| fn propose(&self, query: LinearProposalQuery) -> Result<LinearProposalSourceResponse> { | ||
| let response = self.driver.propose(query)?; | ||
| let proposal = response.proposal.unwrap_or_default(); |
There was a problem hiding this comment.
'Error from native plugin callback is swallowed by .unwrap_or_default()
The response.proposal field in NativeProposalIngress::propose is a Result<Option<LinearProposal>, String>. When the plugin callback returns an error, this becomes Err(String) — and .unwrap_or_default() replaces it with Ok(None) (i.e., "abstained"). The original error message from the native plugin is lost.
The ingress adapter then converts the outer Result to a different type before returning up through skippy-server into linear_decode.rs where errors become fatal bailouts at line 168 (expect("linear proposal failed")). So this silently turns "plugin crashed" into "abstained", which means decode continues without ever knowing something went wrong.
Suggestion: Propagate the error or map it to a distinct telemetry outcome rather than defaulting to Ok(None).'
| fn next(&self) -> QueuedPluginCommand { | ||
| let mut commands = self | ||
| .commands | ||
| .lock() |
There was a problem hiding this comment.
'PluginCommandQueue::next() blocks indefinitely on unbounded condvar wait
queue.next() at line 634 uses an infinite loop with no timeout on the condvar wait. If the queue is stopped (mark_stopped) but a thread already entered wait(), it will block forever — there's no corresponding notify_all() in the stop path that would wake threads currently blocked inside next().
Looking at line 628, mark_stopped does call self.available.notify_all() after setting the flag. But if a thread is between checking pop_next (which returns None) and calling .wait(), it will enter an infinite loop:
- Thread checks stopped → false
- Pop next → None
- Calls wait(blocked here while stop happens + notifies)
- Wait wakes, but no new command was added...
Actually mark_stopped does notify_all so threads wake up — they'll see the queue is empty and loop back to wait again indefinitely unless a Shutdown or more commands arrive before shutdown completes.
Consider: Add a timeout-based wait (wait_timeout) that re-checks stopped.load() each iteration, breaking when stopped instead of blocking forever.'
|
Left inline comments on four items not covered by CodeRabbit. Summary below (each linked to the threaded discussion): Functional correctness: 🟠 Double
|
2 participants
Summary
Root cause
The native serving-plugin driver used a single FIFO for lifecycle callbacks, proposal queries, and passive proposal outcomes. A callback ahead of a proposal could consume its deadline before the plugin was reached.
Validation
cargo test -p mesh-native-serving-plugin-host --libcargo test -p skippy-server --libcargo test -p mesh-llm-host-runtime runtime::survey::tests --libcargo check -p mesh-llmcargo clippy -p mesh-native-serving-plugin-host -p skippy-server --all-targets -- -D warningsjust buildcargo clippy -p mesh-llm --all-targets -- -D warningsremains blocked by pre-existing unfulfilleddead_codelint expectations inmesh-llm-host-runtime.Closes #1178
Summary by CodeRabbit
New Features
Bug Fixes
Documentation