From c5d55fe9cc70ac15156e09d68ce444bf54157419 Mon Sep 17 00:00:00 2001 From: leynos Date: Thu, 22 Oct 2026 22:43:47 +0200 Subject: [PATCH 01/21] Cover pool permit races and baselines (#646) Add pool-size 1, 4, and 16 benchmark workloads and deterministic coverage for rotation, saturation, cancellation, close, and slot ownership. Keep the same measurement boundaries for the borrowed-race comparison. --- .../runtime_ownership_baselines/scenarios.rs | 21 +- .../runtime_ownership_baselines_alloc/main.rs | 16 +- .../pool_bracket.rs | 7 +- .../reporting.rs | 39 +-- .../scopes.rs | 46 ++-- src/client/pool/client_pool_tests.rs | 100 ++++++- src/client/pool/core_tests/permit_race.rs | 256 ++++++++++++++++++ src/client/pool/core_tests/sequence_tests.rs | 3 + .../client_pool_benchmark_support/mod.rs | 1 + .../tests/contended.rs | 51 +++- .../client_pool_benchmark_support/workload.rs | 25 ++ .../src/runtime_ownership_benchmarks/mod.rs | 1 + 12 files changed, 471 insertions(+), 95 deletions(-) create mode 100644 src/client/pool/core_tests/permit_race.rs diff --git a/benches/runtime_ownership_baselines/scenarios.rs b/benches/runtime_ownership_baselines/scenarios.rs index dd2ce2b15..86d9dac66 100644 --- a/benches/runtime_ownership_baselines/scenarios.rs +++ b/benches/runtime_ownership_baselines/scenarios.rs @@ -20,6 +20,7 @@ use wireframe_testing::runtime_ownership_benchmarks::{ PushWorkload, StartupPath, StartupWorkload, + contended_fifo_pool_workloads, }; /// Criterion sample count for every runtime-ownership group. @@ -245,21 +246,17 @@ pub(crate) fn uncontended_pool_workloads() -> Vec { .collect() } -/// The contended pool workloads reported as bounded observations, not timings. +/// Contended workloads reported as bounded timing observations. /// -/// Both fairness policies run over one pool size: the issue asks for the -/// policies to be compared, not for every policy to be repeated at every size. +/// FIFO runs cover all baseline pool sizes. Round-robin remains the existing +/// size-four comparison so its label and workload stay directly comparable. pub(crate) fn contended_pool_workloads() -> Vec { - [ - AcquisitionMode::ContendedFifo, - AcquisitionMode::ContendedRoundRobin, - ] - .into_iter() - .map(|mode| PoolWorkload { + let mut workloads = contended_fifo_pool_workloads().to_vec(); + workloads.push(PoolWorkload { pool_size: CONTENDED_POOL_SIZE, - mode, + mode: AcquisitionMode::ContendedRoundRobin, contenders: CONTENDED_POOL_CONTENDERS, ..PoolWorkload::default() - }) - .collect() + }); + workloads } diff --git a/benches/runtime_ownership_baselines_alloc/main.rs b/benches/runtime_ownership_baselines_alloc/main.rs index c91cc1a61..d3d2bc638 100644 --- a/benches/runtime_ownership_baselines_alloc/main.rs +++ b/benches/runtime_ownership_baselines_alloc/main.rs @@ -39,14 +39,14 @@ //! //! # Contended pool acquisition //! -//! Contended pool acquisition is counted once, after `Criterion::final_summary`, -//! and reported as a `tracing` event rather than timed. Sustained contention -//! intermittently fails or stalls the pool acquire path — a pre-existing defect -//! recorded in `docs/runtime-ownership-baselines.md` and belonging to ADR 013 / -//! issue #647. Criterion's timing pass runs far more iterations than an -//! allocation window, so it reaches that interleaving almost every run; that is -//! how this binary failed on its first execution. The count itself is unaffected -//! because it comes from a single counted window. +//! FIFO contended pool acquisition is counted once per baseline pool size, after +//! `Criterion::final_summary`, and reported as a `tracing` event rather than +//! timed. Sustained contention intermittently fails or stalls the pool acquire +//! path — a pre-existing defect recorded in `docs/runtime-ownership-baselines.md` +//! and belonging to ADR 013 / issue #647. Criterion's timing pass runs far more +//! iterations than an allocation window, so it reaches that interleaving almost +//! every run; that is how this binary failed on its first execution. Each count +//! comes from one bracketed window. //! //! # Layout //! diff --git a/benches/runtime_ownership_baselines_alloc/pool_bracket.rs b/benches/runtime_ownership_baselines_alloc/pool_bracket.rs index b15fceb1e..80035a27d 100644 --- a/benches/runtime_ownership_baselines_alloc/pool_bracket.rs +++ b/benches/runtime_ownership_baselines_alloc/pool_bracket.rs @@ -13,10 +13,7 @@ use wireframe_testing::{ runtime_ownership_benchmarks::{PoolHarness, PoolWorkload}, }; -use crate::{ - probes::expect_bench, - scopes::{ALLOCATION_ITERATIONS, AllocationWorkloads}, -}; +use crate::{probes::expect_bench, scopes::ALLOCATION_ITERATIONS}; /// Run the pool scope the way the timing pass should replay it. /// @@ -103,7 +100,7 @@ fn pool_bracket_readings( } /// Build the workload the bracket checks measure, at their own parameters. -fn bracket_workload() -> PoolWorkload { AllocationWorkloads::new().pool } +fn bracket_workload() -> PoolWorkload { PoolWorkload::default() } /// Prove the pool scope's bracketed window really excludes pool construction. /// diff --git a/benches/runtime_ownership_baselines_alloc/reporting.rs b/benches/runtime_ownership_baselines_alloc/reporting.rs index fa1fb116b..946a27736 100644 --- a/benches/runtime_ownership_baselines_alloc/reporting.rs +++ b/benches/runtime_ownership_baselines_alloc/reporting.rs @@ -2,15 +2,14 @@ //! //! A count is recorded before its timing pass and emitted under a label whose //! identity does not depend on it, so Criterion's saved results stay comparable -//! between runs. The contended pool scope is published as an event alone: it -//! cannot survive a Criterion timing pass, but the issue still asks for its -//! number. +//! between runs. FIFO contended pool counts are published as events alone: they +//! cannot survive a Criterion timing pass, but the baseline needs their counts. use std::{hint::black_box, time::Duration}; use criterion::{BenchmarkId, Criterion}; use tokio::runtime::Runtime; -use wireframe_testing::runtime_ownership_benchmarks::{AcquisitionMode, PoolWorkload}; +use wireframe_testing::runtime_ownership_benchmarks::contended_fifo_pool_workloads; use crate::{ pool_bracket::count_pool_acquisition_window, @@ -58,26 +57,6 @@ where total } -/// Contended pool workload, recorded as an observation rather than a timing. -/// -/// Contenders must outnumber slots for the pool to be exhausted, and the -/// default `contenders` of 16 clears the default pool size of 4 comfortably. -/// -/// This workload is deliberately absent from `allocation_scopes`: sustained -/// contention intermittently stalls the pool acquire path or fails it with a -/// spurious `disconnected`, and Criterion's timing pass runs far more -/// iterations than the four in an allocation window, so it reaches that -/// interleaving almost every run. Measuring it here once, under a counted -/// window, records the number the issue asks for without handing the scenario -/// to a pass that cannot survive it. See `docs/runtime-ownership-baselines.md`. -fn contended_pool_workload() -> PoolWorkload { - PoolWorkload { - pool_size: 4, - mode: AcquisitionMode::ContendedFifo, - ..PoolWorkload::default() - } -} - /// Emit a successful contended allocation count as an event. /// /// Kept as its own function because a `tracing` macro expands to an @@ -111,7 +90,7 @@ fn report_contended_allocations(label: &str, outcome: Result) { } } -/// Record the contended pool allocation count as a bounded observation. +/// Record FIFO contended allocation counts across the baseline pool sizes. /// /// Run once after Criterion has finished, so a failure is reported rather than /// aborting the benchmark run. The count comes from the same bracketed window the @@ -119,11 +98,11 @@ fn report_contended_allocations(label: &str, outcome: Result) { /// and the timing are dropped, because repetition is what the acquire-path defect /// makes unreliable. See `docs/runtime-ownership-baselines.md`. pub(crate) fn observe_contended_pool_allocations(runtime: &Runtime) { - let workload = contended_pool_workload(); - let label = workload.label(); - let outcome = count_pool_acquisition_window(runtime, workload, ALLOCATION_ITERATIONS); - - report_contended_allocations(&label, outcome); + for workload in contended_fifo_pool_workloads() { + let label = workload.label(); + let outcome = count_pool_acquisition_window(runtime, workload, ALLOCATION_ITERATIONS); + report_contended_allocations(&label, outcome); + } } /// Install a subscriber for the allocation binary's own events only. diff --git a/benches/runtime_ownership_baselines_alloc/scopes.rs b/benches/runtime_ownership_baselines_alloc/scopes.rs index 94d1ace1c..768039486 100644 --- a/benches/runtime_ownership_baselines_alloc/scopes.rs +++ b/benches/runtime_ownership_baselines_alloc/scopes.rs @@ -17,6 +17,7 @@ use wireframe_testing::{ AcquisitionMode, ActorWorkload, CancellationMode, + POOL_SIZES, PoolWorkload, PushScenario, PushWorkload, @@ -128,7 +129,7 @@ impl<'a> AllocationScope<'a> { /// measurement's own counters are dropped: a scope closure's only job is to /// perform the work the counting window observes. Adapting every scope through /// one helper is what keeps [`allocation_scopes`] short enough to read and keeps -/// the six scopes visibly identical in shape. +/// the scopes visibly identical in shape. fn run_measurement( runtime: &Runtime, measurement: impl std::future::Future>, @@ -141,11 +142,10 @@ fn run_measurement( /// The workload parameters this binary records allocations for, by scope. /// -/// Split from [`allocation_scopes`] because the six literal constructions alone -/// push that function past the workspace's line limit. Keeping the parameters in -/// one block also makes them easy to compare at a glance, which is their point: -/// each differs from the timing binary's defaults deliberately, to keep the -/// allocation windows short enough to compare as small integers. +/// Split from [`allocation_scopes`] so its closures stay below the workspace's +/// function-size limit. Keeping the parameters in one block also makes them +/// easy to compare at a glance, which is their point: each differs from the +/// timing binary's defaults deliberately, to keep allocation windows short. pub(crate) struct AllocationWorkloads { /// Actor drain, with cancellation disabled so the window covers the loop. actor: ActorWorkload, @@ -157,12 +157,12 @@ pub(crate) struct AllocationWorkloads { startup_prepared: StartupWorkload, /// The per-connection factory path, which prepares once per connection. startup_per_connection: StartupWorkload, - /// Uncontended acquire/release, the only pool mode safe to time. - pub(crate) pool: PoolWorkload, + /// Uncontended acquire/release at the named baseline pool sizes. + pool_workloads: [PoolWorkload; POOL_SIZES.len()], } impl AllocationWorkloads { - /// Build the six scopes at their allocation-window parameters. + /// Build the scopes at their allocation-window parameters. pub(crate) fn new() -> Self { Self { actor: ActorWorkload { @@ -191,11 +191,11 @@ impl AllocationWorkloads { path: StartupPath::PerConnectionFactory, ..StartupWorkload::default() }, - pool: PoolWorkload { - pool_size: 4, + pool_workloads: POOL_SIZES.map(|pool_size| PoolWorkload { + pool_size, mode: AcquisitionMode::Uncontended, ..PoolWorkload::default() - }, + }), } } } @@ -209,7 +209,7 @@ pub(crate) fn allocation_scopes(runtime: &Runtime) -> Vec> { let workloads = AllocationWorkloads::new(); let iterations = ALLOCATION_ITERATIONS; - vec![ + let mut scopes = vec![ AllocationScope::windowed(workloads.actor.label(), move || { run_measurement(runtime, measure_actor_drain(workloads.actor, iterations)) }), @@ -231,12 +231,18 @@ pub(crate) fn allocation_scopes(runtime: &Runtime) -> Vec> { measure_startup(workloads.startup_per_connection, iterations), ) }), - AllocationScope { - label: workloads.pool.label(), - count: Box::new(move || { - count_pool_acquisition_window(runtime, workloads.pool, iterations) + ]; + scopes.extend( + workloads + .pool_workloads + .into_iter() + .map(|workload| AllocationScope { + label: workload.label(), + count: Box::new(move || { + count_pool_acquisition_window(runtime, workload, iterations) + }), + replay: Box::new(move || run_pool_acquisition(runtime, workload, iterations)), }), - replay: Box::new(move || run_pool_acquisition(runtime, workloads.pool, iterations)), - }, - ] + ); + scopes } diff --git a/src/client/pool/client_pool_tests.rs b/src/client/pool/client_pool_tests.rs index 692887441..8c89f1733 100644 --- a/src/client/pool/client_pool_tests.rs +++ b/src/client/pool/client_pool_tests.rs @@ -1,6 +1,8 @@ //! Tests for pool, handle, lease, and shared-core lifetimes. use std::{ + error::Error, + io, sync::{ Arc, atomic::{AtomicUsize, Ordering}, @@ -11,6 +13,7 @@ use std::{ use crate::{ client::pool::{ PoolFairnessPolicy, + PoolTaskChurnObserver, client_pool::{PoolSchedulerControl, WireframeClientPool}, core::PoolCore, lease::PooledClientLease, @@ -47,6 +50,30 @@ fn test_lease(pool: &TestPool) -> PooledClientLease { .expect("the selected slot should resolve in the pool core") } +fn has_unexpected_eof(error: &(dyn Error + 'static)) -> bool { + let mut source = Some(error); + while let Some(error) = source { + if error + .downcast_ref::() + .is_some_and(|io_error| io_error.kind() == io::ErrorKind::UnexpectedEof) + { + return true; + } + source = error.source(); + } + false +} + +async fn wait_for_capacity_wait(observer: &PoolTaskChurnObserver, message: &str) { + tokio::time::timeout(Duration::from_secs(1), async { + while observer.snapshot().capacity_waits == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect(message); +} + #[tokio::test] async fn pool_and_handle_drop_keep_lease_slot_alive_until_final_lease_drop() { let drops = Arc::new(AtomicUsize::new(0)); @@ -79,30 +106,81 @@ async fn close_rejects_new_and_queued_acquisitions_while_an_existing_lease_keeps let mut queued_handle = pool.handle(); let queued_acquisition = tokio::spawn(async move { queued_handle.acquire().await }); - tokio::time::timeout(Duration::from_secs(1), async { - while observer.snapshot().capacity_waits == 0 { - tokio::task::yield_now().await; - } - }) - .await - .expect("the second acquisition should enter the wait path before close"); + wait_for_capacity_wait( + &observer, + "the second acquisition should enter the wait path before close", + ) + .await; pool.close().await; + let new_result = handle.acquire().await; + let Err(new_error) = new_result else { + panic!("a closed pool must reject new leases"); + }; assert!( - handle.acquire().await.is_err(), - "a closed pool must reject new leases" + has_unexpected_eof(&new_error), + "a closed pool should report the disconnected UnexpectedEof error: {new_error}" ); let queued_result = tokio::time::timeout(Duration::from_secs(1), queued_acquisition) .await .expect("pool close should complete the queued acquisition") .expect("the queued acquisition task should complete"); + let Err(queued_error) = queued_result else { + panic!("a queued acquisition must not receive a lease after close"); + }; assert!( - queued_result.is_err(), - "a queued acquisition must not receive a lease after close" + has_unexpected_eof(&queued_error), + "a queued acquisition should report the disconnected UnexpectedEof error: {queued_error}" ); assert!( lease.slot().is_some(), "an existing lease must retain its selected slot through close" ); } + +#[tokio::test] +async fn cancelled_queued_acquisition_releases_capacity_and_core_ownership() { + let drops = Arc::new(AtomicUsize::new(0)); + let pool = test_pool(Some(Arc::clone(&drops))); + let mut owner_handle = pool.handle(); + let lease = test_lease(&pool); + let observer = pool.task_churn_observer(); + let mut queued_handle = pool.handle(); + let queued_acquisition = tokio::spawn(async move { queued_handle.acquire().await }); + + wait_for_capacity_wait( + &observer, + "the acquisition should enter the capacity wait before cancellation", + ) + .await; + + queued_acquisition.abort(); + let cancelled = tokio::time::timeout(Duration::from_secs(1), queued_acquisition) + .await + .expect("the cancelled acquisition task should stop promptly"); + let Err(join_error) = cancelled else { + panic!("aborting the acquisition should cancel its task"); + }; + assert!( + join_error.is_cancelled(), + "the queued acquisition task should have been cancelled" + ); + + drop(lease); + let later_lease = tokio::time::timeout(Duration::from_secs(1), owner_handle.acquire()) + .await + .expect("capacity released after cancellation should wake a fresh acquisition") + .expect("cancellation must not consume the released slot permit"); + drop(later_lease); + drop(owner_handle); + drop(pool); + + tokio::time::timeout(Duration::from_secs(1), async { + while drops.load(Ordering::Acquire) != 2 { + tokio::task::yield_now().await; + } + }) + .await + .expect("the cancelled race must not retain the pool core or its slot"); +} diff --git a/src/client/pool/core_tests/permit_race.rs b/src/client/pool/core_tests/permit_race.rs new file mode 100644 index 000000000..0e17985b0 --- /dev/null +++ b/src/client/pool/core_tests/permit_race.rs @@ -0,0 +1,256 @@ +//! Rotation, cancellation, and resource-drop coverage for permit selection. + +use std::{sync::Arc, time::Duration}; + +use rstest::rstest; +use tokio::sync::OwnedSemaphorePermit; + +use super::{TestCore, four_slot_core, test_core}; + +fn hold_all_permits(core: &TestCore) -> Vec { + core.slots + .iter() + .map(|slot| { + slot.try_acquire_permit() + .expect("each slot should have one available permit") + }) + .collect() +} + +fn hold_every_permit_except(core: &TestCore, free_index: usize) -> Vec { + core.slots + .iter() + .enumerate() + .filter(|(slot_index, _)| *slot_index != free_index) + .map(|(_, slot)| { + slot.try_acquire_permit() + .expect("every slot other than the free slot should be held") + }) + .collect() +} + +fn assert_all_slots_are_available(core: &TestCore) { + let permits = hold_all_permits(core); + drop(permits); +} + +#[rstest] +#[case::one_slot(1, &[0, 0, 0])] +#[case::two_slots(2, &[0, 1, 0])] +#[case::four_slots(4, &[0, 1, 2, 3, 0])] +#[tokio::test] +async fn immediate_probe_rotates_across_slots( + #[case] slot_count: usize, + #[case] expected: &[usize], +) { + let core = test_core(slot_count); + let selected = expected + .iter() + .map(|_| { + let (slot_index, permit) = core + .try_acquire_permit() + .expect("a free slot should grant an immediate permit"); + drop(permit); + slot_index + }) + .collect::>(); + + assert_eq!(selected.as_slice(), expected); +} + +#[rstest] +#[tokio::test(flavor = "current_thread")] +async fn failed_and_cancelled_acquisitions_each_advance_rotation_once() { + let core = test_core(2); + let held = hold_all_permits(&core); + + assert!(core.try_acquire_permit().is_none()); + drop(held); + + let (after_failed_try, permit) = core + .try_acquire_permit() + .expect("the next immediate probe should find free capacity"); + assert_eq!(after_failed_try, 1); + drop(permit); + + let held = hold_all_permits(&core); + { + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + assert!(futures::poll!(acquisition.as_mut()).is_pending()); + } + drop(held); + + let (after_cancelled_race, permit) = core + .try_acquire_permit() + .expect("the next immediate probe should find free capacity"); + assert_eq!(after_cancelled_race, 1); + drop(permit); +} + +#[rstest] +#[tokio::test] +async fn empty_core_rejects_both_immediate_and_async_acquisition() { + let core = test_core(0); + + assert!(core.try_acquire_permit().is_none()); + assert!(core.acquire_slot_permit().await.is_err()); +} + +#[rstest] +#[tokio::test] +async fn async_probe_rotates_across_four_slots(four_slot_core: Arc) { + let mut selected = Vec::new(); + for _ in 0..5 { + let (slot_index, permit) = four_slot_core + .acquire_slot_permit() + .await + .expect("a free slot should grant an immediate async permit"); + drop(permit); + selected.push(slot_index); + } + + assert_eq!(selected, [0, 1, 2, 3, 0]); +} + +#[rstest] +#[tokio::test] +async fn sync_and_async_probes_share_one_rotation_cursor(four_slot_core: Arc) { + let (first, permit) = four_slot_core + .try_acquire_permit() + .expect("the first immediate probe should succeed"); + drop(permit); + let (second, permit) = four_slot_core + .acquire_slot_permit() + .await + .expect("the second immediate probe should succeed"); + drop(permit); + let (third, permit) = four_slot_core + .try_acquire_permit() + .expect("the third immediate probe should succeed"); + drop(permit); + let (fourth, permit) = four_slot_core + .acquire_slot_permit() + .await + .expect("the fourth immediate probe should succeed"); + + assert_eq!([first, second, third, fourth], [0, 1, 2, 3]); + drop(permit); +} + +#[rstest] +#[case::two_first(2, 0)] +#[case::two_second(2, 1)] +#[case::four_first(4, 0)] +#[case::four_second(4, 1)] +#[case::four_third(4, 2)] +#[case::four_fourth(4, 3)] +#[tokio::test(flavor = "current_thread")] +async fn immediate_probe_finds_each_free_slot( + #[case] slot_count: usize, + #[case] free_index: usize, +) { + let core = test_core(slot_count); + let _held = hold_every_permit_except(&core, free_index); + + let (selected, permit) = core + .acquire_slot_permit() + .await + .expect("the only free slot should grant its permit"); + + assert_eq!(selected, free_index); + drop(permit); +} + +#[rstest] +#[case::two_first(2, 0)] +#[case::two_second(2, 1)] +#[case::four_first(4, 0)] +#[case::four_second(4, 1)] +#[case::four_third(4, 2)] +#[case::four_fourth(4, 3)] +#[tokio::test(flavor = "current_thread")] +async fn blocked_race_returns_the_single_slot_released_after_registration( + #[case] slot_count: usize, + #[case] free_index: usize, +) { + let core = test_core(slot_count); + let mut held = hold_all_permits(&core); + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + assert!(futures::poll!(acquisition.as_mut()).is_pending()); + + drop(held.swap_remove(free_index)); + let (selected, permit) = tokio::time::timeout(Duration::from_secs(1), acquisition) + .await + .expect("released capacity should wake the blocked race") + .expect("the released slot should grant exactly one permit"); + + assert_eq!(selected, free_index); + assert!( + core.slot(selected) + .expect("the selected slot should exist") + .try_acquire_permit() + .is_none(), + "the winner's permit must remain owned by the acquisition" + ); + drop(permit); + drop(held); + assert_all_slots_are_available(&core); +} + +#[rstest] +#[tokio::test(flavor = "current_thread")] +async fn simultaneous_readiness_returns_one_permit_and_drops_the_loser() { + let core = test_core(2); + let held = hold_all_permits(&core); + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + assert!(futures::poll!(acquisition.as_mut()).is_pending()); + + drop(held); + let (winner_index, winner_permit) = tokio::time::timeout(Duration::from_secs(1), acquisition) + .await + .expect("both released slots should wake the race") + .expect("one of the ready slots should win"); + let loser_index = 1_usize.saturating_sub(winner_index); + + assert!( + core.slot(winner_index) + .expect("the winner should be a valid slot") + .try_acquire_permit() + .is_none(), + "the returned permit must reserve exactly the winning slot" + ); + let loser_permit = core + .slot(loser_index) + .expect("the other slot should exist") + .try_acquire_permit(); + assert!( + loser_permit.is_some(), + "a losing ready future must not retain its acquired permit" + ); + + drop(loser_permit); + drop(winner_permit); + assert_all_slots_are_available(&core); +} + +#[rstest] +#[tokio::test(flavor = "current_thread")] +async fn cancelling_a_blocked_race_leaves_every_slot_usable() { + let core = test_core(4); + let held = hold_all_permits(&core); + { + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + assert!(futures::poll!(acquisition.as_mut()).is_pending()); + } + + drop(held); + assert_all_slots_are_available(&core); + let (slot_index, permit) = + tokio::time::timeout(Duration::from_secs(1), core.acquire_slot_permit()) + .await + .expect("a fresh acquisition should complete after cancellation") + .expect("cancellation must not lose any slot permit"); + + assert!(slot_index < core.slots.len()); + drop(permit); +} diff --git a/src/client/pool/core_tests/sequence_tests.rs b/src/client/pool/core_tests/sequence_tests.rs index f5ce43861..c9f4e7df9 100644 --- a/src/client/pool/core_tests/sequence_tests.rs +++ b/src/client/pool/core_tests/sequence_tests.rs @@ -363,3 +363,6 @@ impl GeneratedSlotSequence { Ok(()) } } + +#[path = "core_tests/permit_race.rs"] +mod permit_race; diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/mod.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/mod.rs index 9294de731..daba018df 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/mod.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/mod.rs @@ -131,4 +131,5 @@ pub use workload::{ POOL_SIZES, PoolMeasurement, PoolWorkload, + contended_fifo_pool_workloads, }; diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs index 77e942f75..c0c2d62c8 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs @@ -4,6 +4,7 @@ use crate::runtime_ownership_benchmarks::client_pool_benchmark_support::{ AcquisitionMode, PoolMeasurement, PoolWorkload, + contended_fifo_pool_workloads, measure_pool_acquisition, }; @@ -20,6 +21,33 @@ use crate::runtime_ownership_benchmarks::client_pool_benchmark_support::{ /// serious defect, and is reported as such. const CONTENDED_BURST_ATTEMPTS: usize = 8; +#[test] +fn fifo_baseline_workloads_cover_each_pool_size_and_can_contend() { + let workloads = contended_fifo_pool_workloads(); + + assert_eq!(workloads.map(|workload| workload.pool_size), [1, 4, 16]); + assert_eq!(workloads.map(|workload| workload.contenders), [16, 16, 17]); + assert_eq!( + workloads.map(|workload| workload.mode), + [AcquisitionMode::ContendedFifo; 3] + ); + for workload in workloads { + assert!( + workload.validate().is_ok(), + "{} must exceed its slot count to enter the contended path", + workload.label() + ); + } + assert_eq!( + workloads[1].label(), + "contended_fifo_pool_4_contenders_16_rounds_8" + ); + assert_eq!( + workloads[2].label(), + "contended_fifo_pool_16_contenders_17_rounds_8" + ); +} + /// Run one contended burst, retrying the acquire path's known spurious failures. /// /// Both recorded failure modes — the spurious `disconnected`, and the stall @@ -79,24 +107,29 @@ async fn contended_mode_needs_more_contenders_than_slots() { /// of contention: every release notifies the scheduler whether or not anyone is /// waiting. #[rstest::rstest] -#[case(AcquisitionMode::ContendedFifo)] -#[case(AcquisitionMode::ContendedRoundRobin)] +#[case::one_slot_fifo(1, 16, AcquisitionMode::ContendedFifo)] +#[case::four_slots_fifo(4, 16, AcquisitionMode::ContendedFifo)] +#[case::sixteen_slots_fifo(16, 17, AcquisitionMode::ContendedFifo)] +#[case::four_slots_round_robin(4, 16, AcquisitionMode::ContendedRoundRobin)] #[tokio::test] -async fn contended_acquisition_observes_capacity_waits(#[case] mode: AcquisitionMode) { - const CONTENDERS: usize = 16; +async fn contended_acquisition_observes_capacity_waits( + #[case] pool_size: usize, + #[case] contenders: usize, + #[case] mode: AcquisitionMode, +) { let measurement = contended_burst_with_retry(PoolWorkload { - pool_size: 4, + pool_size, mode, acquisitions: 8, - contenders: CONTENDERS, + contenders, }) .await .expect("contended measurement failed in every retry attempt"); assert!( - measurement.capacity_waits >= u64::try_from(CONTENDERS).unwrap_or(u64::MAX), - "a burst that empties a 4-slot pool and holds it empty must park every one of \ - {CONTENDERS} contenders, but {} waits were recorded", + measurement.capacity_waits >= u64::try_from(contenders).unwrap_or(u64::MAX), + "a burst that empties a {pool_size}-slot pool and holds it empty must park every one of \ + {contenders} contenders, but {} waits were recorded", measurement.capacity_waits ); assert!( diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/workload.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/workload.rs index c670624e2..52e8e1a56 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/workload.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/workload.rs @@ -213,6 +213,31 @@ impl PoolWorkload { } } +/// FIFO contended workloads for every pool size used by runtime baselines. +/// +/// The default of sixteen contenders exceeds pool sizes one and four. A +/// 16-slot pool needs one additional contender for the harness to observe a +/// capacity wait, so its label records seventeen. +/// +/// # Examples +/// +/// ``` +/// use wireframe_testing::runtime_ownership_benchmarks::contended_fifo_pool_workloads; +/// +/// let workloads = contended_fifo_pool_workloads(); +/// assert_eq!(workloads[0].pool_size, 1); +/// assert_eq!(workloads[2].contenders, 17); +/// ``` +#[must_use] +pub fn contended_fifo_pool_workloads() -> [PoolWorkload; POOL_SIZES.len()] { + POOL_SIZES.map(|pool_size| PoolWorkload { + pool_size, + mode: AcquisitionMode::ContendedFifo, + contenders: DEFAULT_CONTENDERS.max(pool_size.saturating_add(1)), + ..PoolWorkload::default() + }) +} + /// Result of measuring one pool workload. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub struct PoolMeasurement { diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/mod.rs b/wireframe_testing/src/runtime_ownership_benchmarks/mod.rs index e58c4b947..ec335562d 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/mod.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/mod.rs @@ -58,6 +58,7 @@ pub use client_pool_benchmark_support::{ PoolHarness, PoolMeasurement, PoolWorkload, + contended_fifo_pool_workloads, measure_contended_latency, measure_pool_acquisition, }; From 7d776159d84d25e45eb103e3f4b467f212d4da61 Mon Sep 17 00:00:00 2001 From: leynos Date: Thu, 22 Oct 2026 22:53:14 +0200 Subject: [PATCH 02/21] Race pool permits on borrowed slot futures (#646) Probe the rotated slots without allocation before building a borrowed `FuturesUnordered` race. Drop losing acquisitions before rechecking shutdown so cancellation releases queued waiters and acquired permits. --- src/client/pool/core.rs | 92 ++++++++++++++++++---------- src/client/pool/scheduler/service.rs | 3 +- 2 files changed, 59 insertions(+), 36 deletions(-) diff --git a/src/client/pool/core.rs b/src/client/pool/core.rs index cba95e071..46663a509 100644 --- a/src/client/pool/core.rs +++ b/src/client/pool/core.rs @@ -2,14 +2,10 @@ use std::{ future::Future, - pin::Pin, - sync::{ - Arc, - atomic::{AtomicBool, AtomicUsize, Ordering}, - }, + sync::atomic::{AtomicBool, AtomicUsize, Ordering}, }; -use futures::future::select_all; +use futures::stream::{FuturesUnordered, StreamExt}; use tokio::sync::{Notify, OwnedSemaphorePermit}; use super::slot::PoolSlot; @@ -22,9 +18,6 @@ use crate::{ /// Result of selecting a slot while retaining its independent admission token. pub(crate) type SlotPermit = (usize, OwnedSemaphorePermit); -/// Owned future used by the existing permit race. -type SlotPermitFuture = Pin> + Send>>; - /// One shared root for the pool's fixed physical slots and shutdown state. pub(crate) struct PoolCore where @@ -138,8 +131,14 @@ where } } - /// Race slot permits and return the first available capacity or error. - pub(crate) async fn acquire_slot_permit(self: &Arc) -> Result { + /// Probe rotated slots before racing borrowed futures over blocked slots. + /// + /// A free slot at the rotated starting point wins before any waiters are + /// registered. If every slot is full, the race futures borrow the stable + /// slot storage; when polled, each future registers one FIFO waiter on its + /// semaphore. The serial service worker runs at most one race at a time for + /// a handle. + pub(crate) async fn acquire_slot_permit(&self) -> Result { let result = self.acquire_slot_permit_inner().await; let outcome = match &result { Ok(_) => PoolAcquisitionOutcome::Granted, @@ -152,30 +151,25 @@ where result } - /// Race one owned permit future per slot, keeping the core alive for each. - async fn acquire_slot_permit_inner(self: &Arc) -> Result { + /// Try rotated slots before registering blocked semaphore waiters. + async fn acquire_slot_permit_inner(&self) -> Result { if self.is_shutdown() || self.slots.is_empty() { return Err(ClientError::disconnected()); } - let waiters = self - .ordered_slot_indices() - .map(|slot_index| { - // Keep the existing owned-future race in this slice. The - // futures retain the core root instead of individual slots; - // the scoped-race follow-up can remove these per-slot clones. - let core = Arc::clone(self); - Box::pin(async move { - let slot = core - .slot(slot_index) - .ok_or_else(ClientError::disconnected)?; - let permit = slot.acquire_permit().await?; - Ok((slot_index, permit)) - }) as SlotPermitFuture - }) - .collect::>(); - let (result, ..) = select_all(waiters).await; - let (slot_index, permit) = result?; + let ordered_indices = self.ordered_slot_indices(); + let selected = ordered_indices.clone().find_map(|slot_index| { + self.slot(slot_index)? + .try_acquire_permit() + .map(|permit| (slot_index, permit)) + }); + + let (slot_index, permit) = if let Some(selected) = selected { + selected + } else { + self.acquire_from_blocked_slots(ordered_indices).await? + }; + if self.is_shutdown() { drop(permit); Err(ClientError::disconnected()) @@ -184,7 +178,34 @@ where } } - /// Rotate slot indices without allocating or cloning slot owners. + /// Race borrowed FIFO acquisitions after every immediate probe fails. + async fn acquire_from_blocked_slots( + &self, + ordered_indices: impl Iterator, + ) -> Result { + // FuturesUnordered does not promise poll order, so the caller probes + // first to make a free rotated starting slot deterministic. Each + // future borrows the core's stable slots. Dropping the stream cancels + // losing semaphore acquisitions and releases any assigned permits. + let mut pending = FuturesUnordered::new(); + for slot_index in ordered_indices { + let slot = self + .slot(slot_index) + .ok_or_else(ClientError::disconnected)?; + pending.push(async move { + let permit = slot.acquire_permit().await?; + Ok((slot_index, permit)) + }); + } + let result = pending + .next() + .await + .unwrap_or_else(|| Err(ClientError::disconnected())); + drop(pending); + result + } + + /// Rotate slot indices once for the immediate probe and blocked race. /// /// The cursor is read once per attempt, and only for a non-empty pool, so /// startup rotation stays cheap and an empty pool consumes no cursor value. @@ -193,7 +214,10 @@ where /// `len` only for power-of-two pool sizes. Reducing after the addition /// would therefore repeat one slot and skip another in a three- or /// five-slot pool once the cursor overflowed. - fn ordered_slot_indices(&self) -> impl Iterator + '_ { + /// + /// Each call advances the starting cursor exactly once, even when a slot + /// is selected by the immediate probe. + fn ordered_slot_indices(&self) -> impl Iterator + Clone + '_ { let len = self.slots.len(); let start = if len == 0 { 0 @@ -222,7 +246,7 @@ fn rotation_start(len: usize, cursor: usize) -> usize { /// `start` must be below `len`, which [`rotation_start`] guarantees, so the /// additions cannot overflow; `wrapping_rem` is the lint-permitted spelling of /// the remainder rather than an overflow workaround here. -fn rotation_indices(len: usize, start: usize) -> impl Iterator { +fn rotation_indices(len: usize, start: usize) -> impl Iterator + Clone { (0..len).map(move |offset| start.wrapping_add(offset).wrapping_rem(len.max(1))) } diff --git a/src/client/pool/scheduler/service.rs b/src/client/pool/scheduler/service.rs index 02bc024c3..6968c79bd 100644 --- a/src/client/pool/scheduler/service.rs +++ b/src/client/pool/scheduler/service.rs @@ -41,9 +41,8 @@ where core: Arc>, control: Arc>, ) { - let permit_race_core = Arc::clone(&core); let result = tokio::select! { - permit_result = permit_race_core.acquire_slot_permit() => { + permit_result = core.acquire_slot_permit() => { permit_result.and_then(|(slot_index, permit)| { if core.is_shutdown() { drop(permit); From 980b9949fadc58d713702a65f28784cd9718ffe4 Mon Sep 17 00:00:00 2001 From: leynos Date: Thu, 22 Oct 2026 23:59:10 +0200 Subject: [PATCH 03/21] Strengthen pool race mutation coverage (#646) Verify that a shut-down core rejects a new acquisition before it queues behind saturated slots. Record the shutdown check in the ADR verification notes and preserve the before/after permit-race measurements. --- ...lient-pool-scheduler-and-slot-ownership.md | 34 +++++++---- docs/runtime-ownership-baselines.md | 57 +++++++++++++++++++ src/client/pool/core_tests/permit_race.rs | 14 +++++ 3 files changed, 94 insertions(+), 11 deletions(-) diff --git a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md index 4c8a8aa0e..beeb864fa 100644 --- a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md +++ b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md @@ -541,12 +541,21 @@ are internal scaffolding removed before this ADR flips to Accepted. ### Phase 4: use scoped permit races -Remove the per-slot `Arc` clones and `'static` boxed slot futures -from the current permit race -([#646](https://github.com/leynos/wireframe/issues/646)). The current code -clones the core once for each slot future. The scoped-race rework deliberately -follows the actor phase so the races are written once against the actor's grant -loop rather than twice. +Use the fixed slot storage and index-based leases introduced by #645 to remove +the per-slot `Arc` clones and `'static` boxed futures from permit +acquisition ([#646](https://github.com/leynos/wireframe/issues/646)). Advance +the round-robin cursor once, probe the rotated slots without allocation, then +use `FuturesUnordered` only when every slot is blocked. Each blocked future +borrows its stable `PoolSlot` and registers a FIFO semaphore waiter when +polled. Dropping the stream cancels losing acquisitions and releases any +permits they acquired before the winner was selected. + +The immediate probe preserves deterministic start-slot behaviour because +`FuturesUnordered` does not promise poll order. It allocates no storage +proportional to pool size when a probe succeeds. A fully blocked race allocates +one `FuturesUnordered` task node per slot. The serial service worker permits at +most one active race per logical handle, and the scheduler keeps its existing +fairness policy. ### Phase 5: integrate existing safety work @@ -575,11 +584,14 @@ Complete or coordinate the following, plus any remaining useful scope from rather than passing. - Dropping all public handles without `close` eventually releases the pool core, including with the `JoinHandle` held by the non-shared close owner. -- Loom or deterministic state-machine tests cover enqueue, capacity, - cancellation, deregistration, and shutdown races, naming at minimum: the - pending-flag RAII clear when a permit race is dropped mid-`select!`, and a - permit won exactly as its waiter's receiver drops. No loom coverage exists - for the pool today; the only loom suite exercises the push queues. +- Deterministic current-thread tests manually poll pool permit races to cover + registration, simultaneous readiness, winner and loser permit ownership, + outer cancellation, and close. A saturated-core test verifies shutdown + rejects an acquisition before it queues behind slot waiters. This uses + Tokio's actual semaphore and scheduler primitives, whose queue registration + and drop behaviour a downstream Loom build cannot schedule. The Loom suites + cover push queues and connection-gauge transitions; they do not model the + client pool. - A panic injected into the scheduler task surfaces as `SchedulerFailed` with the required log event and metric, not as a silent hang or a clean `PoolClosed`. diff --git a/docs/runtime-ownership-baselines.md b/docs/runtime-ownership-baselines.md index 14172c106..db7ef5d60 100644 --- a/docs/runtime-ownership-baselines.md +++ b/docs/runtime-ownership-baselines.md @@ -647,6 +647,63 @@ eightfold to sixteenfold. Both figures move between captures by more than the correction accounts for, which is the same caution the allocation table states: at this scope, prefer the spawn count. +## Issue #646 permit-race comparison + +Both captures use `make bench-runtime-ownership` on the named machine and the +same pool workloads: 256 uncontended acquire/release operations per timing +sample, 20 Criterion samples, 20 contended bursts, and four iterations for +allocation observations. The pre-change run used the Phase 1 benchmark worktree +based on `ab630169`, with the benchmark scenarios already registered; the +production race still used boxed `select_all` futures. The after-change run used +`d29c51d`. The captures were taken on 2026-10-22. + +The uncontended scenarios use the pool's immediate `try_acquire_permit` path, +not the blocked race. Each timing cell gives the Criterion lower, estimate, and +upper bounds in microseconds. Those bounds show the observed jitter across the +20 samples; they are not per-function measurements. + +| Slots | Before latency (lower / estimate / upper, µs) | After latency (lower / estimate / upper, µs) | Before allocations | After allocations | +| ----- | --------------------------------------------- | -------------------------------------------- | ------------------ | ----------------- | +| 1 | 158.50 / 164.95 / 171.08 | 143.70 / 148.48 / 154.49 | 1222 | 1233 | +| 4 | 138.28 / 146.01 / 152.66 | 120.83 / 124.68 / 128.41 | 1407 | 1124 | +| 16 | 147.24 / 151.21 / 155.15 | 115.07 / 120.78 / 127.88 | 1616 | 1103 | + +Allocation counts cover four iterations of each burst. They include runtime and +benchmark-harness work and do not attribute allocations to +`acquire_slot_permit`. The source-level fast-path property is that +`acquire_slot_permit` allocates nothing when its probe succeeds. When every +slot is blocked, `FuturesUnordered` allocates one task node for each slot +future. + +The contended distribution records successful bursts only; `outcomes` reports +successful bursts out of 20. Each distribution cell gives median, p90, and +maximum in microseconds so the spread remains visible even when failures remove +bursts from the sample. + +| Slots | Contenders | Before outcomes | Before median / p90 / max (µs) | After outcomes | After median / p90 / max (µs) | +| ----- | ---------- | --------------- | ------------------------------ | -------------- | ----------------------------- | +| 1 | 16 | 20/20 | 269.404 / 536.208 / 1371.982 | 20/20 | 400.096 / 432.627 / 496.668 | +| 4 | 16 | 20/20 | 305.135 / 408.506 / 914.004 | 18/20 | 318.435 / 460.577 / 708.211 | +| 16 | 17 | 4/20 | 374.066 / 850.513 / 850.513 | 3/20 | 354.006 / 601.319 / 601.319 | + +The contended allocation observations cover four iterations and include +scheduler and harness allocations. Rows with failed contenders produced no +allocation count. + +| Slots | Contenders | Before allocations | After allocations | +| ----- | ---------- | ------------------ | ----------------- | +| 1 | 16 | 1839 | 2074 | +| 4 | 16 | 2177 | 1440 | +| 16 | 17 | unavailable | unavailable | + +The 16-slot contended run failed with `connection closed by peer` before and +after the change, so its latency cells describe only a small successful subset +and its allocation cell is unavailable. The same error appeared in two of 20 +size-4 bursts after the change. This slice therefore does not establish a +change in the spurious-disconnect rate described below. Treat the contended +latencies and process-global allocation counts as workload observations, not +per-function attribution or proof of a throughput change. + ## Known defects in contended pool acquisition Capturing the pool baseline surfaced two pre-existing defects in the acquire diff --git a/src/client/pool/core_tests/permit_race.rs b/src/client/pool/core_tests/permit_race.rs index 0e17985b0..ff01811ff 100644 --- a/src/client/pool/core_tests/permit_race.rs +++ b/src/client/pool/core_tests/permit_race.rs @@ -96,6 +96,20 @@ async fn empty_core_rejects_both_immediate_and_async_acquisition() { assert!(core.acquire_slot_permit().await.is_err()); } +#[rstest] +#[tokio::test(flavor = "current_thread")] +async fn shutdown_rejects_acquisition_before_waiting_for_slot_capacity() { + let core = test_core(2); + let held = hold_all_permits(&core); + core.shutdown(); + + let result = tokio::time::timeout(Duration::from_millis(100), core.acquire_slot_permit()).await; + + assert!(matches!(result, Ok(Err(_)))); + drop(held); + assert_all_slots_are_available(&core); +} + #[rstest] #[tokio::test] async fn async_probe_rotates_across_four_slots(four_slot_core: Arc) { From 1b33796fcb8068ebd32486c04ceaf095c3a01f45 Mon Sep 17 00:00:00 2001 From: leynos Date: Fri, 23 Oct 2026 00:33:16 +0200 Subject: [PATCH 04/21] Integrate permit races with pool rotation helpers (#646) Place permit-race coverage in the split `core_tests` module and use its shared fixtures. Record reuse of main's rotation helpers and preserve the bounded acquisition metric wrappers around immediate and asynchronous selection. --- docs/adr-013-client-pool-scheduler-and-slot-ownership.md | 7 +++++++ src/client/pool/core_tests/mod.rs | 6 ++++-- src/client/pool/core_tests/permit_race.rs | 2 +- src/client/pool/core_tests/sequence_tests.rs | 3 --- 4 files changed, 12 insertions(+), 6 deletions(-) diff --git a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md index beeb864fa..b79985ae3 100644 --- a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md +++ b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md @@ -557,6 +557,13 @@ one `FuturesUnordered` task node per slot. The serial service worker permits at most one active race per logical handle, and the scheduler keeps its existing fairness policy. +The race reuses main's `rotation_start` and `rotation_indices` helpers from +[#536](https://github.com/leynos/wireframe/issues/536). The iterator exposes +`Clone` so the probe and blocked race share one rotated order after exactly one +cursor advance, without materializing an index collection. The bounded +acquisition metric wrappers introduced alongside those helpers remain around +both immediate and asynchronous selection. + ### Phase 5: integrate existing safety work Complete or coordinate the following, plus any remaining useful scope from diff --git a/src/client/pool/core_tests/mod.rs b/src/client/pool/core_tests/mod.rs index 538407834..56ae32335 100644 --- a/src/client/pool/core_tests/mod.rs +++ b/src/client/pool/core_tests/mod.rs @@ -2,9 +2,11 @@ //! //! The suite is split by concern so each file stays reviewable: //! `support` holds the shared cores and fixtures, `sequence_tests` covers slot -//! identity and lifecycle, and `metric_tests` covers the labels the real -//! acquisition wrappers record. +//! identity and lifecycle, `permit_race` covers asynchronous selection and +//! cancellation, and `metric_tests` covers the labels the real acquisition +//! wrappers record. mod metric_tests; +mod permit_race; mod sequence_tests; mod support; diff --git a/src/client/pool/core_tests/permit_race.rs b/src/client/pool/core_tests/permit_race.rs index ff01811ff..b85a02984 100644 --- a/src/client/pool/core_tests/permit_race.rs +++ b/src/client/pool/core_tests/permit_race.rs @@ -5,7 +5,7 @@ use std::{sync::Arc, time::Duration}; use rstest::rstest; use tokio::sync::OwnedSemaphorePermit; -use super::{TestCore, four_slot_core, test_core}; +use super::support::{TestCore, four_slot_core, test_core}; fn hold_all_permits(core: &TestCore) -> Vec { core.slots diff --git a/src/client/pool/core_tests/sequence_tests.rs b/src/client/pool/core_tests/sequence_tests.rs index c9f4e7df9..f5ce43861 100644 --- a/src/client/pool/core_tests/sequence_tests.rs +++ b/src/client/pool/core_tests/sequence_tests.rs @@ -363,6 +363,3 @@ impl GeneratedSlotSequence { Ok(()) } } - -#[path = "core_tests/permit_race.rs"] -mod permit_race; From 3200cf3e0872f9d4d307663495691dd0ee2aebdf Mon Sep 17 00:00:00 2001 From: leynos Date: Fri, 23 Oct 2026 22:33:39 +0200 Subject: [PATCH 05/21] Document borrowed permit selection and isolate flaky size-16 case (#646) Add an accessible sequence diagram for the rotated probe and borrowed FuturesUnordered race. Clarify scheduler scope and benchmark measurement boundaries, and keep the size-16 FIFO workload registered while tracking its intermittent disconnect test under #647. --- ...lient-pool-scheduler-and-slot-ownership.md | 2 +- docs/runtime-ownership-baselines.md | 5 ++- docs/wireframe-client-design.md | 32 ++++++++++++++ src/client/pool/core.rs | 2 +- .../tests/contended.rs | 44 ++++++++++++++++--- 5 files changed, 74 insertions(+), 11 deletions(-) diff --git a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md index b79985ae3..6b9c547aa 100644 --- a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md +++ b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md @@ -554,7 +554,7 @@ The immediate probe preserves deterministic start-slot behaviour because `FuturesUnordered` does not promise poll order. It allocates no storage proportional to pool size when a probe succeeds. A fully blocked race allocates one `FuturesUnordered` task node per slot. The serial service worker permits at -most one active race per logical handle, and the scheduler keeps its existing +most one active race per pool scheduler, and the scheduler keeps its existing fairness policy. The race reuses main's `rotation_start` and `rotation_indices` helpers from diff --git a/docs/runtime-ownership-baselines.md b/docs/runtime-ownership-baselines.md index db7ef5d60..8ec4fb674 100644 --- a/docs/runtime-ownership-baselines.md +++ b/docs/runtime-ownership-baselines.md @@ -687,8 +687,9 @@ bursts from the sample. | 16 | 17 | 4/20 | 374.066 / 850.513 / 850.513 | 3/20 | 354.006 / 601.319 / 601.319 | The contended allocation observations cover four iterations and include -scheduler and harness allocations. Rows with failed contenders produced no -allocation count. +scheduler and harness allocations. The allocation window runs separately from +the timing bursts; a row has no allocation count only when its own +four-iteration allocation window fails. | Slots | Contenders | Before allocations | After allocations | | ----- | ---------- | ------------------ | ----------------- | diff --git a/docs/wireframe-client-design.md b/docs/wireframe-client-design.md index 9ad3cea18..c401f88b2 100644 --- a/docs/wireframe-client-design.md +++ b/docs/wireframe-client-design.md @@ -685,6 +685,38 @@ sequenceDiagram _Figure 2: Bounded waiter admission and saturation rejection._ +For screen readers: the sequence shows the scheduler asking `PoolCore` to +acquire a slot permit. The core obtains the rotated slot order and probes each +slot. If a permit is available, it returns immediately. If every slot is +blocked, it races futures that borrow the stable slots; the first completed +acquisition returns its slot index and permit, then dropping the stream cancels +the losing acquisitions. The core returns the slot and permit or a disconnected +error. + +```mermaid +sequenceDiagram + participant Scheduler + participant PoolCore + participant PoolSlot + participant FuturesUnordered + + Scheduler->>PoolCore: acquire_slot_permit() + PoolCore->>PoolCore: ordered_slot_indices() + PoolCore->>PoolSlot: try_acquire_permit() for rotated slots + alt permit available + PoolSlot-->>PoolCore: OwnedSemaphorePermit + else every slot blocked + PoolCore->>FuturesUnordered: push borrowed slot acquisitions + FuturesUnordered->>PoolSlot: acquire_permit() for each slot + PoolSlot-->>FuturesUnordered: first available permit + FuturesUnordered-->>PoolCore: slot_index and permit + PoolCore->>FuturesUnordered: drop(pending) + end + PoolCore-->>Scheduler: SlotPermit or disconnected +``` + +_Figure 3: Rotated slot-permit probing and the borrowed blocked-slot race._ + ## Decision record for 17.4.2 - Decision: document only the currently reachable troubleshooting surface for diff --git a/src/client/pool/core.rs b/src/client/pool/core.rs index 46663a509..0233a4a3a 100644 --- a/src/client/pool/core.rs +++ b/src/client/pool/core.rs @@ -137,7 +137,7 @@ where /// registered. If every slot is full, the race futures borrow the stable /// slot storage; when polled, each future registers one FIFO waiter on its /// semaphore. The serial service worker runs at most one race at a time for - /// a handle. + /// the pool scheduler. pub(crate) async fn acquire_slot_permit(&self) -> Result { let result = self.acquire_slot_permit_inner().await; let outcome = match &result { diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs index c0c2d62c8..33d96883b 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs @@ -39,12 +39,12 @@ fn fifo_baseline_workloads_cover_each_pool_size_and_can_contend() { ); } assert_eq!( - workloads[1].label(), - "contended_fifo_pool_4_contenders_16_rounds_8" - ); - assert_eq!( - workloads[2].label(), - "contended_fifo_pool_16_contenders_17_rounds_8" + workloads.map(PoolWorkload::label), + [ + "contended_fifo_pool_1_contenders_16_rounds_8", + "contended_fifo_pool_4_contenders_16_rounds_8", + "contended_fifo_pool_16_contenders_17_rounds_8", + ] ); } @@ -109,7 +109,6 @@ async fn contended_mode_needs_more_contenders_than_slots() { #[rstest::rstest] #[case::one_slot_fifo(1, 16, AcquisitionMode::ContendedFifo)] #[case::four_slots_fifo(4, 16, AcquisitionMode::ContendedFifo)] -#[case::sixteen_slots_fifo(16, 17, AcquisitionMode::ContendedFifo)] #[case::four_slots_round_robin(4, 16, AcquisitionMode::ContendedRoundRobin)] #[tokio::test] async fn contended_acquisition_observes_capacity_waits( @@ -145,3 +144,34 @@ async fn contended_acquisition_observes_capacity_waits( "contended contenders must still be granted leases" ); } + +/// Keep the size-16 workload available to benchmarks, but not in the test suite +/// until its tracked spurious disconnect is resolved. +#[tokio::test] +#[ignore = "FIXME: flaky 16-slot disconnect is tracked by issue #647"] +async fn sixteen_slots_fifo_observes_capacity_waits() { + let measurement = contended_burst_with_retry(PoolWorkload { + pool_size: 16, + mode: AcquisitionMode::ContendedFifo, + acquisitions: 8, + contenders: 17, + }) + .await + .expect("contended measurement failed in every retry attempt"); + + assert!( + measurement.capacity_waits >= 17, + "a burst that empties a 16-slot pool and holds it empty must park all 17 contenders, but \ + {} waits were recorded", + measurement.capacity_waits + ); + assert!(measurement.contended(), "the measurement must be contended"); + assert!( + measurement.scheduler_spawns >= 1, + "handing capacity to a parked waiter must enter the scheduler slow path at least once" + ); + assert!( + measurement.acquisitions >= 1, + "contended contenders must still be granted leases" + ); +} From 8ee76750e3fa3391d25717153eb06ecb3ea49e55 Mon Sep 17 00:00:00 2001 From: leynos Date: Tue, 27 Oct 2026 19:30:29 +0100 Subject: [PATCH 06/21] Observe and verify blocked permit races (#646) Record blocked race outcomes and durations, including cancellation and shutdown, so contention can be diagnosed without per-slot labels. Cover bounded readiness sets and retain capacity observations when a benchmark burst fails after parking waiters. Update the ownership guidance and enable the Tokio utility feature used by request streaming so metrics-disabled pool builds remain usable. --- ...lient-pool-scheduler-and-slot-ownership.md | 11 +++ docs/developers-guide.md | 40 ++++++--- src/client/pool/core.rs | 14 +++- src/client/pool/core_tests/metric_tests.rs | 60 ++++++++++++++ src/client/pool/core_tests/permit_race.rs | 44 ++++++++++ src/metrics/client_pool.rs | 81 ++++++++++++++++++- src/metrics/mod.rs | 14 +++- .../tests/contended.rs | 31 +++++++ 8 files changed, 277 insertions(+), 18 deletions(-) diff --git a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md index 6b9c547aa..c6651d7e0 100644 --- a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md +++ b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md @@ -557,6 +557,10 @@ one `FuturesUnordered` task node per slot. The serial service worker permits at most one active race per pool scheduler, and the scheduler keeps its existing fairness policy. +The service worker retains its `Arc` across `tokio::select!`, so the +acquisition can borrow the core while shutdown and closed-sender branches +remain active. This avoids a per-race or per-slot core clone. + The race reuses main's `rotation_start` and `rotation_indices` helpers from [#536](https://github.com/leynos/wireframe/issues/536). The iterator exposes `Clone` so the probe and blocked race share one rotated order after exactly one @@ -564,6 +568,10 @@ cursor advance, without materializing an index collection. The bounded acquisition metric wrappers introduced alongside those helpers remain around both immediate and asynchronous selection. +Blocked races record their starts, terminal outcomes, and durations. A guard +records cancellation when the scheduler drops an unfinished acquisition; the +immediate probe does not create a blocked-race observation. + ### Phase 5: integrate existing safety work Complete or coordinate the following, plus any remaining useful scope from @@ -599,6 +607,9 @@ Complete or coordinate the following, plus any remaining useful scope from and drop behaviour a downstream Loom build cannot schedule. The Loom suites cover push queues and connection-gauge transitions; they do not model the client pool. +- A bounded test enumerates every non-empty readiness set for pools of two, + three, and four slots. It checks that exactly one ready slot remains leased + and every losing slot returns its capacity. - A panic injected into the scheduler task surfaces as `SchedulerFailed` with the required log event and metric, not as a silent hang or a clean `PoolClosed`. diff --git a/docs/developers-guide.md b/docs/developers-guide.md index 9a301d64b..c18d81926 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -331,10 +331,10 @@ boundaries for Epic 635: phased client-pool ownership change. Phase 2 stores fixed slots by value in `PoolCore` and keeps scheduler state in `PoolSchedulerControl`; pools and handles retain both roots, while leases retain the core, a checked slot - index, a permit, and an optional control handle. The spawn-on-demand - scheduler and per-slot `Arc` permit-race clones remain in this - intermediate phase. The persistent owner task and scoped permit race belong - to later phases. + index, a permit, and an optional control handle. Phase 4 probes the rotated + slots and races borrowed acquisitions when all slots are blocked. The + spawn-on-demand scheduler remains; a persistent owner task is planned for + [#647](https://github.com/leynos/wireframe/issues/647). The first implementation slice is now in place: consuming `WireframeApp::prepare().await` returns an immutable `PreparedApp` or a typed @@ -521,14 +521,30 @@ of two: a wrapped `usize` differs from its true value by a multiple of reducing after the addition would repeat one slot and skip another once the cursor overflowed. An empty pool consumes no cursor value. The synchronous path borrows each slot in rotation order, copies the winning slot's index, and moves -its permit into `PooledClientLease::new`; the asynchronous path boxes one -permit future per index, each of which clones the shared `Arc` rather -than a slot. `src/client/pool/rotation_tests.rs` pins the ordering, the -overflow cases, and the one-increment-per-attempt cursor contract. Every -acquisition attempt also increments `wireframe_pool_acquisitions_total` with -bounded `mode` (`immediate` or `permit_race`) and `outcome` (`granted`, -`exhausted`, `shutdown`, or `error`) labels, so saturation and shutdown are -observable without adding per-slot or caller identifiers. +its permit into `PooledClientLease::new`. The asynchronous path borrows the +core and probes the same rotation without allocation proportional to pool size. +When all slots are blocked, `FuturesUnordered` races borrowed slot futures; +dropping the stream cancels the losers and releases their permits. Each blocked +race allocates a task node per slot. `src/client/pool/rotation_tests.rs` pins +the ordering, the overflow cases, and the one-increment-per-attempt cursor +contract. Every acquisition attempt also increments +`wireframe_pool_acquisitions_total` with bounded `mode` (`immediate` or +`permit_race`) and `outcome` (`granted`, `exhausted`, `shutdown`, or `error`) +labels, so saturation and shutdown are observable without adding per-slot or +caller identifiers. + +`wireframe_pool_permit_races_total` counts blocked race starts and their +`granted`, `shutdown`, `error`, or `cancelled` outcomes. +`wireframe_pool_permit_race_duration_seconds` records their duration with the +same terminal labels. Compare starts with terminal outcomes to inspect stuck +races, and graph duration and cancellation rates to investigate contention. The +metrics use no slot or caller identifiers; the immediate probe does not start a +blocked-race observation. + +The `tokio-util` dependency enables `io` directly because request streaming uses +`StreamReader`. This also keeps the pool build usable with the `metrics` +feature disabled, without relying on feature unification from the metrics +exporter. Each `SlotConnection` checkout starts armed. `dispatch_on_connection` disarms the guard only after the operation future returns; dropping an armed guard diff --git a/src/client/pool/core.rs b/src/client/pool/core.rs index 0233a4a3a..6ca5d3918 100644 --- a/src/client/pool/core.rs +++ b/src/client/pool/core.rs @@ -11,7 +11,7 @@ use tokio::sync::{Notify, OwnedSemaphorePermit}; use super::slot::PoolSlot; use crate::{ client::ClientError, - metrics::{PoolAcquisitionMode, PoolAcquisitionOutcome}, + metrics::{PoolAcquisitionMode, PoolAcquisitionOutcome, PoolPermitRaceGuard}, serializer::Serializer, }; @@ -139,7 +139,8 @@ where /// semaphore. The serial service worker runs at most one race at a time for /// the pool scheduler. pub(crate) async fn acquire_slot_permit(&self) -> Result { - let result = self.acquire_slot_permit_inner().await; + let mut race = None; + let result = self.acquire_slot_permit_inner(&mut race).await; let outcome = match &result { Ok(_) => PoolAcquisitionOutcome::Granted, Err(_) if self.is_shutdown() || self.slots.is_empty() => { @@ -148,11 +149,17 @@ where Err(_) => PoolAcquisitionOutcome::Error, }; crate::metrics::inc_pool_acquisition(PoolAcquisitionMode::PermitRace, outcome); + if let Some(race) = race { + race.finish(outcome); + } result } /// Try rotated slots before registering blocked semaphore waiters. - async fn acquire_slot_permit_inner(&self) -> Result { + async fn acquire_slot_permit_inner<'a>( + &'a self, + race: &mut Option>, + ) -> Result { if self.is_shutdown() || self.slots.is_empty() { return Err(ClientError::disconnected()); } @@ -167,6 +174,7 @@ where let (slot_index, permit) = if let Some(selected) = selected { selected } else { + *race = Some(PoolPermitRaceGuard::start(&self.shutdown)); self.acquire_from_blocked_slots(ordered_indices).await? }; diff --git a/src/client/pool/core_tests/metric_tests.rs b/src/client/pool/core_tests/metric_tests.rs index aa36b50d0..e6d6797a3 100644 --- a/src/client/pool/core_tests/metric_tests.rs +++ b/src/client/pool/core_tests/metric_tests.rs @@ -168,3 +168,63 @@ fn expect_pool_acquisition(observability: &mut ObservabilityHandle, mode: &str, ) .expect("the acquisition wrapper should record the expected label pair"); } + +/// Exercise blocked-race metrics through the real acquisition future. +#[cfg(feature = "metrics")] +#[rstest] +#[case::granted("granted", false)] +#[case::shutdown("shutdown", false)] +#[case::shutdown_drop("shutdown", true)] +#[case::cancelled("cancelled", true)] +#[test] +fn blocked_race_records_terminal_outcome_and_duration( + #[case] outcome: &str, + #[case] drop_pending: bool, +) { + let mut observability = ObservabilityHandle::new(); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build a current-thread runtime"); + + metrics::with_local_recorder(observability.recorder(), || { + runtime.block_on(async { + let core = test_core(1); + let mut held = core + .slot(0) + .expect("the only slot should exist") + .try_acquire_permit(); + assert!(held.is_some()); + { + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + assert!(futures::poll!(acquisition.as_mut()).is_pending()); + if outcome == "shutdown" { + core.shutdown(); + } + if !drop_pending { + drop(held.take()); + let result = acquisition.as_mut().await; + assert_eq!(result.is_ok(), outcome == "granted"); + } + } + drop(held); + }); + }); + observability.snapshot(); + observability + .assert_counter( + crate::metrics::POOL_PERMIT_RACES, + [("outcome", "started")], + 1, + ) + .expect("the blocked race should record its start"); + observability + .assert_counter(crate::metrics::POOL_PERMIT_RACES, [("outcome", outcome)], 1) + .expect("the blocked race should record its terminal outcome"); + observability + .assert_histogram_recorded( + crate::metrics::POOL_PERMIT_RACE_DURATION, + [("outcome", outcome)], + ) + .expect("the blocked race should record its duration"); +} diff --git a/src/client/pool/core_tests/permit_race.rs b/src/client/pool/core_tests/permit_race.rs index b85a02984..ddb3342f1 100644 --- a/src/client/pool/core_tests/permit_race.rs +++ b/src/client/pool/core_tests/permit_race.rs @@ -268,3 +268,47 @@ async fn cancelling_a_blocked_race_leaves_every_slot_usable() { assert!(slot_index < core.slots.len()); drop(permit); } + +/// Exhaust all non-empty readiness sets for small blocked races. +#[rstest] +#[case::two_slots(2)] +#[case::three_slots(3)] +#[case::four_slots(4)] +#[tokio::test(flavor = "current_thread")] +async fn blocked_race_preserves_capacity_for_every_ready_set(#[case] slot_count: usize) { + for ready in 1..(1_usize << slot_count) { + let core = test_core(slot_count); + let mut held = hold_all_permits(&core) + .into_iter() + .map(Some) + .collect::>(); + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + assert!(futures::poll!(acquisition.as_mut()).is_pending()); + + for (index, permit) in held.iter_mut().enumerate() { + if ready & (1 << index) != 0 { + drop(permit.take()); + } + } + let (winner, permit) = tokio::time::timeout(Duration::from_secs(1), acquisition) + .await + .expect("a ready slot must wake the race") + .expect("a ready slot must grant a permit"); + assert_ne!(ready & (1 << winner), 0); + + for index in 0..slot_count { + let available = core + .slot(index) + .expect("the model slot should exist") + .try_acquire_permit(); + assert_eq!( + available.is_some(), + index != winner && ready & (1 << index) != 0, + "unexpected capacity for slot {index}, ready set {ready:#b}" + ); + } + drop(permit); + drop(held); + assert_all_slots_are_available(&core); + } +} diff --git a/src/metrics/client_pool.rs b/src/metrics/client_pool.rs index 10e0ccf46..2a1225227 100644 --- a/src/metrics/client_pool.rs +++ b/src/metrics/client_pool.rs @@ -2,8 +2,15 @@ // The import is used only by the real (non-loom) increment functions; a // `loom` build compiles the tracing-only test matrix and would see it unused. +#[cfg(all(feature = "pool", not(loom)))] +use std::sync::atomic::AtomicBool; +#[cfg(all(feature = "metrics", feature = "pool", not(loom)))] +use std::sync::atomic::Ordering; +#[cfg(all(feature = "metrics", feature = "pool", not(loom)))] +use std::time::Instant; + #[cfg(all(feature = "metrics", feature = "pool", not(loom)))] -use metrics::counter; +use metrics::{counter, histogram}; /// Name of the counter tracking pooled sockets discarded before safe reuse. /// @@ -64,6 +71,78 @@ pub(crate) fn inc_pool_pending_response_discard() {} /// ``` pub const POOL_ACQUISITIONS: &str = "wireframe_pool_acquisitions_total"; +/// Counter of blocked permit races, labelled by `started`, `granted`, +/// `shutdown`, `error`, or `cancelled` outcome. +pub const POOL_PERMIT_RACES: &str = "wireframe_pool_permit_races_total"; + +/// Duration of blocked permit races in seconds, labelled by terminal outcome. +pub const POOL_PERMIT_RACE_DURATION: &str = "wireframe_pool_permit_race_duration_seconds"; + +/// Records cancellation when a blocked race is dropped before completion. +#[cfg(all(feature = "pool", not(loom)))] +pub(crate) struct PoolPermitRaceGuard<'a> { + /// Classify a dropped race as shutdown when pool closure caused it. + shutdown: &'a AtomicBool, + /// Measure the time spent waiting after the probe found no capacity. + #[cfg(feature = "metrics")] + started: Instant, + /// Prevent the drop handler from reporting a completed race twice. + #[cfg(feature = "metrics")] + finished: bool, +} + +#[cfg(all(feature = "pool", not(loom)))] +impl<'a> PoolPermitRaceGuard<'a> { + /// Start observing a race only after every immediate probe fails. + pub(crate) fn start(shutdown: &'a AtomicBool) -> Self { + #[cfg(feature = "metrics")] + counter!(POOL_PERMIT_RACES, "outcome" => "started").increment(1); + Self { + shutdown, + #[cfg(feature = "metrics")] + started: Instant::now(), + #[cfg(feature = "metrics")] + finished: false, + } + } + + /// Record a completed race after its shutdown recheck. + #[cfg(feature = "metrics")] + pub(crate) fn finish(mut self, outcome: PoolAcquisitionOutcome) { + self.record(outcome.as_str()); + self.finished = true; + } + + /// Finish the no-op observer when metrics are disabled. + #[cfg(not(feature = "metrics"))] + pub(crate) fn finish(self, _outcome: PoolAcquisitionOutcome) {} + + /// Record one terminal outcome and its elapsed wait. + #[cfg(feature = "metrics")] + fn record(&self, outcome: &'static str) { + counter!(POOL_PERMIT_RACES, "outcome" => outcome).increment(1); + histogram!(POOL_PERMIT_RACE_DURATION, "outcome" => outcome) + .record(self.started.elapsed().as_secs_f64()); + } +} + +#[cfg(all(feature = "pool", not(loom)))] +impl Drop for PoolPermitRaceGuard<'_> { + fn drop(&mut self) { + #[cfg(feature = "metrics")] + if !self.finished { + let outcome = if self.shutdown.load(Ordering::Acquire) { + "shutdown" + } else { + "cancelled" + }; + self.record(outcome); + } + #[cfg(not(feature = "metrics"))] + let _ = self.shutdown; + } +} + /// Bounded acquisition paths that decide how a pooled slot is selected. /// /// Gated like [`inc_pool_acquisition`]: under `cfg(loom)` the client module diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs index f54199d1c..2fb5b0c17 100644 --- a/src/metrics/mod.rs +++ b/src/metrics/mod.rs @@ -22,12 +22,22 @@ use metrics::{counter, gauge, histogram}; mod client_pool; #[cfg(all(feature = "pool", not(loom)))] -pub(crate) use client_pool::{PoolAcquisitionMode, PoolAcquisitionOutcome, inc_pool_acquisition}; +pub(crate) use client_pool::{ + PoolAcquisitionMode, + PoolAcquisitionOutcome, + PoolPermitRaceGuard, + inc_pool_acquisition, +}; mod decode_rejections; mod server_startup; mod server_supervisor; -pub use client_pool::{POOL_ACQUISITIONS, POOL_CONNECTION_DISCARDS}; +pub use client_pool::{ + POOL_ACQUISITIONS, + POOL_CONNECTION_DISCARDS, + POOL_PERMIT_RACE_DURATION, + POOL_PERMIT_RACES, +}; #[cfg(all(feature = "pool", not(loom)))] pub(crate) use client_pool::{ inc_pool_incomplete_operation_discard, diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs index 33d96883b..fc0b71fda 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs @@ -5,6 +5,7 @@ use crate::runtime_ownership_benchmarks::client_pool_benchmark_support::{ PoolMeasurement, PoolWorkload, contended_fifo_pool_workloads, + measure_contended_latency, measure_pool_acquisition, }; @@ -115,6 +116,36 @@ async fn contended_acquisition_observes_capacity_waits( #[case] pool_size: usize, #[case] contenders: usize, #[case] mode: AcquisitionMode, +) { + let workload = PoolWorkload { + pool_size, + mode, + acquisitions: 8, + contenders, + }; + let latency = measure_contended_latency(workload, 1) + .await + .expect("the sampling harness must finish"); + let churn = latency.churn(); + + assert!( + churn.capacity_waits >= u64::try_from(contenders).unwrap_or(u64::MAX), + "a burst that empties a {pool_size}-slot pool and holds it empty must park every one of \ + {contenders} contenders, but {} waits were recorded", + churn.capacity_waits + ); +} + +/// A successful contended burst must also deliver leases and exercise the scheduler. +#[rstest::rstest] +#[case::one_slot_fifo(1, 16, AcquisitionMode::ContendedFifo)] +#[case::four_slots_fifo(4, 16, AcquisitionMode::ContendedFifo)] +#[case::four_slots_round_robin(4, 16, AcquisitionMode::ContendedRoundRobin)] +#[tokio::test] +async fn contended_acquisition_grants_leases( + #[case] pool_size: usize, + #[case] contenders: usize, + #[case] mode: AcquisitionMode, ) { let measurement = contended_burst_with_retry(PoolWorkload { pool_size, From cf69683e2f82506e8cf57a590b282aa5c6ea4b03 Mon Sep 17 00:00:00 2001 From: leynos Date: Tue, 27 Oct 2026 20:35:58 +0100 Subject: [PATCH 07/21] Share contended burst assertions (#646) Keep the same capacity, scheduler, and lease checks for each workload while using one helper for the ignored sixteen-slot case. --- .../tests/contended.rs | 42 +++++++------------ 1 file changed, 14 insertions(+), 28 deletions(-) diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs index fc0b71fda..6ed3bea6c 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs @@ -146,15 +146,23 @@ async fn contended_acquisition_grants_leases( #[case] pool_size: usize, #[case] contenders: usize, #[case] mode: AcquisitionMode, -) { +) -> Result<(), String> { + assert_contended_grants(pool_size, contenders, mode).await +} + +/// Check successful bursts share the same capacity and progress contract. +async fn assert_contended_grants( + pool_size: usize, + contenders: usize, + mode: AcquisitionMode, +) -> Result<(), String> { let measurement = contended_burst_with_retry(PoolWorkload { pool_size, mode, acquisitions: 8, contenders, }) - .await - .expect("contended measurement failed in every retry attempt"); + .await?; assert!( measurement.capacity_waits >= u64::try_from(contenders).unwrap_or(u64::MAX), @@ -174,35 +182,13 @@ async fn contended_acquisition_grants_leases( measurement.acquisitions >= 1, "contended contenders must still be granted leases" ); + Ok(()) } /// Keep the size-16 workload available to benchmarks, but not in the test suite /// until its tracked spurious disconnect is resolved. #[tokio::test] #[ignore = "FIXME: flaky 16-slot disconnect is tracked by issue #647"] -async fn sixteen_slots_fifo_observes_capacity_waits() { - let measurement = contended_burst_with_retry(PoolWorkload { - pool_size: 16, - mode: AcquisitionMode::ContendedFifo, - acquisitions: 8, - contenders: 17, - }) - .await - .expect("contended measurement failed in every retry attempt"); - - assert!( - measurement.capacity_waits >= 17, - "a burst that empties a 16-slot pool and holds it empty must park all 17 contenders, but \ - {} waits were recorded", - measurement.capacity_waits - ); - assert!(measurement.contended(), "the measurement must be contended"); - assert!( - measurement.scheduler_spawns >= 1, - "handing capacity to a parked waiter must enter the scheduler slow path at least once" - ); - assert!( - measurement.acquisitions >= 1, - "contended contenders must still be granted leases" - ); +async fn sixteen_slots_fifo_observes_capacity_waits() -> Result<(), String> { + assert_contended_grants(16, 17, AcquisitionMode::ContendedFifo).await } From 14ff182cef77b12fc8a97ffb9df1150c57b7b5a1 Mon Sep 17 00:00:00 2001 From: leynos Date: Tue, 27 Oct 2026 20:50:33 +0100 Subject: [PATCH 08/21] Cover permit race transition schedules (#646) Enumerate readiness sets and release orders for small pools, including cancellation and shutdown after registration. Check winner ownership, losing capacity, and subsequent acquisitions on real semaphores. --- ...lient-pool-scheduler-and-slot-ownership.md | 7 +- src/client/pool/core_tests/mod.rs | 1 + src/client/pool/core_tests/permit_race.rs | 4 +- .../pool/core_tests/transition_model.rs | 143 ++++++++++++++++++ 4 files changed, 150 insertions(+), 5 deletions(-) create mode 100644 src/client/pool/core_tests/transition_model.rs diff --git a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md index c6651d7e0..3a0616011 100644 --- a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md +++ b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md @@ -607,9 +607,10 @@ Complete or coordinate the following, plus any remaining useful scope from and drop behaviour a downstream Loom build cannot schedule. The Loom suites cover push queues and connection-gauge transitions; they do not model the client pool. -- A bounded test enumerates every non-empty readiness set for pools of two, - three, and four slots. It checks that exactly one ready slot remains leased - and every losing slot returns its capacity. +- Bounded tests enumerate every non-empty readiness set and release order for + pools of two, three, and four slots. They check winner and loser capacity, + cancellation after release, shutdown after registration, and cancellation + after shutdown, using Tokio's actual semaphores. - A panic injected into the scheduler task surfaces as `SchedulerFailed` with the required log event and metric, not as a silent hang or a clean `PoolClosed`. diff --git a/src/client/pool/core_tests/mod.rs b/src/client/pool/core_tests/mod.rs index 56ae32335..9f63a412a 100644 --- a/src/client/pool/core_tests/mod.rs +++ b/src/client/pool/core_tests/mod.rs @@ -10,3 +10,4 @@ mod metric_tests; mod permit_race; mod sequence_tests; mod support; +mod transition_model; diff --git a/src/client/pool/core_tests/permit_race.rs b/src/client/pool/core_tests/permit_race.rs index ddb3342f1..c29ca7479 100644 --- a/src/client/pool/core_tests/permit_race.rs +++ b/src/client/pool/core_tests/permit_race.rs @@ -7,7 +7,7 @@ use tokio::sync::OwnedSemaphorePermit; use super::support::{TestCore, four_slot_core, test_core}; -fn hold_all_permits(core: &TestCore) -> Vec { +pub(super) fn hold_all_permits(core: &TestCore) -> Vec { core.slots .iter() .map(|slot| { @@ -29,7 +29,7 @@ fn hold_every_permit_except(core: &TestCore, free_index: usize) -> Vec, result: &mut Vec>) { + if remaining.is_empty() { + result.push(prefix.clone()); + return; + } + for (position, &slot) in remaining.iter().enumerate() { + let rest = remaining + .iter() + .enumerate() + .filter_map(|(index, &value)| (index != position).then_some(value)) + .collect::>(); + prefix.push(slot); + release_orders(&rest, prefix, result); + let _ = prefix.pop(); + } +} + +/// Terminal action after a set of slots has become ready. +#[derive(Clone, Copy, Debug)] +enum Terminal { + /// Let one ready acquisition complete. + Grant, + /// Drop the outer acquisition before it is polled again. + Cancel, + /// Publish shutdown before polling the ready acquisitions. + Shutdown, + /// Publish shutdown, then drop the still-pending acquisition. + ShutdownCancel, +} + +/// Exercise all non-empty readiness subsets and their release permutations. +#[rstest] +#[case::grant_two(2, Terminal::Grant)] +#[case::grant_three(3, Terminal::Grant)] +#[case::grant_four(4, Terminal::Grant)] +#[case::cancel_two(2, Terminal::Cancel)] +#[case::cancel_three(3, Terminal::Cancel)] +#[case::cancel_four(4, Terminal::Cancel)] +#[case::shutdown_two(2, Terminal::Shutdown)] +#[case::shutdown_three(3, Terminal::Shutdown)] +#[case::shutdown_four(4, Terminal::Shutdown)] +#[case::shutdown_cancel_two(2, Terminal::ShutdownCancel)] +#[case::shutdown_cancel_three(3, Terminal::ShutdownCancel)] +#[case::shutdown_cancel_four(4, Terminal::ShutdownCancel)] +#[tokio::test(flavor = "current_thread")] +async fn blocked_race_transition_schedules_preserve_capacity( + #[case] slot_count: usize, + #[case] terminal: Terminal, +) { + for ready in 1..(1_usize << slot_count) { + let slots = (0..slot_count) + .filter(|index| ready & (1 << *index) != 0) + .collect::>(); + let mut orders = Vec::new(); + release_orders(&slots, &mut Vec::new(), &mut orders); + for order in orders { + let core = test_core(slot_count); + let mut held = hold_all_permits(&core) + .into_iter() + .map(Some) + .collect::>(); + { + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + assert!(futures::poll!(acquisition.as_mut()).is_pending()); + for index in &order { + drop(held.get_mut(*index).expect("slot index must exist").take()); + } + match terminal { + Terminal::Grant => { + let (winner, permit) = + tokio::time::timeout(Duration::from_secs(1), acquisition.as_mut()) + .await + .expect("released capacity must wake the race") + .expect("a ready slot must grant a permit"); + assert_ne!(ready & (1 << winner), 0); + for index in 0..slot_count { + let available = core + .slot(index) + .expect("slot index must exist") + .try_acquire_permit(); + assert_eq!( + available.is_some(), + index != winner && ready & (1 << index) != 0, + "unexpected capacity for {terminal:?}, order {order:?}" + ); + } + drop(permit); + } + Terminal::Cancel => {} + Terminal::ShutdownCancel => core.shutdown(), + Terminal::Shutdown => { + core.shutdown(); + let error = + tokio::time::timeout(Duration::from_secs(1), acquisition.as_mut()) + .await + .expect("ready capacity must wake the shutdown race") + .expect_err("shutdown must reject the permit"); + assert!(matches!( + error, + ClientError::Wireframe(WireframeError::Io(ref source)) + if source.kind() == ErrorKind::UnexpectedEof + )); + } + } + } + drop(held); + assert_all_slots_are_available(&core); + if matches!(terminal, Terminal::Cancel) { + let (_, permit) = + tokio::time::timeout(Duration::from_secs(1), core.acquire_slot_permit()) + .await + .expect("a later race must not remain queued") + .expect("a later race must acquire capacity"); + drop(permit); + assert_all_slots_are_available(&core); + } + if matches!(terminal, Terminal::Shutdown | Terminal::ShutdownCancel) { + let error = core + .acquire_slot_permit() + .await + .expect_err("a shut-down core must reject later acquisitions"); + assert!(matches!( + error, + ClientError::Wireframe(WireframeError::Io(ref source)) + if source.kind() == ErrorKind::UnexpectedEof + )); + } + } + } +} From 3337e6abe63ffdd62748aad3d9f1516780c6bbe4 Mon Sep 17 00:00:00 2001 From: leynos Date: Tue, 27 Oct 2026 20:57:35 +0100 Subject: [PATCH 09/21] Simplify permit race schedule checks (#646) Separate schedule generation and outcome checks so the bounded model is readable while retaining its release, cancellation, and shutdown cases. --- .../pool/core_tests/transition_model.rs | 265 ++++++++++++------ 1 file changed, 182 insertions(+), 83 deletions(-) diff --git a/src/client/pool/core_tests/transition_model.rs b/src/client/pool/core_tests/transition_model.rs index d2a39e0d9..5f33e5c9f 100644 --- a/src/client/pool/core_tests/transition_model.rs +++ b/src/client/pool/core_tests/transition_model.rs @@ -1,14 +1,17 @@ //! Bounded release, cancellation, and shutdown schedules for blocked races. -use std::{io::ErrorKind, time::Duration}; +use std::{future::Future, io::ErrorKind, pin::Pin, task::Poll, time::Duration}; use rstest::rstest; use super::{ - permit_race::{assert_all_slots_are_available, hold_all_permits}, - support::test_core, + permit_race::hold_all_permits, + support::{TestCore, test_core}, +}; +use crate::{ + WireframeError, + client::{ClientError, pool::core::SlotPermit}, }; -use crate::{WireframeError, client::ClientError}; /// Enumerate each ordering of a ready subset without relying on poll order. fn release_orders(remaining: &[usize], prefix: &mut Vec, result: &mut Vec>) { @@ -28,6 +31,20 @@ fn release_orders(remaining: &[usize], prefix: &mut Vec, result: &mut Vec } } +/// Enumerate every non-empty ready set and each release order for that set. +fn readiness_orders(slot_count: usize) -> Vec<(usize, Vec)> { + let mut schedules = Vec::new(); + for ready in 1..(1_usize << slot_count) { + let slots = (0..slot_count) + .filter(|index| ready & (1 << *index) != 0) + .collect::>(); + let mut orders = Vec::new(); + release_orders(&slots, &mut Vec::new(), &mut orders); + schedules.extend(orders.into_iter().map(|order| (ready, order))); + } + schedules +} + /// Terminal action after a set of slots has become ready. #[derive(Clone, Copy, Debug)] enum Terminal { @@ -41,6 +58,157 @@ enum Terminal { ShutdownCancel, } +/// One bounded readiness and terminal-action scenario. +struct Schedule { + /// Number of stable slots in the core. + slot_count: usize, + /// Bit set of slots released after registration. + ready: usize, + /// Order in which the selected slot permits are released. + order: Vec, + /// Action taken after the selected slots are ready. + terminal: Terminal, +} + +/// Recognize the typed disconnected error used by pool shutdown. +fn check_disconnected(error: ClientError) -> Result<(), String> { + match error { + ClientError::Wireframe(WireframeError::Io(source)) + if source.kind() == ErrorKind::UnexpectedEof => + { + Ok(()) + } + other => Err(format!("expected disconnected error, got {other}")), + } +} + +/// Check every raw slot can grant a permit at the same time. +fn check_all_slots_available(core: &TestCore) -> Result<(), String> { + let permits = core + .slots + .iter() + .map(|slot| { + slot.try_acquire_permit() + .ok_or_else(|| "a slot lost capacity".to_owned()) + }) + .collect::, _>>()?; + drop(permits); + Ok(()) +} + +/// Verify the winner is ready and no losing future retains released capacity. +async fn check_grant( + core: &TestCore, + schedule: &Schedule, + acquisition: Pin<&mut F>, +) -> Result<(), String> +where + F: Future>, +{ + let (winner, permit) = tokio::time::timeout(Duration::from_secs(1), acquisition) + .await + .map_err(|error| format!("released capacity did not wake the race: {error}"))? + .map_err(|error| format!("a ready slot did not grant a permit: {error}"))?; + if schedule.ready & (1 << winner) == 0 { + return Err(format!( + "winner {winner} was not ready for {:?}", + schedule.order + )); + } + for index in 0..schedule.slot_count { + let available = core + .slot(index) + .ok_or("slot index must exist")? + .try_acquire_permit(); + let expected = index != winner && schedule.ready & (1 << index) != 0; + if available.is_some() != expected { + return Err(format!( + "unexpected capacity for {:?}, order {:?}", + schedule.terminal, schedule.order + )); + } + } + drop(permit); + Ok(()) +} + +/// Publish shutdown and verify a ready in-flight race rejects its permit. +async fn check_inflight_shutdown(core: &TestCore, acquisition: Pin<&mut F>) -> Result<(), String> +where + F: Future>, +{ + core.shutdown(); + let result = tokio::time::timeout(Duration::from_secs(1), acquisition) + .await + .map_err(|error| format!("ready capacity did not wake the shutdown race: {error}"))?; + match result { + Err(error) => check_disconnected(error), + Ok(_) => Err("shutdown granted a permit".to_owned()), + } +} + +/// Apply the terminal action while the pinned acquisition remains in scope. +async fn apply_terminal( + core: &TestCore, + schedule: &Schedule, + acquisition: Pin<&mut F>, +) -> Result<(), String> +where + F: Future>, +{ + match schedule.terminal { + Terminal::Grant => check_grant(core, schedule, acquisition).await, + Terminal::Cancel => Ok(()), + Terminal::Shutdown => check_inflight_shutdown(core, acquisition).await, + Terminal::ShutdownCancel => { + core.shutdown(); + Ok(()) + } + } +} + +/// Check a later acquisition after cancellation or shutdown. +async fn check_later_acquisition(core: &TestCore, terminal: Terminal) -> Result<(), String> { + match terminal { + Terminal::Grant => Ok(()), + Terminal::Cancel => { + let (_, permit) = + tokio::time::timeout(Duration::from_secs(1), core.acquire_slot_permit()) + .await + .map_err(|error| format!("a later race remained queued: {error}"))? + .map_err(|error| format!("a later race did not acquire capacity: {error}"))?; + drop(permit); + check_all_slots_available(core) + } + Terminal::Shutdown | Terminal::ShutdownCancel => match core.acquire_slot_permit().await { + Err(error) => check_disconnected(error), + Ok(_) => Err("a shut-down core granted a later permit".to_owned()), + }, + } +} + +/// Register one race, release its selected slots, and check the terminal state. +async fn run_schedule(schedule: &Schedule) -> Result<(), String> { + let core = test_core(schedule.slot_count); + let mut held = hold_all_permits(&core) + .into_iter() + .map(Some) + .collect::>(); + { + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + if !matches!(futures::poll!(acquisition.as_mut()), Poll::Pending) { + return Err("a saturated race must register and remain pending".to_owned()); + } + for index in &schedule.order { + drop(held.get_mut(*index).ok_or("slot index must exist")?.take()); + } + apply_terminal(&core, schedule, acquisition.as_mut()).await?; + } + drop(held); + check_all_slots_available(&core)?; + check_later_acquisition(&core, schedule.terminal).await +} + /// Exercise all non-empty readiness subsets and their release permutations. #[rstest] #[case::grant_two(2, Terminal::Grant)] @@ -60,84 +228,15 @@ async fn blocked_race_transition_schedules_preserve_capacity( #[case] slot_count: usize, #[case] terminal: Terminal, ) { - for ready in 1..(1_usize << slot_count) { - let slots = (0..slot_count) - .filter(|index| ready & (1 << *index) != 0) - .collect::>(); - let mut orders = Vec::new(); - release_orders(&slots, &mut Vec::new(), &mut orders); - for order in orders { - let core = test_core(slot_count); - let mut held = hold_all_permits(&core) - .into_iter() - .map(Some) - .collect::>(); - { - let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); - assert!(futures::poll!(acquisition.as_mut()).is_pending()); - for index in &order { - drop(held.get_mut(*index).expect("slot index must exist").take()); - } - match terminal { - Terminal::Grant => { - let (winner, permit) = - tokio::time::timeout(Duration::from_secs(1), acquisition.as_mut()) - .await - .expect("released capacity must wake the race") - .expect("a ready slot must grant a permit"); - assert_ne!(ready & (1 << winner), 0); - for index in 0..slot_count { - let available = core - .slot(index) - .expect("slot index must exist") - .try_acquire_permit(); - assert_eq!( - available.is_some(), - index != winner && ready & (1 << index) != 0, - "unexpected capacity for {terminal:?}, order {order:?}" - ); - } - drop(permit); - } - Terminal::Cancel => {} - Terminal::ShutdownCancel => core.shutdown(), - Terminal::Shutdown => { - core.shutdown(); - let error = - tokio::time::timeout(Duration::from_secs(1), acquisition.as_mut()) - .await - .expect("ready capacity must wake the shutdown race") - .expect_err("shutdown must reject the permit"); - assert!(matches!( - error, - ClientError::Wireframe(WireframeError::Io(ref source)) - if source.kind() == ErrorKind::UnexpectedEof - )); - } - } - } - drop(held); - assert_all_slots_are_available(&core); - if matches!(terminal, Terminal::Cancel) { - let (_, permit) = - tokio::time::timeout(Duration::from_secs(1), core.acquire_slot_permit()) - .await - .expect("a later race must not remain queued") - .expect("a later race must acquire capacity"); - drop(permit); - assert_all_slots_are_available(&core); - } - if matches!(terminal, Terminal::Shutdown | Terminal::ShutdownCancel) { - let error = core - .acquire_slot_permit() - .await - .expect_err("a shut-down core must reject later acquisitions"); - assert!(matches!( - error, - ClientError::Wireframe(WireframeError::Io(ref source)) - if source.kind() == ErrorKind::UnexpectedEof - )); - } - } + for (ready, order) in readiness_orders(slot_count) { + let schedule = Schedule { + slot_count, + ready, + order, + terminal, + }; + run_schedule(&schedule) + .await + .expect("the bounded race schedule must preserve capacity"); } } From ffafc69a3ac9fdca3ba0e8e458625d4341ab86b7 Mon Sep 17 00:00:00 2001 From: leynos Date: Tue, 27 Oct 2026 21:05:28 +0100 Subject: [PATCH 10/21] Bound shutdown schedule checks (#646) Use named terminal predicates and bound later shutdown acquisitions so a regression fails the model instead of leaving its test waiting. --- .../pool/core_tests/transition_model.rs | 66 +++++++++++-------- 1 file changed, 40 insertions(+), 26 deletions(-) diff --git a/src/client/pool/core_tests/transition_model.rs b/src/client/pool/core_tests/transition_model.rs index 5f33e5c9f..c74e3d779 100644 --- a/src/client/pool/core_tests/transition_model.rs +++ b/src/client/pool/core_tests/transition_model.rs @@ -58,6 +58,20 @@ enum Terminal { ShutdownCancel, } +impl Terminal { + /// Report whether the scenario should return a winning permit. + const fn is_grant(self) -> bool { matches!(self, Self::Grant) } + + /// Report whether cancellation should leave the core open. + const fn is_cancel(self) -> bool { matches!(self, Self::Cancel) } + + /// Report whether shutdown is published before the race is dropped. + const fn is_shutdown(self) -> bool { matches!(self, Self::Shutdown | Self::ShutdownCancel) } + + /// Report whether the shutdown race is polled after capacity is released. + const fn is_inflight_shutdown(self) -> bool { matches!(self, Self::Shutdown) } +} + /// One bounded readiness and terminal-action scenario. struct Schedule { /// Number of stable slots in the core. @@ -132,12 +146,11 @@ where Ok(()) } -/// Publish shutdown and verify a ready in-flight race rejects its permit. -async fn check_inflight_shutdown(core: &TestCore, acquisition: Pin<&mut F>) -> Result<(), String> +/// Verify a ready in-flight race rejects its permit after shutdown. +async fn check_inflight_shutdown(acquisition: Pin<&mut F>) -> Result<(), String> where F: Future>, { - core.shutdown(); let result = tokio::time::timeout(Duration::from_secs(1), acquisition) .await .map_err(|error| format!("ready capacity did not wake the shutdown race: {error}"))?; @@ -156,34 +169,35 @@ async fn apply_terminal( where F: Future>, { - match schedule.terminal { - Terminal::Grant => check_grant(core, schedule, acquisition).await, - Terminal::Cancel => Ok(()), - Terminal::Shutdown => check_inflight_shutdown(core, acquisition).await, - Terminal::ShutdownCancel => { - core.shutdown(); - Ok(()) - } + if schedule.terminal.is_shutdown() { + core.shutdown(); + } + if schedule.terminal.is_grant() { + return check_grant(core, schedule, acquisition).await; } + if schedule.terminal.is_inflight_shutdown() { + return check_inflight_shutdown(acquisition).await; + } + Ok(()) } /// Check a later acquisition after cancellation or shutdown. async fn check_later_acquisition(core: &TestCore, terminal: Terminal) -> Result<(), String> { - match terminal { - Terminal::Grant => Ok(()), - Terminal::Cancel => { - let (_, permit) = - tokio::time::timeout(Duration::from_secs(1), core.acquire_slot_permit()) - .await - .map_err(|error| format!("a later race remained queued: {error}"))? - .map_err(|error| format!("a later race did not acquire capacity: {error}"))?; - drop(permit); - check_all_slots_available(core) - } - Terminal::Shutdown | Terminal::ShutdownCancel => match core.acquire_slot_permit().await { - Err(error) => check_disconnected(error), - Ok(_) => Err("a shut-down core granted a later permit".to_owned()), - }, + if terminal.is_grant() { + return Ok(()); + } + let result = tokio::time::timeout(Duration::from_secs(1), core.acquire_slot_permit()) + .await + .map_err(|error| format!("a later race remained queued: {error}"))?; + if terminal.is_cancel() { + let (_, permit) = + result.map_err(|error| format!("a later race did not acquire capacity: {error}"))?; + drop(permit); + return check_all_slots_available(core); + } + match result { + Err(error) => check_disconnected(error), + Ok(_) => Err("a shut-down core granted a later permit".to_owned()), } } From 2df4b5a282daa8365f43672e69098d8cc7d1f914 Mon Sep 17 00:00:00 2001 From: leynos Date: Wed, 28 Oct 2026 00:43:17 +0100 Subject: [PATCH 11/21] Make permit race timing testable (#646) Inject a borrowed monotonic clock so recorder tests can verify blocked race durations exactly without changing the production ownership model. Document the public metrics and clarify asynchronous acquisition labels. --- ...lient-pool-scheduler-and-slot-ownership.md | 4 +- docs/runtime-ownership-baselines.md | 4 +- docs/users-guide.md | 17 +++- src/metrics/client_pool.rs | 84 +++++++++++++++++-- 4 files changed, 99 insertions(+), 10 deletions(-) diff --git a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md index 3a0616011..fa79add62 100644 --- a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md +++ b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md @@ -570,7 +570,9 @@ both immediate and asynchronous selection. Blocked races record their starts, terminal outcomes, and durations. A guard records cancellation when the scheduler drops an unfinished acquisition; the -immediate probe does not create a blocked-race observation. +immediate probe does not create a blocked-race observation. The guard borrows a +monotonic clock: production uses `Instant`, and recorder-backed tests inject a +deterministic clock to verify the duration value. ### Phase 5: integrate existing safety work diff --git a/docs/runtime-ownership-baselines.md b/docs/runtime-ownership-baselines.md index 8ec4fb674..bb5f32488 100644 --- a/docs/runtime-ownership-baselines.md +++ b/docs/runtime-ownership-baselines.md @@ -677,8 +677,8 @@ future. The contended distribution records successful bursts only; `outcomes` reports successful bursts out of 20. Each distribution cell gives median, p90, and -maximum in microseconds so the spread remains visible even when failures remove -bursts from the sample. +maximum in microseconds, so the spread remains visible even when failures +remove bursts from the sample. | Slots | Contenders | Before outcomes | Before median / p90 / max (µs) | After outcomes | After median / p90 / max (µs) | | ----- | ---------- | --------------- | ------------------------------ | -------------- | ----------------------------- | diff --git a/docs/users-guide.md b/docs/users-guide.md index 4c1bf9222..63b30dccf 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -1939,7 +1939,8 @@ When the `metrics` feature is enabled, every acquisition attempt increments activity cannot create new time series: - `mode` records which path made the attempt: `immediate` for the synchronous - fast path, or `permit_race` for the asynchronous race across slots. + fast path, or `permit_race` for asynchronous acquisition, including its + immediate probe and any subsequent race across slots. - `outcome` records the result: `granted`, `exhausted`, `shutdown`, or `error`. `exhausted` only applies to `immediate`; the asynchronous path either gains a @@ -1952,6 +1953,20 @@ rising `outcome="exhausted"` rate means acquisitions are arriving faster than the configured permits can serve them, and a rising `outcome="error"` rate on the `permit_race` mode means slots are failing instead of granting capacity. +When the asynchronous probe finds every slot busy, +`wireframe_pool_permit_races_total` records a blocked race with +`outcome="started"`. It records one terminal outcome: `granted`, `shutdown`, +`error`, or `cancelled` when the acquisition is dropped before completion. +`wireframe_pool_permit_race_duration_seconds` is a histogram of blocked-race +waits in seconds with those terminal outcome labels. Immediate probes that find +capacity do not start a blocked-race observation. + +Compare the rates of starts and terminal outcomes to investigate races that +remain active, and graph duration and cancellation rates to understand +contention. A `shutdown` outcome indicates pool closure; `cancelled` indicates +an acquisition dropped while the pool was still open. These metrics use fixed +labels and do not identify slots or callers. + ### Client preamble exchange The client builder supports an optional preamble exchange before framing diff --git a/src/metrics/client_pool.rs b/src/metrics/client_pool.rs index 2a1225227..2001f9112 100644 --- a/src/metrics/client_pool.rs +++ b/src/metrics/client_pool.rs @@ -78,6 +78,10 @@ pub const POOL_PERMIT_RACES: &str = "wireframe_pool_permit_races_total"; /// Duration of blocked permit races in seconds, labelled by terminal outcome. pub const POOL_PERMIT_RACE_DURATION: &str = "wireframe_pool_permit_race_duration_seconds"; +/// Production clock used to measure blocked races. +#[cfg(all(feature = "metrics", feature = "pool", not(loom)))] +static POOL_PERMIT_RACE_CLOCK: fn() -> Instant = Instant::now; + /// Records cancellation when a blocked race is dropped before completion. #[cfg(all(feature = "pool", not(loom)))] pub(crate) struct PoolPermitRaceGuard<'a> { @@ -86,6 +90,9 @@ pub(crate) struct PoolPermitRaceGuard<'a> { /// Measure the time spent waiting after the probe found no capacity. #[cfg(feature = "metrics")] started: Instant, + /// Monotonic time source, borrowed for the duration of the race. + #[cfg(feature = "metrics")] + clock: &'a (dyn Fn() -> Instant + Send + Sync), /// Prevent the drop handler from reporting a completed race twice. #[cfg(feature = "metrics")] finished: bool, @@ -96,12 +103,26 @@ impl<'a> PoolPermitRaceGuard<'a> { /// Start observing a race only after every immediate probe fails. pub(crate) fn start(shutdown: &'a AtomicBool) -> Self { #[cfg(feature = "metrics")] + { + Self::start_with_clock(shutdown, &POOL_PERMIT_RACE_CLOCK) + } + #[cfg(not(feature = "metrics"))] + { + Self { shutdown } + } + } + + /// Begin a blocked race using a supplied monotonic clock. + #[cfg(feature = "metrics")] + fn start_with_clock( + shutdown: &'a AtomicBool, + clock: &'a (dyn Fn() -> Instant + Send + Sync), + ) -> Self { counter!(POOL_PERMIT_RACES, "outcome" => "started").increment(1); Self { shutdown, - #[cfg(feature = "metrics")] - started: Instant::now(), - #[cfg(feature = "metrics")] + started: clock(), + clock, finished: false, } } @@ -121,8 +142,11 @@ impl<'a> PoolPermitRaceGuard<'a> { #[cfg(feature = "metrics")] fn record(&self, outcome: &'static str) { counter!(POOL_PERMIT_RACES, "outcome" => outcome).increment(1); - histogram!(POOL_PERMIT_RACE_DURATION, "outcome" => outcome) - .record(self.started.elapsed().as_secs_f64()); + histogram!(POOL_PERMIT_RACE_DURATION, "outcome" => outcome).record( + (self.clock)() + .saturating_duration_since(self.started) + .as_secs_f64(), + ); } } @@ -152,7 +176,9 @@ impl Drop for PoolPermitRaceGuard<'_> { pub(crate) enum PoolAcquisitionMode { /// The synchronous fast path tries each slot once without waiting. Immediate, - /// The asynchronous path races one permit future per slot. + /// The asynchronous path probes rotated slots, then races borrowed permit + /// futures only when all slots are blocked. [`POOL_PERMIT_RACES`] counts + /// the blocked races separately. PermitRace, } @@ -221,14 +247,21 @@ pub(crate) fn inc_pool_acquisition(_mode: PoolAcquisitionMode, _outcome: PoolAcq mod tests { //! Tests for bounded client-pool discard and acquisition labels. + use std::{ + sync::atomic::{AtomicBool, AtomicU64, Ordering}, + time::{Duration, Instant}, + }; + use metrics_util::debugging::{DebugValue, DebuggingRecorder}; use rstest::rstest; use super::{ POOL_ACQUISITIONS, POOL_CONNECTION_DISCARDS, + POOL_PERMIT_RACE_DURATION, PoolAcquisitionMode, PoolAcquisitionOutcome, + PoolPermitRaceGuard, inc_pool_acquisition, inc_pool_incomplete_operation_discard, inc_pool_pending_response_discard, @@ -305,4 +338,43 @@ mod tests { "acquisition metric for {mode:?}/{outcome:?} was not recorded" ); } + + #[rstest] + #[case::granted(Some(PoolAcquisitionOutcome::Granted), "granted")] + #[case::cancelled(None, "cancelled")] + fn blocked_race_records_duration_from_injected_clock( + #[case] outcome: Option, + #[case] expected_label: &str, + ) { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let shutdown = AtomicBool::new(false); + let elapsed = AtomicU64::new(0); + let origin = Instant::now(); + let clock = || origin + Duration::from_secs(elapsed.load(Ordering::Relaxed)); + + ::metrics::with_local_recorder(&recorder, || { + let guard = PoolPermitRaceGuard::start_with_clock(&shutdown, &clock); + elapsed.store(7, Ordering::Relaxed); + if let Some(outcome) = outcome { + guard.finish(outcome); + } else { + drop(guard); + } + }); + + let found = snapshotter.snapshot().into_vec().iter().any(|(key, _, _, value)| { + key.key().name() == POOL_PERMIT_RACE_DURATION + && key.key().labels().any(|label| { + label.key() == "outcome" && label.value() == expected_label + }) + && matches!(value, DebugValue::Histogram(samples) + if samples.len() == 1 + && samples.first().is_some_and(|sample| sample.0.to_bits() == 7.0_f64.to_bits())) + }); + assert!( + found, + "the histogram must record the injected seven-second duration" + ); + } } From 2338ffe97ee459e3268a83df7d2772f21d2f6665 Mon Sep 17 00:00:00 2001 From: leynos Date: Wed, 28 Oct 2026 05:33:56 +0100 Subject: [PATCH 12/21] Record upstream pool integration Document the module moves and dependency feature already supplied by main during the permit-race rebase. --- docs/adr-013-client-pool-scheduler-and-slot-ownership.md | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md index fa79add62..382b0fd3b 100644 --- a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md +++ b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md @@ -568,6 +568,11 @@ cursor advance, without materializing an index collection. The bounded acquisition metric wrappers introduced alongside those helpers remain around both immediate and asynchronous selection. +After main moved self-named modules in #798, the scheduler and metrics entry +points remain in their respective `mod.rs` files. Main also enables the +`tokio-util` `io` feature, so the metrics-disabled pool build retains that +upstream dependency setting without a separate branch change. + Blocked races record their starts, terminal outcomes, and durations. A guard records cancellation when the scheduler drops an unfinished acquisition; the immediate probe does not create a blocked-race observation. The guard borrows a From a66d588b0bf9ed9f5e867959b5f30b9fb2f1c837 Mon Sep 17 00:00:00 2001 From: leynos Date: Wed, 28 Oct 2026 05:45:08 +0100 Subject: [PATCH 13/21] Strengthen permit race validation Exercise free probes and staged slot releases, and check the pool without metrics in CI. Stabilize the contended workload assertion and clarify the integration notes. --- .github/workflows/ci.yml | 2 + Makefile | 5 ++- ...lient-pool-scheduler-and-slot-ownership.md | 9 ++-- src/client/pool/core_tests/metric_tests.rs | 29 +++++++++++++ .../pool/core_tests/transition_model.rs | 42 +++++++++++++++++++ .../tests/contended.rs | 14 +++---- 6 files changed, 87 insertions(+), 14 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 65f9001a4..eb31a57f4 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -134,6 +134,8 @@ jobs: # about whether the gate ran. - name: Feature-gate compile checks run: make test-feature-gate + - name: Pool without metrics compile check + run: make test-pool-no-metrics # Coverage runs on pull requests only and drives the ratchet gate. # Main-branch coverage — the CodeScene upload plus the authoritative # ratchet baseline every pull request compares against — is produced diff --git a/Makefile b/Makefile index 6302483bf..37ba4be28 100644 --- a/Makefile +++ b/Makefile @@ -1,4 +1,4 @@ -.PHONY: help all clean test test-doc test-ui test-feature-gate bless-ui test-workflow-contracts test-loom doctest-benchmark +.PHONY: help all clean test test-doc test-ui test-feature-gate test-pool-no-metrics bless-ui test-workflow-contracts test-loom doctest-benchmark .PHONY: bench-codec bench-runtime-ownership bench build dev-build dev-test release lint fmt check-fmt markdownlint nixie typecheck .PHONY: spelling .PHONY: install-kani check-kani-version install-verus run-verus test-verification kani \ @@ -123,6 +123,9 @@ test-ui: ## Run trybuild compile-time API tests test-feature-gate: ## Verify feature-gated APIs are actually absent when disabled ./scripts/check-feature-gate.sh +test-pool-no-metrics: ## Check the pool builds without metrics + $(CARGO) $(DEV_FAST) check --lib --no-default-features --features pool,serializer-bincode $(BUILD_JOBS) + # Developer-only: review every changed snapshot; this target must not run in CI. bless-ui: ## Refresh trybuild snapshots for deliberate fixture changes TRYBUILD=overwrite $(CARGO) test --test compile_error diff --git a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md index 382b0fd3b..9fc850597 100644 --- a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md +++ b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md @@ -561,17 +561,16 @@ The service worker retains its `Arc` across `tokio::select!`, so the acquisition can borrow the core while shutdown and closed-sender branches remain active. This avoids a per-race or per-slot core clone. -The race reuses main's `rotation_start` and `rotation_indices` helpers from +The race reuses the `rotation_start` and `rotation_indices` helpers from [#536](https://github.com/leynos/wireframe/issues/536). The iterator exposes `Clone` so the probe and blocked race share one rotated order after exactly one cursor advance, without materializing an index collection. The bounded acquisition metric wrappers introduced alongside those helpers remain around both immediate and asynchronous selection. -After main moved self-named modules in #798, the scheduler and metrics entry -points remain in their respective `mod.rs` files. Main also enables the -`tokio-util` `io` feature, so the metrics-disabled pool build retains that -upstream dependency setting without a separate branch change. +The scheduler and metrics entry points use their respective `mod.rs` files, +following the module layout established by #798. The metrics-disabled pool +build uses the existing `tokio-util` `io` feature. Blocked races record their starts, terminal outcomes, and durations. A guard records cancellation when the scheduler drops an unfinished acquisition; the diff --git a/src/client/pool/core_tests/metric_tests.rs b/src/client/pool/core_tests/metric_tests.rs index e6d6797a3..6d029281b 100644 --- a/src/client/pool/core_tests/metric_tests.rs +++ b/src/client/pool/core_tests/metric_tests.rs @@ -169,6 +169,35 @@ fn expect_pool_acquisition(observability: &mut ObservabilityHandle, mode: &str, .expect("the acquisition wrapper should record the expected label pair"); } +/// A free rotated probe must not start a blocked race. +#[cfg(feature = "metrics")] +#[test] +fn free_async_probe_skips_blocked_race() { + let mut observability = ObservabilityHandle::new(); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("build a current-thread runtime"); + metrics::with_local_recorder(observability.recorder(), || { + runtime.block_on(async { + let core = test_core(4); + for expected in 0..4 { + let (index, permit) = core.acquire_slot_permit().await.expect("free slot grants"); + assert_eq!(index, expected); + drop(permit); + } + }); + }); + observability.snapshot(); + observability + .assert_counter( + crate::metrics::POOL_PERMIT_RACES, + [("outcome", "started")], + 0, + ) + .expect("free probes must not start a blocked race"); +} + /// Exercise blocked-race metrics through the real acquisition future. #[cfg(feature = "metrics")] #[rstest] diff --git a/src/client/pool/core_tests/transition_model.rs b/src/client/pool/core_tests/transition_model.rs index c74e3d779..d81fc4bca 100644 --- a/src/client/pool/core_tests/transition_model.rs +++ b/src/client/pool/core_tests/transition_model.rs @@ -254,3 +254,45 @@ async fn blocked_race_transition_schedules_preserve_capacity( .expect("the bounded race schedule must preserve capacity"); } } + +/// Poll after one release while the remaining slot permits stay held. +#[rstest] +#[case(2)] +#[case(4)] +#[tokio::test(flavor = "current_thread")] +async fn first_released_slot_wins_before_later_releases(#[case] slot_count: usize) { + for first in 0..slot_count { + let core = test_core(slot_count); + let mut held = hold_all_permits(&core) + .into_iter() + .map(Some) + .collect::>(); + let mut acquisition = std::pin::pin!(core.acquire_slot_permit()); + assert!(futures::poll!(acquisition.as_mut()).is_pending()); + drop(held.get_mut(first).expect("slot permit exists").take()); + let (winner, permit) = tokio::time::timeout(Duration::from_secs(1), acquisition.as_mut()) + .await + .expect("released slot must wake the race") + .expect("released slot grants"); + assert_eq!(winner, first); + assert!( + core.slot(first) + .expect("winner exists") + .try_acquire_permit() + .is_none() + ); + drop(held); + for index in 0..slot_count { + if index != winner { + assert!( + core.slot(index) + .expect("slot exists") + .try_acquire_permit() + .is_some() + ); + } + } + drop(permit); + check_all_slots_available(&core).expect("every slot must retain capacity"); + } +} diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs index 6ed3bea6c..b295733e7 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs @@ -5,7 +5,6 @@ use crate::runtime_ownership_benchmarks::client_pool_benchmark_support::{ PoolMeasurement, PoolWorkload, contended_fifo_pool_workloads, - measure_contended_latency, measure_pool_acquisition, }; @@ -123,16 +122,15 @@ async fn contended_acquisition_observes_capacity_waits( acquisitions: 8, contenders, }; - let latency = measure_contended_latency(workload, 1) + let measurement = contended_burst_with_retry(workload) .await - .expect("the sampling harness must finish"); - let churn = latency.churn(); + .expect("a contended burst must finish"); assert!( - churn.capacity_waits >= u64::try_from(contenders).unwrap_or(u64::MAX), + measurement.capacity_waits >= u64::try_from(contenders).unwrap_or(u64::MAX), "a burst that empties a {pool_size}-slot pool and holds it empty must park every one of \ {contenders} contenders, but {} waits were recorded", - churn.capacity_waits + measurement.capacity_waits ); } @@ -188,7 +186,7 @@ async fn assert_contended_grants( /// Keep the size-16 workload available to benchmarks, but not in the test suite /// until its tracked spurious disconnect is resolved. #[tokio::test] -#[ignore = "FIXME: flaky 16-slot disconnect is tracked by issue #647"] -async fn sixteen_slots_fifo_observes_capacity_waits() -> Result<(), String> { +#[ignore = "FIXME: flaky 16-slot disconnect: https://github.com/leynos/wireframe/issues/647"] +async fn sixteen_slots_fifo_grants_leases() -> Result<(), String> { assert_contended_grants(16, 17, AcquisitionMode::ContendedFifo).await } From 8d2df33d8f12088ee5c2710131b76d90fd593ac5 Mon Sep 17 00:00:00 2001 From: leynos Date: Wed, 28 Oct 2026 06:12:32 +0100 Subject: [PATCH 14/21] Gate metric labels on metrics support Compile metric-only label helpers only with metrics enabled, and enforce warning-free pool checks in the local CI target. --- Makefile | 2 +- src/metrics/mod.rs | 2 ++ src/metrics/server_startup.rs | 2 +- 3 files changed, 4 insertions(+), 2 deletions(-) diff --git a/Makefile b/Makefile index 37ba4be28..7ccf86267 100644 --- a/Makefile +++ b/Makefile @@ -124,7 +124,7 @@ test-feature-gate: ## Verify feature-gated APIs are actually absent when disable ./scripts/check-feature-gate.sh test-pool-no-metrics: ## Check the pool builds without metrics - $(CARGO) $(DEV_FAST) check --lib --no-default-features --features pool,serializer-bincode $(BUILD_JOBS) + RUSTFLAGS="$(DEV_WARNING_FLAGS)" $(CARGO) $(DEV_FAST) check --lib --no-default-features --features pool,serializer-bincode $(BUILD_JOBS) # Developer-only: review every changed snapshot; this target must not run in CI. bless-ui: ## Refresh trybuild snapshots for deliberate fixture changes diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs index 2fb5b0c17..14796f560 100644 --- a/src/metrics/mod.rs +++ b/src/metrics/mod.rs @@ -162,6 +162,7 @@ pub(crate) enum PreparationOutcome { Failure, } +#[cfg(feature = "metrics")] impl PreparationOutcome { /// Return the stable metric label value for this preparation outcome. pub(crate) const fn as_str(self) -> &'static str { @@ -181,6 +182,7 @@ pub enum Direction { Outbound, } +#[cfg(feature = "metrics")] impl Direction { /// Supplies the bounded metric label value, avoiding call-site strings /// that could make inbound and outbound observations incomparable. diff --git a/src/metrics/server_startup.rs b/src/metrics/server_startup.rs index cd6ef4f48..aadcc131f 100644 --- a/src/metrics/server_startup.rs +++ b/src/metrics/server_startup.rs @@ -52,7 +52,7 @@ pub(crate) enum ServerStartupOutcome { Cancelled, } -#[cfg(not(loom))] +#[cfg(all(feature = "metrics", not(loom)))] impl ServerStartupOutcome { /// Return the stable metric label value for this outcome. pub(crate) const fn as_str(self) -> &'static str { From 6a8e59e4a14fa8b190bc4fb59744bb776667db75 Mon Sep 17 00:00:00 2001 From: leynos Date: Wed, 28 Oct 2026 10:16:24 +0100 Subject: [PATCH 15/21] Clarify pool acquisition metrics and gates Describe cancellation in acquisition counters and document the warnings-denied pool configuration checked in CI. --- docs/developers-guide.md | 14 +++++++++----- docs/users-guide.md | 7 ++++--- 2 files changed, 13 insertions(+), 8 deletions(-) diff --git a/docs/developers-guide.md b/docs/developers-guide.md index c18d81926..67449b8d7 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -527,11 +527,12 @@ When all slots are blocked, `FuturesUnordered` races borrowed slot futures; dropping the stream cancels the losers and releases their permits. Each blocked race allocates a task node per slot. `src/client/pool/rotation_tests.rs` pins the ordering, the overflow cases, and the one-increment-per-attempt cursor -contract. Every acquisition attempt also increments -`wireframe_pool_acquisitions_total` with bounded `mode` (`immediate` or -`permit_race`) and `outcome` (`granted`, `exhausted`, `shutdown`, or `error`) -labels, so saturation and shutdown are observable without adding per-slot or -caller identifiers. +contract. Synchronous acquisition attempts and completed asynchronous attempts +increment `wireframe_pool_acquisitions_total` with bounded `mode` (`immediate` +or `permit_race`) and `outcome` (`granted`, `exhausted`, `shutdown`, or +`error`) labels. A blocked asynchronous acquisition cancelled before completion +does not increment this counter. Saturation and shutdown remain observable +without adding per-slot or caller identifiers. `wireframe_pool_permit_races_total` counts blocked race starts and their `granted`, `shutdown`, `error`, or `cancelled` outcomes. @@ -652,6 +653,9 @@ Use the Makefile targets as the contributor entrypoint for routine validation: `whitaker --all -- --all-targets --all-features`. - `make test` runs the main automated test suite with warnings treated as errors. +- `make test-pool-no-metrics` checks the `pool,serializer-bincode` feature set + with default features disabled and warnings denied. The required `build-test` + CI job runs it in the `Pool without metrics compile check` step. Install Whitaker through the standalone installer described in the [Whitaker user's guide](whitaker-users-guide.md) so local linting matches diff --git a/docs/users-guide.md b/docs/users-guide.md index 63b30dccf..5591fe5b1 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -1934,9 +1934,10 @@ acquisition waits for a permit rather than reusing a slot that is already at its #### Client pool metrics -When the `metrics` feature is enabled, every acquisition attempt increments -`wireframe_pool_acquisitions_total`. The metric has two bounded labels, so pool -activity cannot create new time series: +When the `metrics` feature is enabled, synchronous attempts and completed +asynchronous attempts increment `wireframe_pool_acquisitions_total`. An +asynchronous acquisition cancelled while waiting is excluded. The metric has +two bounded labels, so pool activity cannot create new time series: - `mode` records which path made the attempt: `immediate` for the synchronous fast path, or `permit_race` for asynchronous acquisition, including its From ebaa54257adb60b1301a6818bf85d98f5dddb550 Mon Sep 17 00:00:00 2001 From: leynos Date: Wed, 28 Oct 2026 10:44:37 +0100 Subject: [PATCH 16/21] Record toolchain validation boundary Preserve historical benchmark metadata while following current repository toolchain and CI pins for validation. --- docs/adr-013-client-pool-scheduler-and-slot-ownership.md | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md index 9fc850597..db68aae16 100644 --- a/docs/adr-013-client-pool-scheduler-and-slot-ownership.md +++ b/docs/adr-013-client-pool-scheduler-and-slot-ownership.md @@ -570,7 +570,9 @@ both immediate and asynchronous selection. The scheduler and metrics entry points use their respective `mod.rs` files, following the module layout established by #798. The metrics-disabled pool -build uses the existing `tokio-util` `io` feature. +build uses the existing `tokio-util` `io` feature. Validation follows the +repository-pinned Rust toolchain and CI action versions; the historical +baseline captures retain the toolchain metadata from their runs. Blocked races record their starts, terminal outcomes, and durations. A guard records cancellation when the scheduler drops an unfinished acquisition; the From a2300cb2c3f55632e184f42c2cb5118102db4889 Mon Sep 17 00:00:00 2001 From: leynos Date: Wed, 28 Oct 2026 10:56:49 +0100 Subject: [PATCH 17/21] Remove duplicate contended burst test Keep the shared grant test, which already verifies the same wait floor and scheduler progress for each workload. --- .../tests/contended.rs | 40 ------------------- 1 file changed, 40 deletions(-) diff --git a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs index b295733e7..8a84bba4b 100644 --- a/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs +++ b/wireframe_testing/src/runtime_ownership_benchmarks/client_pool_benchmark_support/tests/contended.rs @@ -94,46 +94,6 @@ async fn contended_mode_needs_more_contenders_than_slots() { ); } -/// A contended burst must be *observed* to contend, not assumed to. -/// -/// The burst empties the pool and holds it empty while every contender asks for -/// capacity, so each contender must park: a contended workload with `c` -/// contenders records at least `c` waits before any capacity is returned. That -/// is a floor, not an equality — the contended rounds that follow routinely push -/// it higher — but it is enough to falsify a burst that never actually -/// contended, which is what a start barrier alone fails to rule out. -/// -/// The spawn count is kept beside it as task-churn evidence. It is not evidence -/// of contention: every release notifies the scheduler whether or not anyone is -/// waiting. -#[rstest::rstest] -#[case::one_slot_fifo(1, 16, AcquisitionMode::ContendedFifo)] -#[case::four_slots_fifo(4, 16, AcquisitionMode::ContendedFifo)] -#[case::four_slots_round_robin(4, 16, AcquisitionMode::ContendedRoundRobin)] -#[tokio::test] -async fn contended_acquisition_observes_capacity_waits( - #[case] pool_size: usize, - #[case] contenders: usize, - #[case] mode: AcquisitionMode, -) { - let workload = PoolWorkload { - pool_size, - mode, - acquisitions: 8, - contenders, - }; - let measurement = contended_burst_with_retry(workload) - .await - .expect("a contended burst must finish"); - - assert!( - measurement.capacity_waits >= u64::try_from(contenders).unwrap_or(u64::MAX), - "a burst that empties a {pool_size}-slot pool and holds it empty must park every one of \ - {contenders} contenders, but {} waits were recorded", - measurement.capacity_waits - ); -} - /// A successful contended burst must also deliver leases and exercise the scheduler. #[rstest::rstest] #[case::one_slot_fifo(1, 16, AcquisitionMode::ContendedFifo)] From 5763a422dd3b46a6a1783357df452b986a72d52c Mon Sep 17 00:00:00 2001 From: leynos Date: Wed, 28 Oct 2026 11:14:41 +0100 Subject: [PATCH 18/21] Separate race starts from outcomes Count blocked race starts separately so aggregated terminal outcomes count each completed race once. Clarify the revisions behind allocation observations. --- docs/developers-guide.md | 5 +++-- docs/runtime-ownership-baselines.md | 5 +++++ docs/users-guide.md | 12 ++++++------ src/client/pool/core_tests/metric_tests.rs | 12 ++---------- src/metrics/client_pool.rs | 7 +++++-- src/metrics/mod.rs | 1 + 6 files changed, 22 insertions(+), 20 deletions(-) diff --git a/docs/developers-guide.md b/docs/developers-guide.md index 67449b8d7..29404ae20 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -534,8 +534,9 @@ or `permit_race`) and `outcome` (`granted`, `exhausted`, `shutdown`, or does not increment this counter. Saturation and shutdown remain observable without adding per-slot or caller identifiers. -`wireframe_pool_permit_races_total` counts blocked race starts and their -`granted`, `shutdown`, `error`, or `cancelled` outcomes. +`wireframe_pool_permit_races_started_total` counts blocked race starts. +`wireframe_pool_permit_races_total` counts their `granted`, `shutdown`, +`error`, or `cancelled` terminal outcomes. `wireframe_pool_permit_race_duration_seconds` records their duration with the same terminal labels. Compare starts with terminal outcomes to inspect stuck races, and graph duration and cancellation rates to investigate contention. The diff --git a/docs/runtime-ownership-baselines.md b/docs/runtime-ownership-baselines.md index bb5f32488..99568af34 100644 --- a/docs/runtime-ownership-baselines.md +++ b/docs/runtime-ownership-baselines.md @@ -657,6 +657,11 @@ based on `ab630169`, with the benchmark scenarios already registered; the production race still used boxed `select_all` futures. The after-change run used `d29c51d`. The captures were taken on 2026-10-22. +The earlier #639 allocation row of 2052 for four slots predates the #645 +`PoolCore` work. This comparison uses post-#645 worktrees on both sides, so its +1407 and 1124 counts describe different revisions from that earlier row and +should not be compared with its same-binary jitter range. + The uncontended scenarios use the pool's immediate `try_acquire_permit` path, not the blocked race. Each timing cell gives the Criterion lower, estimate, and upper bounds in microseconds. Those bounds show the observed jitter across the diff --git a/docs/users-guide.md b/docs/users-guide.md index 5591fe5b1..5006bc74f 100644 --- a/docs/users-guide.md +++ b/docs/users-guide.md @@ -1955,12 +1955,12 @@ the configured permits can serve them, and a rising `outcome="error"` rate on the `permit_race` mode means slots are failing instead of granting capacity. When the asynchronous probe finds every slot busy, -`wireframe_pool_permit_races_total` records a blocked race with -`outcome="started"`. It records one terminal outcome: `granted`, `shutdown`, -`error`, or `cancelled` when the acquisition is dropped before completion. -`wireframe_pool_permit_race_duration_seconds` is a histogram of blocked-race -waits in seconds with those terminal outcome labels. Immediate probes that find -capacity do not start a blocked-race observation. +`wireframe_pool_permit_races_started_total` counts the start of a blocked race. +`wireframe_pool_permit_races_total` records one terminal outcome: `granted`, +`shutdown`, `error`, or `cancelled` when the acquisition is dropped before +completion. `wireframe_pool_permit_race_duration_seconds` is a histogram of +blocked-race waits in seconds with those terminal outcome labels. Immediate +probes that find capacity do not start a blocked-race observation. Compare the rates of starts and terminal outcomes to investigate races that remain active, and graph duration and cancellation rates to understand diff --git a/src/client/pool/core_tests/metric_tests.rs b/src/client/pool/core_tests/metric_tests.rs index 6d029281b..2bbdf8fcb 100644 --- a/src/client/pool/core_tests/metric_tests.rs +++ b/src/client/pool/core_tests/metric_tests.rs @@ -190,11 +190,7 @@ fn free_async_probe_skips_blocked_race() { }); observability.snapshot(); observability - .assert_counter( - crate::metrics::POOL_PERMIT_RACES, - [("outcome", "started")], - 0, - ) + .assert_counter(crate::metrics::POOL_PERMIT_RACES_STARTED, [], 0) .expect("free probes must not start a blocked race"); } @@ -241,11 +237,7 @@ fn blocked_race_records_terminal_outcome_and_duration( }); observability.snapshot(); observability - .assert_counter( - crate::metrics::POOL_PERMIT_RACES, - [("outcome", "started")], - 1, - ) + .assert_counter(crate::metrics::POOL_PERMIT_RACES_STARTED, [], 1) .expect("the blocked race should record its start"); observability .assert_counter(crate::metrics::POOL_PERMIT_RACES, [("outcome", outcome)], 1) diff --git a/src/metrics/client_pool.rs b/src/metrics/client_pool.rs index 2001f9112..29670920d 100644 --- a/src/metrics/client_pool.rs +++ b/src/metrics/client_pool.rs @@ -71,7 +71,10 @@ pub(crate) fn inc_pool_pending_response_discard() {} /// ``` pub const POOL_ACQUISITIONS: &str = "wireframe_pool_acquisitions_total"; -/// Counter of blocked permit races, labelled by `started`, `granted`, +/// Counter of blocked permit races that have started. +pub const POOL_PERMIT_RACES_STARTED: &str = "wireframe_pool_permit_races_started_total"; + +/// Counter of completed blocked permit races, labelled by `granted`, /// `shutdown`, `error`, or `cancelled` outcome. pub const POOL_PERMIT_RACES: &str = "wireframe_pool_permit_races_total"; @@ -118,7 +121,7 @@ impl<'a> PoolPermitRaceGuard<'a> { shutdown: &'a AtomicBool, clock: &'a (dyn Fn() -> Instant + Send + Sync), ) -> Self { - counter!(POOL_PERMIT_RACES, "outcome" => "started").increment(1); + counter!(POOL_PERMIT_RACES_STARTED).increment(1); Self { shutdown, started: clock(), diff --git a/src/metrics/mod.rs b/src/metrics/mod.rs index 14796f560..074bf3b5a 100644 --- a/src/metrics/mod.rs +++ b/src/metrics/mod.rs @@ -37,6 +37,7 @@ pub use client_pool::{ POOL_CONNECTION_DISCARDS, POOL_PERMIT_RACE_DURATION, POOL_PERMIT_RACES, + POOL_PERMIT_RACES_STARTED, }; #[cfg(all(feature = "pool", not(loom)))] pub(crate) use client_pool::{ From 7f8fd19cc845dc8319e724f357d1a2791290dfa3 Mon Sep 17 00:00:00 2001 From: leynos Date: Wed, 28 Oct 2026 14:04:53 +0100 Subject: [PATCH 19/21] Clarify benchmark measurement bounds Describe Criterion confidence intervals and scope allocation claims to permit selection, including the later metrics instrumentation. --- docs/runtime-ownership-baselines.md | 24 +++++++++++++----------- 1 file changed, 13 insertions(+), 11 deletions(-) diff --git a/docs/runtime-ownership-baselines.md b/docs/runtime-ownership-baselines.md index 99568af34..834513193 100644 --- a/docs/runtime-ownership-baselines.md +++ b/docs/runtime-ownership-baselines.md @@ -107,8 +107,8 @@ contended burst carries a timeout. Every figure below is a single run on the named machine at the sample budget the binaries pin (20 samples for timings, 10 for allocations). Criterion -reports an interval, not a point; the figure quoted is the middle estimate, -with the interval beside it so the run-to-run spread is visible. +reports a point estimate with confidence-interval bounds; those bounds do not +represent the observed spread between runs. A baseline without its revision is not reproducible, so each capture is named with the commit that produced it. Two captures are recorded here, and a table @@ -650,8 +650,8 @@ at this scope, prefer the spawn count. ## Issue #646 permit-race comparison Both captures use `make bench-runtime-ownership` on the named machine and the -same pool workloads: 256 uncontended acquire/release operations per timing -sample, 20 Criterion samples, 20 contended bursts, and four iterations for +same pool workloads: 256 uncontended acquire/release operations per benchmark +iteration, 20 Criterion samples, 20 contended bursts, and four iterations for allocation observations. The pre-change run used the Phase 1 benchmark worktree based on `ab630169`, with the benchmark scenarios already registered; the production race still used boxed `select_all` futures. The after-change run used @@ -663,9 +663,10 @@ The earlier #639 allocation row of 2052 for four slots predates the #645 should not be compared with its same-binary jitter range. The uncontended scenarios use the pool's immediate `try_acquire_permit` path, -not the blocked race. Each timing cell gives the Criterion lower, estimate, and -upper bounds in microseconds. Those bounds show the observed jitter across the -20 samples; they are not per-function measurements. +not the blocked race. Each timing cell gives Criterion's lower confidence +bound, point estimate, and upper confidence bound for the estimated +per-iteration time in microseconds; these are not per-function measurements or +sample extrema. | Slots | Before latency (lower / estimate / upper, µs) | After latency (lower / estimate / upper, µs) | Before allocations | After allocations | | ----- | --------------------------------------------- | -------------------------------------------- | ------------------ | ----------------- | @@ -675,10 +676,11 @@ upper bounds in microseconds. Those bounds show the observed jitter across the Allocation counts cover four iterations of each burst. They include runtime and benchmark-harness work and do not attribute allocations to -`acquire_slot_permit`. The source-level fast-path property is that -`acquire_slot_permit` allocates nothing when its probe succeeds. When every -slot is blocked, `FuturesUnordered` allocates one task node for each slot -future. +`acquire_slot_permit`. Permit selection allocates nothing when its probe +succeeds; metrics-enabled acquisition recording may allocate a fixed label key. +The captures predate the later acquisition instrumentation, so they do not +measure that cost. When every slot is blocked, `FuturesUnordered` allocates one +task node for each slot future. The contended distribution records successful bursts only; `outcomes` reports successful bursts out of 20. Each distribution cell gives median, p90, and From 4efd353a9d65e3602310ad366d3084a686a118c8 Mon Sep 17 00:00:00 2001 From: leynos Date: Wed, 28 Oct 2026 18:12:21 +0100 Subject: [PATCH 20/21] Preserve Dylint clippy checkout in CI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Select the legacy master branch for Dylint’s rust-clippy clone after the upstream default changed. Scope the wrapper to lint and test both forwarding and a real clone. --- .github/workflows/ci.yml | 8 +- docs/developers-guide.md | 6 ++ scripts/dylint-git/git | 11 +++ tests/workflow_contracts/ci_workflow_test.py | 8 +- tests/workflow_contracts/dylint_git_test.py | 83 ++++++++++++++++++++ 5 files changed, 114 insertions(+), 2 deletions(-) create mode 100755 scripts/dylint-git/git create mode 100644 tests/workflow_contracts/dylint_git_test.py diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index eb31a57f4..199d9d070 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -95,7 +95,13 @@ jobs: - name: Install Whitaker uses: leynos/shared-actions/.github/actions/install-whitaker@8e6a325765ba382691d1b1c65fcccf1fc1d1b00c - name: Lint - run: make lint + # Dylint 6.0.1 expects a local master branch in its rust-clippy clone. + # Limit the compatibility wrapper to this step; all other Git calls + # pass through to the runner's normal Git executable. + run: | + export DYLINT_REAL_GIT="$(command -v git)" + export PATH="$GITHUB_WORKSPACE/scripts/dylint-git:$PATH" + make lint - name: Setup uv uses: astral-sh/setup-uv@18d369b23da25a81e9458561248e1aafff6a70fa with: diff --git a/docs/developers-guide.md b/docs/developers-guide.md index 29404ae20..856db4488 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -662,6 +662,12 @@ Install Whitaker through the standalone installer described in the [Whitaker user's guide](whitaker-users-guide.md) so local linting matches continuous integration (CI). +The CI lint step uses a scoped Git wrapper for Dylint 6.0.1: its driver build +expects a local `master` branch in its `rust-clippy` clone, whose default +branch is now `main`. The wrapper selects `master` for that clone and forwards +other Git commands unchanged. Remove it when the Dylint dependency no longer +assumes the old default branch. + ### CodeScene coverage baseline The `Coverage (main)` workflow in `.github/workflows/coverage-main.yml` runs on diff --git a/scripts/dylint-git/git b/scripts/dylint-git/git new file mode 100755 index 000000000..a2f714389 --- /dev/null +++ b/scripts/dylint-git/git @@ -0,0 +1,11 @@ +#!/usr/bin/env bash +# Dylint 6.0.1 expects a local master branch after cloning rust-clippy. +# Its default branch is now main, so select master for this clone only. +set -eu + +if [[ "${1-}" == clone && "${2-}" == https://github.com/rust-lang/rust-clippy ]]; then + shift 2 + exec "${DYLINT_REAL_GIT:?}" clone --branch master https://github.com/rust-lang/rust-clippy "$@" +fi + +exec "${DYLINT_REAL_GIT:?}" "$@" diff --git a/tests/workflow_contracts/ci_workflow_test.py b/tests/workflow_contracts/ci_workflow_test.py index c1384a46d..75abd68c2 100644 --- a/tests/workflow_contracts/ci_workflow_test.py +++ b/tests/workflow_contracts/ci_workflow_test.py @@ -256,8 +256,14 @@ def _assert_installed_before_lint(install: dict[str, object]) -> None: workflow = yaml.safe_load(WORKFLOW_PATH.read_text(encoding="utf-8")) steps = cast("list[dict[str, object]]", workflow["jobs"]["build-test"]["steps"]) install_index = next((i for i, step in enumerate(steps) if step is install or step == install), None) + lint_command = ( + 'export DYLINT_REAL_GIT="$(command -v git)"\n' + 'export PATH="$GITHUB_WORKSPACE/scripts/dylint-git:$PATH"\n' + "make lint" + ) lint_index = next( - (i for i, step in enumerate(steps) if step.get("run") == "make lint"), None + (i for i, step in enumerate(steps) if str(step.get("run", "")).strip() == lint_command), + None, ) assert install_index is not None, "install-whitaker must be a build-test step" assert lint_index is not None, "build-test must run `make lint`" diff --git a/tests/workflow_contracts/dylint_git_test.py b/tests/workflow_contracts/dylint_git_test.py new file mode 100644 index 000000000..1f3c2b4e3 --- /dev/null +++ b/tests/workflow_contracts/dylint_git_test.py @@ -0,0 +1,83 @@ +"""Verify the CI Git wrapper only changes Dylint's rust-clippy clone.""" + +import json +import os +import subprocess +from pathlib import Path + +import pytest + +WRAPPER = Path(__file__).resolve().parents[2] / "scripts/dylint-git/git" +CLIPPY = "https://github.com/rust-lang/rust-clippy" + + +@pytest.mark.parametrize( + ("arguments", "expected"), + [ + ( + ["clone", CLIPPY, "/tmp/destination", "--quiet"], + ["clone", "--branch", "master", CLIPPY, "/tmp/destination", "--quiet"], + ), + ( + ["clone", "https://github.com/example/other", "/tmp/destination"], + ["clone", "https://github.com/example/other", "/tmp/destination"], + ), + (["--version"], ["--version"]), + ], +) +def test_git_wrapper_forwards_commands( + tmp_path: Path, arguments: list[str], expected: list[str] +) -> None: + """Select the legacy branch only for Dylint's exact clone command.""" + fake_git = tmp_path / "git" + fake_git.write_text( + "#!/usr/bin/env python3\nimport json, sys\nprint(json.dumps(sys.argv[1:]))\n", + encoding="utf-8", + ) + fake_git.chmod(0o755) + env = os.environ | {"DYLINT_REAL_GIT": str(fake_git)} + result = subprocess.run( + [str(WRAPPER), *arguments], env=env, check=True, capture_output=True, text=True + ) + assert json.loads(result.stdout) == expected + + +def test_clippy_clone_checks_out_master_when_default_is_main(tmp_path: Path) -> None: + """A real clone exposes the local branch expected by Dylint.""" + source = tmp_path / "source" + subprocess.run(["git", "init", "-q", "-b", "main", str(source)], check=True) + subprocess.run( + [ + "git", + "-C", + str(source), + "-c", + "user.name=Test", + "-c", + "user.email=test@example.invalid", + "commit", + "--allow-empty", + "-qm", + "initial", + ], + check=True, + ) + subprocess.run(["git", "-C", str(source), "branch", "master"], check=True) + fake_git = tmp_path / "git" + fake_git.write_text( + "#!/usr/bin/env python3\n" + "import os, sys\n" + f"args = [({str(source)!r} if arg == {CLIPPY!r} else arg) for arg in sys.argv[1:]]\n" + "os.execvp('git', ['git', *args])\n", + encoding="utf-8", + ) + fake_git.chmod(0o755) + target = tmp_path / "clone" + env = os.environ | {"DYLINT_REAL_GIT": str(fake_git)} + subprocess.run( + [str(WRAPPER), "clone", CLIPPY, str(target), "--quiet"], env=env, check=True + ) + branch = subprocess.check_output( + ["git", "-C", str(target), "branch", "--show-current"], text=True + ) + assert branch.strip() == "master" From c48718ca47932e2aadf302b166cd19bdb3543c07 Mon Sep 17 00:00:00 2001 From: leynos Date: Wed, 28 Oct 2026 18:21:03 +0100 Subject: [PATCH 21/21] Create Dylint legacy ref after clone Support rust-clippy after its master branch was removed by creating the local reference expected by Dylint at the cloned main revision. --- docs/developers-guide.md | 6 ++--- scripts/dylint-git/git | 7 +++--- tests/workflow_contracts/dylint_git_test.py | 26 ++++++++++++--------- 3 files changed, 22 insertions(+), 17 deletions(-) diff --git a/docs/developers-guide.md b/docs/developers-guide.md index 856db4488..2cfcfa9ec 100644 --- a/docs/developers-guide.md +++ b/docs/developers-guide.md @@ -664,9 +664,9 @@ continuous integration (CI). The CI lint step uses a scoped Git wrapper for Dylint 6.0.1: its driver build expects a local `master` branch in its `rust-clippy` clone, whose default -branch is now `main`. The wrapper selects `master` for that clone and forwards -other Git commands unchanged. Remove it when the Dylint dependency no longer -assumes the old default branch. +branch is now `main`. The wrapper creates the expected local `master` reference +after that clone and forwards other Git commands unchanged. Remove it when the +Dylint dependency no longer assumes the old default branch. ### CodeScene coverage baseline diff --git a/scripts/dylint-git/git b/scripts/dylint-git/git index a2f714389..763610086 100755 --- a/scripts/dylint-git/git +++ b/scripts/dylint-git/git @@ -1,11 +1,12 @@ #!/usr/bin/env bash # Dylint 6.0.1 expects a local master branch after cloning rust-clippy. -# Its default branch is now main, so select master for this clone only. +# Its default branch is now main, so create the expected local ref after clone. set -eu if [[ "${1-}" == clone && "${2-}" == https://github.com/rust-lang/rust-clippy ]]; then - shift 2 - exec "${DYLINT_REAL_GIT:?}" clone --branch master https://github.com/rust-lang/rust-clippy "$@" + destination="${3:?clone destination is required}" + "${DYLINT_REAL_GIT:?}" "$@" + exec "${DYLINT_REAL_GIT:?}" -C "$destination" branch master fi exec "${DYLINT_REAL_GIT:?}" "$@" diff --git a/tests/workflow_contracts/dylint_git_test.py b/tests/workflow_contracts/dylint_git_test.py index 1f3c2b4e3..783c84b12 100644 --- a/tests/workflow_contracts/dylint_git_test.py +++ b/tests/workflow_contracts/dylint_git_test.py @@ -16,19 +16,22 @@ [ ( ["clone", CLIPPY, "/tmp/destination", "--quiet"], - ["clone", "--branch", "master", CLIPPY, "/tmp/destination", "--quiet"], + [ + ["clone", CLIPPY, "/tmp/destination", "--quiet"], + ["-C", "/tmp/destination", "branch", "master"], + ], ), ( ["clone", "https://github.com/example/other", "/tmp/destination"], - ["clone", "https://github.com/example/other", "/tmp/destination"], + [["clone", "https://github.com/example/other", "/tmp/destination"]], ), - (["--version"], ["--version"]), + (["--version"], [["--version"]]), ], ) def test_git_wrapper_forwards_commands( - tmp_path: Path, arguments: list[str], expected: list[str] + tmp_path: Path, arguments: list[str], expected: list[list[str]] ) -> None: - """Select the legacy branch only for Dylint's exact clone command.""" + """Create the legacy ref only for Dylint's exact clone command.""" fake_git = tmp_path / "git" fake_git.write_text( "#!/usr/bin/env python3\nimport json, sys\nprint(json.dumps(sys.argv[1:]))\n", @@ -39,10 +42,10 @@ def test_git_wrapper_forwards_commands( result = subprocess.run( [str(WRAPPER), *arguments], env=env, check=True, capture_output=True, text=True ) - assert json.loads(result.stdout) == expected + assert [json.loads(line) for line in result.stdout.splitlines()] == expected -def test_clippy_clone_checks_out_master_when_default_is_main(tmp_path: Path) -> None: +def test_clippy_clone_creates_master_when_default_is_main(tmp_path: Path) -> None: """A real clone exposes the local branch expected by Dylint.""" source = tmp_path / "source" subprocess.run(["git", "init", "-q", "-b", "main", str(source)], check=True) @@ -62,7 +65,6 @@ def test_clippy_clone_checks_out_master_when_default_is_main(tmp_path: Path) -> ], check=True, ) - subprocess.run(["git", "-C", str(source), "branch", "master"], check=True) fake_git = tmp_path / "git" fake_git.write_text( "#!/usr/bin/env python3\n" @@ -77,7 +79,9 @@ def test_clippy_clone_checks_out_master_when_default_is_main(tmp_path: Path) -> subprocess.run( [str(WRAPPER), "clone", CLIPPY, str(target), "--quiet"], env=env, check=True ) - branch = subprocess.check_output( - ["git", "-C", str(target), "branch", "--show-current"], text=True + refs = subprocess.check_output( + ["git", "-C", str(target), "show-ref", "--heads"], text=True ) - assert branch.strip() == "master" + assert "refs/heads/main" in refs + assert "refs/heads/master" in refs + subprocess.run(["git", "-C", str(target), "checkout", "-q", "master"], check=True)