diff --git a/crates/mesh-native-serving-plugin-host/Cargo.toml b/crates/mesh-native-serving-plugin-host/Cargo.toml index 7b74c9c62..b42f7c422 100644 --- a/crates/mesh-native-serving-plugin-host/Cargo.toml +++ b/crates/mesh-native-serving-plugin-host/Cargo.toml @@ -16,3 +16,6 @@ skippy-server = { path = "../skippy-server", version = "0.72.1" } [lints] workspace = true + +[dev-dependencies] +skippy-server = { path = "../skippy-server", version = "0.72.1", features = ["test-support"] } diff --git a/crates/mesh-native-serving-plugin-host/src/lib.rs b/crates/mesh-native-serving-plugin-host/src/lib.rs index 4dabeb67a..16c7f695e 100644 --- a/crates/mesh-native-serving-plugin-host/src/lib.rs +++ b/crates/mesh-native-serving-plugin-host/src/lib.rs @@ -428,10 +428,12 @@ impl ActivePlugin { }; let status = unsafe { (self.definition.api().finish_generation)(self.instance()?, &event) }; let result = self.call_status("finish generation", status); - self.committed_generated_tokens - .lock() - .map_err(|_| anyhow!("native serving plugin commit state lock poisoned"))? - .remove(&key); + if result.is_ok() { + self.committed_generated_tokens + .lock() + .map_err(|_| anyhow!("native serving plugin commit state lock poisoned"))? + .remove(&key); + } result } @@ -713,7 +715,8 @@ impl LinearProposalIngress for NativeProposalIngress { } fn report(&self, receipt: &LinearProposalReceipt) -> Result<()> { - self.driver.enqueue(PluginCommand::Report(receipt.clone())) + self.driver + .enqueue_terminal(PluginCommand::Report(receipt.clone())) } fn discard( @@ -721,7 +724,7 @@ impl LinearProposalIngress for NativeProposalIngress { decision_id: &OpaqueProposalDecisionId, reason: LinearProposalDiscardReason, ) -> Result<()> { - self.driver.enqueue(PluginCommand::Discard( + self.driver.enqueue_terminal(PluginCommand::Discard( decision_id.as_bytes().to_vec(), reason, )) diff --git a/crates/mesh-native-serving-plugin-host/src/plugin_dispatch.rs b/crates/mesh-native-serving-plugin-host/src/plugin_dispatch.rs index 8913ca655..c78acd705 100644 --- a/crates/mesh-native-serving-plugin-host/src/plugin_dispatch.rs +++ b/crates/mesh-native-serving-plugin-host/src/plugin_dispatch.rs @@ -41,6 +41,10 @@ pub(crate) enum PluginCommand { Proposal(LinearProposalQuery, SyncSender), Report(LinearProposalReceipt), Discard(Vec, LinearProposalDiscardReason), + /// A passive-queue barrier. The primary worker waits for this barrier + /// before finalizing a generation, so all earlier reports/discards have + /// completed before the plugin sees its lifecycle finish callback. + Fence(SyncSender<()>), } pub(crate) struct ProposalResponse { @@ -104,11 +108,19 @@ impl PluginCommandQueue { /// A late candidate is withheld from decode, so its `discard` is the only /// remaining way the plugin can resolve that decision ID. It must not be /// dropped just because ordinary traffic filled the queue. + /// + /// One slot in the terminal reserve is held exclusively for `Fence`, so + /// finish can always be ordered after earlier dispositions. fn try_enqueue_terminal( &self, command: PluginCommand, ) -> std::result::Result<(), PluginCommandQueueError> { - self.enqueue_within(command, PLUGIN_COMMAND_CAPACITY + PLUGIN_TERMINAL_RESERVE) + let capacity = if matches!(command, PluginCommand::Fence(_)) { + PLUGIN_COMMAND_CAPACITY + PLUGIN_TERMINAL_RESERVE + } else { + PLUGIN_COMMAND_CAPACITY + PLUGIN_TERMINAL_RESERVE - 1 + }; + self.enqueue_within(command, capacity) } fn enqueue_within( @@ -311,10 +323,30 @@ impl PluginDriver { self.queue_for(&command).enqueue(command) } + /// Enqueues a terminal proposal disposition using the reserved queue + /// capacity. Every report/discard must be deliverable before finalization + /// can be fenced behind it. + pub(crate) fn enqueue_terminal(&self, command: PluginCommand) -> Result<()> { + self.ensure_healthy()?; + self.queue_for(&command) + .try_enqueue_terminal(command) + .map_err(|error| match error { + PluginCommandQueueError::Full => { + anyhow!("native serving plugin terminal queue is full") + } + PluginCommandQueueError::Stopped => { + anyhow!("native serving plugin passive worker stopped") + } + PluginCommandQueueError::Poisoned => { + anyhow!("native serving plugin terminal queue lock poisoned") + } + }) + } + fn queue_for(&self, command: &PluginCommand) -> &Arc { if matches!( command, - PluginCommand::Report(_) | PluginCommand::Discard(_, _) + PluginCommand::Report(_) | PluginCommand::Discard(_, _) | PluginCommand::Fence(_) ) { &self.passive_queue } else { @@ -336,6 +368,7 @@ impl PluginDriver { LinearProposalSourceOutcome::HostDeadlineExceeded, )); } + let (reply, response) = sync_channel(1); match self .queue @@ -440,12 +473,15 @@ fn plugin_worker( PluginCommand::Begin(event) => (active.begin(&event), true), PluginCommand::Committed(event) => (active.committed(&event), true), PluginCommand::Abort(event) => (active.abort(&event), true), - PluginCommand::Finish(event) => (active.finish(&event), true), + PluginCommand::Finish(event) => ( + finish_after_passive_fence(&passive_queue, || active.finish(&event)), + true, + ), PluginCommand::Proposal(query, reply) => { run_proposal(&active, &passive_queue, enqueued_at, query, &reply); continue; } - PluginCommand::Report(_) | PluginCommand::Discard(_, _) => { + PluginCommand::Report(_) | PluginCommand::Discard(_, _) | PluginCommand::Fence(_) => { unreachable!("passive plugin callbacks must use the passive worker queue") } }; @@ -472,6 +508,32 @@ fn run_proposal( return; } + // Reports and discards run on the passive worker so they cannot consume + // proposal callback time. The primary worker owns this fence, which makes + // proposal ordering serial even when multiple callers enqueue proposals + // concurrently. Unlike the caller-side deadline wait, this completion + // wait is never abandoned: after it completes we recheck the original + // absolute deadline before invoking the plugin. + if let Err(error) = fence_passive(passive_queue) { + eprintln!("native serving plugin proposal fence failed: {error:#}"); + let _ = reply.send(ProposalResponse { + proposal: Err(format!("{error:#}")), + telemetry: LinearProposalSourceTelemetry { + queue_wait_us, + callback_elapsed_us: 0, + outcome: LinearProposalSourceOutcome::SourceError, + }, + }); + return; + } + if Instant::now() >= deadline { + let _ = reply.send(abstention( + queue_wait_us, + LinearProposalSourceOutcome::DeadlineExceededBeforeDispatch, + )); + return; + } + let callback_started = Instant::now(); let result = active.propose(query); // One timestamp classifies both the forwarding decision and the telemetry, @@ -528,6 +590,39 @@ fn discard_late_candidate(passive_queue: &PluginCommandQueue, proposal: &LinearP } } +/// Waits until every passive terminal callback queued before this point has +/// completed. This preserves the causal order between proposal outcomes and +/// the lifecycle finish callback without putting slow passive callbacks on the +/// proposal-deadline queue. +fn fence_passive(passive_queue: &PluginCommandQueue) -> Result<()> { + let (reply, response) = sync_channel(1); + passive_queue + .try_enqueue_terminal(PluginCommand::Fence(reply)) + .map_err(|error| match error { + PluginCommandQueueError::Full => { + anyhow!("native serving plugin terminal queue is full") + } + PluginCommandQueueError::Stopped => { + anyhow!("native serving plugin passive worker stopped") + } + PluginCommandQueueError::Poisoned => { + anyhow!("native serving plugin terminal queue lock poisoned") + } + })?; + response + .recv() + .map_err(|_| anyhow!("native serving plugin passive worker stopped before fence"))?; + Ok(()) +} + +fn finish_after_passive_fence( + passive_queue: &PluginCommandQueue, + finish: impl FnOnce() -> Result, +) -> Result { + fence_passive(passive_queue)?; + finish() +} + fn plugin_passive_worker( active: Arc, queue: Arc, @@ -545,6 +640,9 @@ fn plugin_passive_worker( PluginCommand::Discard(decision_id, reason) => { let _ = active.discard(&decision_id, reason); } + PluginCommand::Fence(reply) => { + let _ = reply.send(()); + } PluginCommand::Begin(_) | PluginCommand::Committed(_) | PluginCommand::Abort(_) diff --git a/crates/mesh-native-serving-plugin-host/src/plugin_dispatch/tests.rs b/crates/mesh-native-serving-plugin-host/src/plugin_dispatch/tests.rs index 221b17d9f..49f179f8e 100644 --- a/crates/mesh-native-serving-plugin-host/src/plugin_dispatch/tests.rs +++ b/crates/mesh-native-serving-plugin-host/src/plugin_dispatch/tests.rs @@ -2,24 +2,52 @@ use std::{ sync::{Arc, atomic::Ordering, mpsc::sync_channel}, + thread, time::{Duration, Instant}, }; use skippy_server::frontend::{ GenerationAbort, GenerationCommit, GenerationLifecycleIngress, GenerationLifecycleObservation, - GenerationStart, LinearProposalDiscardReason, LinearProposalQuery, LinearProposalSourceOutcome, + GenerationStart, LinearProposalDiscardReason, LinearProposalQuery, LinearProposalReceipt, + LinearProposalSourceOutcome, OpaqueProposalDecisionId, }; -use super::{PluginCommand, PluginCommandQueue, PluginCommandQueueError, PluginDriver}; +use super::{ + PluginCommand, PluginCommandQueue, PluginCommandQueueError, PluginDriver, + finish_after_passive_fence, +}; use crate::{ NativeLifecycleIngress, test_support::{ - fake_active, fake_active_with_events, fake_active_with_failing_proposal, - fake_active_with_late_candidate, fake_active_with_observations, fake_active_with_options, - fake_active_with_timing, fake_observations, proposal_query, wait_for_event, + CallbackGate, fake_active, fake_active_with_events, fake_active_with_failing_proposal, + fake_active_with_late_candidate, fake_active_with_late_candidate_and_gates_with_events, + fake_active_with_observations, fake_active_with_options, fake_active_with_timing, + fake_active_with_timing_and_gates, fake_observations, proposal_query, wait_for_event, }, }; +fn queued_fence_count(queue: &PluginCommandQueue) -> usize { + queue + .state + .lock() + .unwrap() + .commands + .iter() + .filter(|queued| matches!(&queued.command, PluginCommand::Fence(_))) + .count() +} + +fn wait_until(predicate: impl Fn() -> bool) { + let deadline = Instant::now() + Duration::from_secs(1); + while !predicate() { + assert!( + Instant::now() < deadline, + "timed out waiting for test state" + ); + thread::yield_now(); + } +} + #[test] fn blocking_plugin_cannot_extend_the_decode_deadline() { let active = fake_active(Duration::from_millis(250)); @@ -103,7 +131,7 @@ fn slow_commit_abstains_before_plugin_dispatch_and_later_positions_recover() { } #[test] -fn running_passive_discard_cannot_delay_the_next_proposal() { +fn slow_passive_discard_cannot_run_the_next_proposal_after_its_deadline() { let (active, events, _) = fake_active_with_timing( Duration::ZERO, Duration::ZERO, @@ -126,10 +154,10 @@ fn running_passive_discard_cannot_delay_the_next_proposal() { assert!(response.proposal.unwrap().is_none()); assert_eq!( response.telemetry.outcome, - LinearProposalSourceOutcome::Abstained + LinearProposalSourceOutcome::HostDeadlineExceeded ); assert!(started.elapsed() < Duration::from_millis(60)); - assert_eq!(*events.lock().unwrap(), ["discard", "proposal"]); + assert_eq!(*events.lock().unwrap(), ["discard"]); } #[test] @@ -243,6 +271,269 @@ fn lifecycle_ingress_shares_plugin_queue_order_with_proposals() { assert_eq!(*events.lock().unwrap(), ["begin", "commit", "proposal"]); } +#[test] +fn finish_waits_for_prior_passive_discard() { + let (active, events, _) = fake_active_with_timing( + Duration::ZERO, + Duration::ZERO, + Duration::from_millis(100), + false, + ); + let driver = Arc::new(PluginDriver::spawn(active).unwrap()); + + driver + .enqueue_terminal(PluginCommand::Discard( + vec![1], + LinearProposalDiscardReason::PositionMismatch, + )) + .unwrap(); + let passive_queue = Arc::clone(&driver.passive_queue); + let finish_events = Arc::clone(&events); + let finish = std::thread::spawn(move || { + finish_after_passive_fence(&passive_queue, || { + finish_events.lock().unwrap().push("finish"); + Ok(()) + }) + }); + + wait_for_event(events.as_ref(), "discard"); + assert_eq!(*events.lock().unwrap(), ["discard"]); + finish.join().unwrap().unwrap(); + assert_eq!(*events.lock().unwrap(), ["discard", "finish"]); +} + +#[test] +fn proposal_waits_for_prior_passive_discard_ack_before_dispatch() { + let discard_gate = CallbackGate::new(); + let (active, events, _) = fake_active_with_timing_and_gates( + Duration::ZERO, + Duration::ZERO, + Duration::ZERO, + false, + Some(Arc::clone(&discard_gate)), + None, + ); + let driver = PluginDriver::spawn(active).unwrap(); + let _release_gate = discard_gate.release_on_drop(); + driver + .enqueue_terminal(PluginCommand::Discard( + vec![1], + LinearProposalDiscardReason::PositionMismatch, + )) + .unwrap(); + discard_gate.wait_until_entered(); + + let passive_queue = Arc::clone(&driver.passive_queue); + let driver_for_proposal = Arc::new(driver); + let proposal_driver = Arc::clone(&driver_for_proposal); + let proposal = thread::spawn(move || { + proposal_driver + .propose(proposal_query(Instant::now() + Duration::from_secs(1))) + .unwrap() + }); + wait_until(|| queued_fence_count(&passive_queue) == 1); + assert_eq!(*events.lock().unwrap(), ["discard"]); + + discard_gate.release(); + let response = proposal.join().unwrap(); + + assert_eq!( + response.telemetry.outcome, + LinearProposalSourceOutcome::Abstained + ); + assert_eq!( + *events.lock().unwrap(), + ["discard", "discard_done", "proposal"] + ); +} + +#[test] +fn proposal_waits_for_prior_passive_report_ack_before_dispatch() { + let report_gate = CallbackGate::new(); + let (active, events, _) = fake_active_with_timing_and_gates( + Duration::ZERO, + Duration::ZERO, + Duration::ZERO, + false, + Some(Arc::clone(&report_gate)), + None, + ); + let driver = PluginDriver::spawn(active).unwrap(); + let _release_gate = report_gate.release_on_drop(); + driver + .enqueue_terminal(PluginCommand::Report(LinearProposalReceipt::test_fixture( + OpaqueProposalDecisionId::new(vec![1]).unwrap(), + ))) + .unwrap(); + report_gate.wait_until_entered(); + + let passive_queue = Arc::clone(&driver.passive_queue); + let driver_for_proposal = Arc::new(driver); + let proposal_driver = Arc::clone(&driver_for_proposal); + let proposal = thread::spawn(move || { + proposal_driver + .propose(proposal_query(Instant::now() + Duration::from_secs(1))) + .unwrap() + }); + wait_until(|| queued_fence_count(&passive_queue) == 1); + assert_eq!(*events.lock().unwrap(), ["commit", "report"]); + + report_gate.release(); + let response = proposal.join().unwrap(); + assert_eq!( + response.telemetry.outcome, + LinearProposalSourceOutcome::Abstained + ); + assert_eq!( + *events.lock().unwrap(), + ["commit", "report", "report_done", "proposal"] + ); +} + +#[test] +fn late_candidate_discard_is_fenced_before_the_following_proposal() { + let discard_gate = CallbackGate::new(); + let (active, events) = fake_active_with_late_candidate_and_gates_with_events( + Duration::from_millis(20), + Some(Arc::clone(&discard_gate)), + None, + ); + let driver = Arc::new(PluginDriver::spawn(active).unwrap()); + let _release_gate = discard_gate.release_on_drop(); + + let (reply, response) = sync_channel(1); + driver + .queue + .try_enqueue(PluginCommand::Proposal( + proposal_query(Instant::now() + Duration::from_millis(5)), + reply, + )) + .unwrap(); + let first = response.recv_timeout(Duration::from_secs(1)).unwrap(); + assert_eq!( + first.telemetry.outcome, + LinearProposalSourceOutcome::CandidateReturnedTooLate + ); + discard_gate.wait_until_entered(); + assert_eq!(*events.lock().unwrap(), ["proposal", "discard"]); + + let next_driver = Arc::clone(&driver); + let next = thread::spawn(move || { + next_driver + .propose(proposal_query(Instant::now() + Duration::from_secs(1))) + .unwrap() + }); + wait_until(|| queued_fence_count(&driver.passive_queue) == 1); + assert_eq!(*events.lock().unwrap(), ["proposal", "discard"]); + + discard_gate.release(); + let second = next.join().unwrap(); + assert_eq!(second.telemetry.outcome, LinearProposalSourceOutcome::Ready); + assert_eq!( + *events.lock().unwrap(), + ["proposal", "discard", "discard_done", "proposal"] + ); +} + +#[test] +fn concurrent_proposals_enqueue_only_one_inflight_passive_fence() { + let discard_gate = CallbackGate::new(); + let (active, events, _) = fake_active_with_timing_and_gates( + Duration::ZERO, + Duration::ZERO, + Duration::ZERO, + false, + Some(Arc::clone(&discard_gate)), + None, + ); + let driver = Arc::new(PluginDriver::spawn(active).unwrap()); + let _release_gate = discard_gate.release_on_drop(); + driver + .enqueue_terminal(PluginCommand::Discard( + vec![1], + LinearProposalDiscardReason::PositionMismatch, + )) + .unwrap(); + discard_gate.wait_until_entered(); + + let callers = (0..8) + .map(|_| { + let driver = Arc::clone(&driver); + thread::spawn(move || { + driver + .propose(proposal_query(Instant::now() + Duration::from_secs(1))) + .unwrap() + }) + }) + .collect::>(); + wait_until(|| queued_fence_count(&driver.passive_queue) == 1); + assert_eq!(queued_fence_count(&driver.passive_queue), 1); + + discard_gate.release(); + for caller in callers { + assert_eq!( + caller.join().unwrap().telemetry.outcome, + LinearProposalSourceOutcome::Abstained + ); + } + assert_eq!( + events + .lock() + .unwrap() + .iter() + .filter(|&&event| event == "discard_done") + .count(), + 1 + ); +} + +#[test] +fn blocked_passive_completion_never_dispatches_expired_proposal() { + let discard_gate = CallbackGate::new(); + let (active, events, _) = fake_active_with_timing_and_gates( + Duration::ZERO, + Duration::ZERO, + Duration::ZERO, + false, + Some(Arc::clone(&discard_gate)), + None, + ); + let driver = Arc::new(PluginDriver::spawn(active).unwrap()); + let _release_gate = discard_gate.release_on_drop(); + driver + .enqueue_terminal(PluginCommand::Discard( + vec![1], + LinearProposalDiscardReason::PositionMismatch, + )) + .unwrap(); + discard_gate.wait_until_entered(); + + let expired_driver = Arc::clone(&driver); + let expired = thread::spawn(move || { + expired_driver + .propose(proposal_query(Instant::now() + Duration::from_millis(20))) + .unwrap() + }); + let expired_result = expired.join().unwrap(); + assert_eq!( + expired_result.telemetry.outcome, + LinearProposalSourceOutcome::HostDeadlineExceeded + ); + + discard_gate.release(); + let follow_up = driver + .propose(proposal_query(Instant::now() + Duration::from_secs(1))) + .unwrap(); + assert_eq!( + follow_up.telemetry.outcome, + LinearProposalSourceOutcome::Abstained + ); + assert_eq!( + *events.lock().unwrap(), + ["discard", "discard_done", "proposal"] + ); +} + #[test] fn proposal_applies_pending_tokens_before_lookup() { let (active, events) = fake_active_with_events(Duration::ZERO); @@ -448,6 +739,30 @@ fn a_full_passive_queue_still_accepts_the_terminal_discard() { .expect("terminal discard must use the reserved headroom"); } +#[test] +fn a_full_terminal_queue_still_accepts_the_passive_fence() { + let queue = PluginCommandQueue::new(); + while queue + .try_enqueue(PluginCommand::Discard( + vec![1], + LinearProposalDiscardReason::PositionMismatch, + )) + .is_ok() + {} + while queue + .try_enqueue_terminal(PluginCommand::Discard( + vec![2], + LinearProposalDiscardReason::DeadlineExceeded, + )) + .is_ok() + {} + + let (ack, _reply) = sync_channel(1); + queue + .try_enqueue_terminal(PluginCommand::Fence(ack)) + .expect("the fence must retain one exclusive terminal slot"); +} + #[test] fn a_late_candidate_delivers_its_discard_to_the_plugin() { // The host withholds a late candidate, so `discard` is the only way the diff --git a/crates/mesh-native-serving-plugin-host/src/test_support.rs b/crates/mesh-native-serving-plugin-host/src/test_support.rs index a420dd69e..b31e3dde8 100644 --- a/crates/mesh-native-serving-plugin-host/src/test_support.rs +++ b/crates/mesh-native-serving-plugin-host/src/test_support.rs @@ -5,7 +5,7 @@ use std::{ ffi::{c_char, c_void}, ptr::NonNull, sync::{ - Arc, Mutex, + Arc, Condvar, Mutex, atomic::{AtomicUsize, Ordering}, }, thread, @@ -19,6 +19,71 @@ use crate::{ActivePlugin, LoadedDefinition, MAX_NATIVE_PLUGIN_PROPOSAL_TOKENS}; pub(crate) static FAKE_NAME: &[u8] = b"test-serving-plugin"; +/// Deterministic callback gate for ordering tests. The callback announces +/// entry, then waits until the test explicitly releases it. +pub(crate) struct CallbackGate { + state: Mutex, + signal: Condvar, +} + +#[derive(Default)] +struct CallbackGateState { + entered: bool, + released: bool, +} + +impl CallbackGate { + pub(crate) fn new() -> Arc { + Arc::new(Self { + state: Mutex::new(CallbackGateState::default()), + signal: Condvar::new(), + }) + } + + fn wait(&self) { + let mut state = self.state.lock().unwrap(); + state.entered = true; + self.signal.notify_all(); + while !state.released { + state = self.signal.wait(state).unwrap(); + } + } + + pub(crate) fn wait_until_entered(&self) { + let mut state = self.state.lock().unwrap(); + let deadline = Instant::now() + Duration::from_secs(2); + while !state.entered { + let remaining = deadline.saturating_duration_since(Instant::now()); + if remaining.is_zero() { + panic!("timed out waiting for gated callback to enter"); + } + let (next_state, timeout) = self.signal.wait_timeout(state, remaining).unwrap(); + state = next_state; + if timeout.timed_out() && !state.entered { + panic!("timed out waiting for gated callback to enter"); + } + } + } + + pub(crate) fn release(&self) { + let mut state = self.state.lock().unwrap(); + state.released = true; + self.signal.notify_all(); + } + + pub(crate) fn release_on_drop(self: &Arc) -> CallbackGateRelease { + CallbackGateRelease(Arc::clone(self)) + } +} + +pub(crate) struct CallbackGateRelease(Arc); + +impl Drop for CallbackGateRelease { + fn drop(&mut self) { + self.0.release(); + } +} + /// Per-instance observations, so tests running in parallel cannot see each /// other's callback counts. pub(crate) struct FakeObservations { @@ -34,6 +99,8 @@ pub(crate) struct FakeState { poll_fails: bool, commit_delay: Duration, report_delay: Duration, + report_gate: Option>, + proposal_gate: Option>, begin_fails: bool, events: Arc>>, abort_count: Arc, @@ -89,9 +156,11 @@ unsafe extern "C" fn fake_abort( } unsafe extern "C" fn fake_finish( - _instance: abi::PluginInstance, + instance: abi::PluginInstance, _event: *const abi::GenerationFinish, ) -> abi::PluginStatus { + let state = unsafe { &*instance.cast::() }; + state.events.lock().unwrap().push("finish"); abi::PluginStatus::OK } @@ -102,6 +171,9 @@ unsafe extern "C" fn fake_start_proposal( ) -> abi::PluginStatus { let state = unsafe { &*instance.cast::() }; state.events.lock().unwrap().push("proposal"); + if let Some(gate) = &state.proposal_gate { + gate.wait(); + } thread::sleep(state.start_delay); unsafe { *operation = 1 }; abi::PluginStatus::OK @@ -146,6 +218,10 @@ unsafe extern "C" fn fake_report_proposal( ) -> abi::PluginStatus { let state = unsafe { &*instance.cast::() }; state.events.lock().unwrap().push("report"); + if let Some(gate) = &state.report_gate { + gate.wait(); + state.events.lock().unwrap().push("report_done"); + } thread::sleep(state.report_delay); abi::PluginStatus::OK } @@ -156,6 +232,10 @@ unsafe extern "C" fn fake_discard_proposal( ) -> abi::PluginStatus { let state = unsafe { &*instance.cast::() }; state.events.lock().unwrap().push("discard"); + if let Some(gate) = &state.report_gate { + gate.wait(); + state.events.lock().unwrap().push("discard_done"); + } thread::sleep(state.report_delay); abi::PluginStatus::OK } @@ -229,6 +309,28 @@ pub(crate) fn fake_active_with_timing( ActivePlugin, Arc>>, Arc, +) { + fake_active_with_timing_and_gates( + start_delay, + commit_delay, + report_delay, + begin_fails, + None, + None, + ) +} + +pub(crate) fn fake_active_with_timing_and_gates( + start_delay: Duration, + commit_delay: Duration, + report_delay: Duration, + begin_fails: bool, + report_gate: Option>, + proposal_gate: Option>, +) -> ( + ActivePlugin, + Arc>>, + Arc, ) { let table = Box::leak(Box::new(fake_table())); let definition = Arc::new(LoadedDefinition { @@ -247,6 +349,8 @@ pub(crate) fn fake_active_with_timing( poll_fails: false, commit_delay, report_delay, + report_gate, + proposal_gate, begin_fails, events: Arc::clone(&events), abort_count: Arc::clone(&abort_count), @@ -266,14 +370,28 @@ pub(crate) fn fake_active_with_timing( } pub(crate) fn fake_active_with_late_candidate(poll_delay: Duration) -> ActivePlugin { - let (active, _, _) = - fake_active_with_timing(Duration::ZERO, Duration::ZERO, Duration::ZERO, false); + fake_active_with_late_candidate_and_gates_with_events(poll_delay, None, None).0 +} + +pub(crate) fn fake_active_with_late_candidate_and_gates_with_events( + poll_delay: Duration, + report_gate: Option>, + proposal_gate: Option>, +) -> (ActivePlugin, Arc>>) { + let (active, events, _) = fake_active_with_timing_and_gates( + Duration::ZERO, + Duration::ZERO, + Duration::ZERO, + false, + report_gate, + proposal_gate, + ); let instance = active.instance.unwrap().as_ptr().cast::(); unsafe { (*instance).poll_delay = poll_delay; (*instance).poll_returns_candidate = true; } - active + (active, events) } /// Clones the per-instance observation handles out of a fake plugin. diff --git a/crates/skippy-server/Cargo.toml b/crates/skippy-server/Cargo.toml index b4f413f2a..1a7338943 100644 --- a/crates/skippy-server/Cargo.toml +++ b/crates/skippy-server/Cargo.toml @@ -14,6 +14,7 @@ path = "src/lib.rs" [features] default = [] dynamic-native-runtime = ["skippy-runtime/dynamic-native-runtime"] +test-support = [] [dependencies] ahash.workspace = true diff --git a/crates/skippy-server/src/frontend/linear_proposal.rs b/crates/skippy-server/src/frontend/linear_proposal.rs index 76df69e88..8de12a63f 100644 --- a/crates/skippy-server/src/frontend/linear_proposal.rs +++ b/crates/skippy-server/src/frontend/linear_proposal.rs @@ -219,6 +219,35 @@ pub struct LinearProposalReceipt { } impl LinearProposalReceipt { + #[cfg(feature = "test-support")] + #[doc(hidden)] + pub fn test_fixture(decision_id: OpaqueProposalDecisionId) -> Self { + Self { + request_id: 1, + session_id: 2, + decision_id, + disposition: LinearProposalDisposition::FullAccept, + proposal_token_count: 1, + verification_rows: 1, + accepted_proposal_tokens: 1, + committed_tokens: vec![1].into_boxed_slice(), + verification_row_predictions: vec![1].into_boxed_slice(), + canonical_prediction_count: 1, + correction_or_boundary_token: Some(1), + base_position: 0, + position_after_verification: 1, + canonical_position: 1, + trimmed_rows: 0, + proposal_elapsed_us: 0, + verification_elapsed_us: 0, + repair_elapsed_us: 0, + total_elapsed_us: 0, + runtime_lock_wait_us: 0, + runtime_lock_hold_us: 0, + runtime_lock_acquires: 0, + } + } + pub(crate) fn insert_telemetry_attrs(&self, attrs: &mut BTreeMap) { attrs.insert( "llama_stage.linear_proposal.disposition".to_string(),