diff --git a/crates/tracedecay-code-index-retention/src/code_index_generations/generation_transactions.rs b/crates/tracedecay-code-index-retention/src/code_index_generations/generation_transactions.rs index dc00b2bbb6..a87577aeee 100644 --- a/crates/tracedecay-code-index-retention/src/code_index_generations/generation_transactions.rs +++ b/crates/tracedecay-code-index-retention/src/code_index_generations/generation_transactions.rs @@ -2,14 +2,14 @@ //! //! Quarantined generations are journaled, then hard-linked into the replay pool before the receipt is durable. +#[cfg(test)] +use std::cell::Cell; use std::collections::BTreeSet; use std::fs::File; use std::io::Read; #[cfg(unix)] use std::os::unix::fs::MetadataExt; use std::path::{Path, PathBuf}; -#[cfg(test)] -use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::Instant; use sha2::{Digest, Sha256}; @@ -264,23 +264,30 @@ pub(super) fn acquire_graph_replay_pool_lock_checked( GraphReplayPoolLockV1::acquire_exclusive(pool_root, deadline, is_cancelled) } +// Per-thread, not process-wide: an acquire runs on its caller's thread, and +// the test harness runs the other acquire tests in parallel on their own +// threads. Shared statics let any concurrent acquire land between a test's +// reset and its read, which is what turned the exact `(1, 0)` proof into an +// occasional `(5, 3)`. #[cfg(test)] -static GRAPH_REPLAY_POOL_ACQUIRE_TRIES: AtomicUsize = AtomicUsize::new(0); -#[cfg(test)] -static GRAPH_REPLAY_POOL_ACQUIRE_WAITS: AtomicUsize = AtomicUsize::new(0); +thread_local! { + static GRAPH_REPLAY_POOL_ACQUIRE_TRIES: Cell = const { Cell::new(0) }; + static GRAPH_REPLAY_POOL_ACQUIRE_WAITS: Cell = const { Cell::new(0) }; +} #[cfg(test)] pub(super) fn reset_graph_replay_pool_acquire_observation() { - GRAPH_REPLAY_POOL_ACQUIRE_TRIES.store(0, Ordering::SeqCst); - GRAPH_REPLAY_POOL_ACQUIRE_WAITS.store(0, Ordering::SeqCst); + GRAPH_REPLAY_POOL_ACQUIRE_TRIES.with(|tries| tries.set(0)); + GRAPH_REPLAY_POOL_ACQUIRE_WAITS.with(|waits| waits.set(0)); } -/// `(non_blocking_tries, wait_for_exclusive_calls)` since the last reset. +/// `(non_blocking_tries, wait_for_exclusive_calls)` on this thread since the +/// last reset. #[cfg(test)] pub(super) fn graph_replay_pool_acquire_observation() -> (usize, usize) { ( - GRAPH_REPLAY_POOL_ACQUIRE_TRIES.load(Ordering::SeqCst), - GRAPH_REPLAY_POOL_ACQUIRE_WAITS.load(Ordering::SeqCst), + GRAPH_REPLAY_POOL_ACQUIRE_TRIES.with(Cell::get), + GRAPH_REPLAY_POOL_ACQUIRE_WAITS.with(Cell::get), ) } @@ -307,7 +314,7 @@ impl GraphReplayPoolLockV1 { // the budget is gone. Windows lock-conflict is `Ok(None)` via // `is_lock_contended`, not Storage. #[cfg(test)] - GRAPH_REPLAY_POOL_ACQUIRE_TRIES.fetch_add(1, Ordering::SeqCst); + GRAPH_REPLAY_POOL_ACQUIRE_TRIES.with(|tries| tries.set(tries.get() + 1)); match try_acquire_code_generation_store_lock(pool_root)? { Some(guard) => { crate::hotpath_observe::retention_replay_pool_acquired(); @@ -327,7 +334,7 @@ impl GraphReplayPoolLockV1 { fn wait_for_exclusive(deadline: Instant) { #[cfg(test)] - GRAPH_REPLAY_POOL_ACQUIRE_WAITS.fetch_add(1, Ordering::SeqCst); + GRAPH_REPLAY_POOL_ACQUIRE_WAITS.with(|waits| waits.set(waits.get() + 1)); crate::hotpath_observe::retention_replay_pool_acquire_wait(); let remaining = deadline.saturating_duration_since(Instant::now()); if remaining.is_zero() { diff --git a/crates/tracedecay-code-index-retention/src/code_index_generations/locking.rs b/crates/tracedecay-code-index-retention/src/code_index_generations/locking.rs index 1d8d867a75..8b33b98a08 100644 --- a/crates/tracedecay-code-index-retention/src/code_index_generations/locking.rs +++ b/crates/tracedecay-code-index-retention/src/code_index_generations/locking.rs @@ -1,7 +1,11 @@ use std::fs::{File, OpenOptions}; use std::path::{Path, PathBuf}; +use std::time::Instant; -use super::{CodeGenerationRetentionErrorV1, SCOPE_RETENTION_LOCK_FILE, STORE_LOCK_FILE, storage}; +use super::{ + CodeGenerationRetentionErrorV1, GRAPH_REPLAY_POOL_ACQUIRE_BUDGET, + GRAPH_REPLAY_POOL_ACQUIRE_POLL, SCOPE_RETENTION_LOCK_FILE, STORE_LOCK_FILE, storage, +}; pub struct CodeGenerationStoreLockV1 { file: File, @@ -36,7 +40,26 @@ impl Drop for CodeGenerationStoreLockV1 { pub fn acquire_code_generation_store_lock( store_root: &Path, ) -> Result { - lock_file(store_root, STORE_LOCK_FILE, true) + acquire_code_generation_store_lock_checked( + store_root, + Instant::now() + GRAPH_REPLAY_POOL_ACQUIRE_BUDGET, + &|| false, + ) +} + +/// Exclusive generation-store lock that stops at `deadline` or cancellation. +/// +/// A free lock is taken even when the deadline has already elapsed, so a +/// caller that only needs one uncontended critical section is not refused. +/// A held lock returns [`CodeGenerationRetentionErrorV1::GenerationStoreBusy`] +/// or [`CodeGenerationRetentionErrorV1::Cancelled`] instead of blocking in +/// `File::lock`, which cannot observe either signal. +pub(super) fn acquire_code_generation_store_lock_checked( + store_root: &Path, + deadline: Instant, + is_cancelled: &dyn Fn() -> bool, +) -> Result { + lock_file(store_root, STORE_LOCK_FILE, true, deadline, is_cancelled) } /// Try to hold the generation store as a reader for one bounded read of @@ -81,7 +104,13 @@ pub fn try_acquire_code_generation_store_lock( pub(super) fn acquire_scope_retention_lock( store_root: &Path, ) -> Result { - lock_file(store_root, SCOPE_RETENTION_LOCK_FILE, false) + lock_file( + store_root, + SCOPE_RETENTION_LOCK_FILE, + false, + Instant::now() + GRAPH_REPLAY_POOL_ACQUIRE_BUDGET, + &|| false, + ) } #[hotpath::measure(label = "code_index_retention.lock")] @@ -89,16 +118,35 @@ fn lock_file( store_root: &Path, lock_file: &str, generation_store: bool, + deadline: Instant, + is_cancelled: &dyn Fn() -> bool, ) -> Result { let store_root = canonical_store_root(store_root)?; - let lock = open_lock_file(&store_root.join(lock_file))?; - lock.lock().map_err(storage)?; - Ok(CodeGenerationStoreLockV1 { - file: lock, - store_root, - generation_store, - shared: false, - }) + let deadline = deadline.min(Instant::now() + GRAPH_REPLAY_POOL_ACQUIRE_BUDGET); + loop { + if is_cancelled() { + return Err(CodeGenerationRetentionErrorV1::Cancelled); + } + let lock = open_lock_file(&store_root.join(lock_file))?; + match lock.try_lock().map_err(std::io::Error::from) { + Ok(()) => { + return Ok(CodeGenerationStoreLockV1 { + file: lock, + store_root, + generation_store, + shared: false, + }); + } + Err(error) if tracedecay_private_fs::is_lock_contended(&error) => { + if Instant::now() >= deadline { + return Err(CodeGenerationRetentionErrorV1::GenerationStoreBusy); + } + let remaining = deadline.saturating_duration_since(Instant::now()); + std::thread::park_timeout(remaining.min(GRAPH_REPLAY_POOL_ACQUIRE_POLL)); + } + Err(error) => return Err(storage(error)), + } + } } fn canonical_store_root(store_root: &Path) -> Result { diff --git a/crates/tracedecay-code-index-retention/src/code_index_generations/tests/graph_replay_pool_lock_tests.rs b/crates/tracedecay-code-index-retention/src/code_index_generations/tests/graph_replay_pool_lock_tests.rs index 44f0b28e87..9a0e6f9fb9 100644 --- a/crates/tracedecay-code-index-retention/src/code_index_generations/tests/graph_replay_pool_lock_tests.rs +++ b/crates/tracedecay-code-index-retention/src/code_index_generations/tests/graph_replay_pool_lock_tests.rs @@ -1,6 +1,7 @@ use std::sync::atomic::{AtomicUsize, Ordering}; use std::time::{Duration, Instant}; +use super::super::locking::acquire_code_generation_store_lock_checked; use super::*; fn ensure_replay_pool(pool_root: &std::path::Path) { @@ -306,6 +307,52 @@ fn checked_acquire_returns_busy_when_the_carried_deadline_has_elapsed() { drop(publisher); } +#[test] +fn store_lock_returns_cancelled_without_waiting_out_the_budget() { + let (_root, store) = isolated_pool(); + let holder = hold_replay_pool(&store); + let started = Instant::now(); + let error = match acquire_code_generation_store_lock_checked( + &store, + Instant::now() + GRAPH_REPLAY_POOL_ACQUIRE_BUDGET, + &|| true, + ) { + Ok(_) => panic!("cancellation must win a held store lock"), + Err(error) => error, + }; + + assert!(matches!(error, CodeGenerationRetentionErrorV1::Cancelled)); + assert!( + started.elapsed() < Duration::from_millis(20), + "cancelled store lock must not poll the budget, took {:?}", + started.elapsed() + ); + drop(holder); +} + +#[test] +fn store_lock_returns_busy_when_the_carried_deadline_has_elapsed() { + let (_root, store) = isolated_pool(); + let holder = hold_replay_pool(&store); + let started = Instant::now(); + let error = match acquire_code_generation_store_lock_checked(&store, Instant::now(), &|| false) + { + Ok(_) => panic!("an elapsed deadline must defer a held store lock"), + Err(error) => error, + }; + + assert!(matches!( + error, + CodeGenerationRetentionErrorV1::GenerationStoreBusy + )); + assert!( + started.elapsed() < Duration::from_millis(20), + "an expired held store lock must not poll, took {:?}", + started.elapsed() + ); + drop(holder); +} + #[test] fn checked_acquire_takes_a_free_pool_and_releases_without_leak() { let (_root, pool) = isolated_pool(); diff --git a/crates/tracedecay-code-index-retention/src/code_index_generations/text_artifacts.rs b/crates/tracedecay-code-index-retention/src/code_index_generations/text_artifacts.rs index d56bcbb6ac..a6e90c8cd9 100644 --- a/crates/tracedecay-code-index-retention/src/code_index_generations/text_artifacts.rs +++ b/crates/tracedecay-code-index-retention/src/code_index_generations/text_artifacts.rs @@ -442,13 +442,15 @@ pub(super) fn plan_collectable_text_artifacts_cancellable( } else { verification }; - verify_unreferenced_completed_text_artifact( + if !verify_unreferenced_completed_text_artifact( &path, digest, metadata.len(), candidate_verification, is_cancelled, - )?; + )? { + continue; + } Some(CodeTextArtifactRetentionCandidateV1 { artifact_file: file_name, kind: CodeTextArtifactRetentionKindV1::Completed, @@ -554,13 +556,19 @@ pub(super) fn verify_completed_text_artifact( is_cancelled: &dyn Fn() -> bool, ) -> Result<(), CodeGenerationRetentionErrorV1> { let digest = sha256_file_component(&descriptor.artifact_digest, "text artifact")?; - verify_unreferenced_completed_text_artifact( + if !verify_unreferenced_completed_text_artifact( path, digest, descriptor.artifact_size_bytes, verification, is_cancelled, - ) + )? { + return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!( + "code text artifact '{}' disappeared while its identity was being verified", + path.display() + ))); + } + Ok(()) } /// A content-addressed path is trusted only after the open file and its path @@ -573,15 +581,23 @@ pub(super) fn verify_unreferenced_completed_text_artifact( expected_size_bytes: u64, verification: GenerationDigestVerificationV1, is_cancelled: &dyn Fn() -> bool, -) -> Result<(), CodeGenerationRetentionErrorV1> { - let before = std::fs::symlink_metadata(path).map_err(storage)?; +) -> Result { + let before = match std::fs::symlink_metadata(path) { + Ok(metadata) => metadata, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), + Err(error) => return Err(storage(error)), + }; if !before.file_type().is_file() || before.len() != expected_size_bytes { return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!( "code text artifact '{}' has an invalid regular-file identity", path.display() ))); } - let file = File::open(path).map_err(storage)?; + let file = match File::open(path) { + Ok(file) => file, + Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(false), + Err(error) => return Err(storage(error)), + }; if !path_still_names_open_file(path, &file, &before)? { return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!( "code text artifact '{}' changed while its identity was being verified", @@ -602,7 +618,7 @@ pub(super) fn verify_unreferenced_completed_text_artifact( path.display() ))); } - Ok(()) + Ok(true) } /// `active_pointer` is the pointer the store carries *now*, which is not @@ -861,13 +877,18 @@ pub(super) fn stage_collectable_text_artifacts_cancellable( } else { GenerationDigestVerificationV1::Full }; - verify_unreferenced_completed_text_artifact( + if !verify_unreferenced_completed_text_artifact( &source, digest, candidate.size_bytes, candidate_verification, is_cancelled, - )?; + )? { + return Err(CodeGenerationRetentionErrorV1::UnsafeState(format!( + "text-artifact candidate '{}' disappeared before quarantine", + candidate.artifact_file + ))); + } } if observe_cancel(is_cancelled) { return Err(CodeGenerationRetentionErrorV1::Cancelled); diff --git a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry.rs b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry.rs index ab44191731..b548f36d21 100644 --- a/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry.rs +++ b/crates/tracedecay-code-index-runtime/src/code_index_scheduler/registry.rs @@ -1266,11 +1266,17 @@ enum ColdMountAdmissionV1 { /// One exact worktree's pending worker wake. `micros == 0` means no pending /// arrival, and every nonzero arrival is held by one nonzero owner token. +/// +/// `attributable` is false for a worker-owned continuation: the slot stays +/// nonzero so freshness still sees the follow-up, but the instant is not an +/// external wake. Publishing it as [`CodeIndexArrivalV1::Observed`] fabricated +/// an event-to-ready receipt for a pass nobody requested. struct PendingWakeStateV1 { micros: u64, trigger: u64, owner: u64, next_owner: u64, + attributable: bool, } /// The single synchronization authority for one worktree's coalesced wake. @@ -1348,6 +1354,7 @@ impl Default for PendingWakeStateV1 { trigger: 0, owner: 0, next_owner: 1, + attributable: false, } } } @@ -1385,6 +1392,7 @@ impl PendingWakeClaimV1 { let claimed_micros = u64::try_from(now_micros().0).unwrap_or(u64::MAX); let owner = state.next_owner(); state.micros = claimed_micros; + state.attributable = true; state.owner = owner; drop(state); Some(Self { @@ -1426,6 +1434,7 @@ impl Drop for PendingWakeClaimV1 { state.micros = 0; state.trigger = 0; state.owner = 0; + state.attributable = false; } } } @@ -1878,6 +1887,7 @@ impl CodeIndexSchedulerRegistryV1 { pending_wake.micros = 0; pending_wake.owner = 0; pending_wake.trigger = 0; + pending_wake.attributable = false; } } } @@ -2065,9 +2075,13 @@ impl CodeIndexSchedulerRegistryV1 { .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); state.owner = state.next_owner(); - if state.micros == 0 { + // A worker continuation occupies the slot without an external instant. + // This wake is the arrival; keep an already-observed one so a later + // stamp cannot shorten the wait that wake already took. + if state.micros == 0 || !state.attributable { state.micros = wake_micros; } + state.attributable = true; state.trigger = Self::pack_trigger(trigger); drop(state); wake.notify_one(); @@ -2085,28 +2099,46 @@ impl CodeIndexSchedulerRegistryV1 { .state .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); - if state.micros != 0 { + // An unattributable continuation is not an arrival. Upgrade it: the + // caller that just proved work is the event the receipt must name. + if state.micros != 0 && state.attributable { return false; } state.owner = state.next_owner(); state.micros = wake_micros; + state.attributable = true; state.trigger = Self::pack_trigger(trigger); drop(state); wake.notify_one(); true } - /// Queue worker-owned continuation work through the same pending-arrival - /// authority as external wakes. This keeps readiness truthful while the - /// continuation waits for shared admission; a bare `Notify` permit is not - /// observable by freshness readers. + /// Queue worker-owned continuation work so freshness still sees it. + /// + /// A bare `Notify` permit is not observable by freshness readers, so the + /// slot is stamped like an arrival. It is not one. The worker decided to + /// continue work an earlier wake already claimed. `attributable = false` + /// keeps that stamp out of the event-to-ready receipt, which otherwise + /// charged a suppressed freshness probe that raced it. fn note_worker_continuation(pending_wake: &PendingWakeV1, wake: &tokio::sync::Notify) { - if !Self::note_wake_if_idle(pending_wake, wake, CodeIndexCadenceTriggerV1::BusyFollowUp) { - // This pass may have consumed the permit for an arrival it has not - // claimed yet. Keep that observable arrival and replenish its - // coalesced permit so the continuation cannot sleep behind it. + let mut state = pending_wake + .state + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if state.micros != 0 { + // An arrival is already queued, or a continuation already occupies + // the slot. Replenish the coalesced permit so the worker cannot + // sleep behind work it has not claimed. + drop(state); wake.notify_one(); + return; } + state.owner = state.next_owner(); + state.micros = u64::try_from(now_micros().0).unwrap_or(u64::MAX); + state.attributable = false; + state.trigger = Self::pack_trigger(CodeIndexCadenceTriggerV1::BusyFollowUp); + drop(state); + wake.notify_one(); } /// Stamp a continuation while `reconcile_in_progress` still reports this pass. @@ -2135,20 +2167,27 @@ impl CodeIndexSchedulerRegistryV1 { pending_wake: &PendingWakeV1, default_trigger: CodeIndexCadenceTriggerV1, ) -> (CodeIndexArrivalV1, CodeIndexCadenceTriggerV1) { - let (wake_micros, packed_trigger) = { + let (wake_micros, packed_trigger, attributable) = { let mut state = pending_wake .state .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); let wake_micros = state.micros; let packed_trigger = state.trigger; + let attributable = state.attributable; state.micros = 0; state.trigger = 0; state.owner = 0; - (wake_micros, packed_trigger) + state.attributable = false; + (wake_micros, packed_trigger, attributable) }; - if wake_micros == 0 { - return (CodeIndexArrivalV1::Unavailable, default_trigger); + if wake_micros == 0 || !attributable { + let trigger = if wake_micros == 0 { + default_trigger + } else { + Self::unpack_trigger(packed_trigger) + }; + return (CodeIndexArrivalV1::Unavailable, trigger); } let trigger = Self::unpack_trigger(packed_trigger); match i64::try_from(wake_micros) { @@ -2178,12 +2217,14 @@ impl CodeIndexSchedulerRegistryV1 { .lock() .unwrap_or_else(std::sync::PoisonError::into_inner); // A wake that arrived while this pass ran is newer, so the restored - // arrival remains the earliest and stays authoritative. - if state.micros != 0 && state.micros <= wake_micros { + // arrival remains the earliest and stays authoritative. A continuation + // occupying the slot is not an arrival and must not hide this one. + if state.attributable && state.micros != 0 && state.micros <= wake_micros { return; } state.owner = state.next_owner(); state.micros = wake_micros; + state.attributable = true; state.trigger = Self::pack_trigger(trigger); } @@ -3557,7 +3598,9 @@ mod feedback_document_path_tests { #[cfg(test)] mod text_slice_fairness_tests { - use super::{CodeIndexCadenceTriggerV1, CodeIndexSchedulerRegistryV1, PendingWakeV1}; + use super::{ + CodeIndexArrivalV1, CodeIndexCadenceTriggerV1, CodeIndexSchedulerRegistryV1, PendingWakeV1, + }; #[test] fn pending_reconcile_is_serviced_between_bounded_text_slices() { @@ -3587,6 +3630,58 @@ mod text_slice_fairness_tests { "text continuation resumes only after reconcile claims the pending arrival" ); } + + #[test] + fn worker_continuation_stays_pending_without_an_observed_arrival() { + let pending = PendingWakeV1::default(); + let wake = tokio::sync::Notify::new(); + CodeIndexSchedulerRegistryV1::note_worker_continuation(&pending, &wake); + assert!( + pending.has_pending_arrival(), + "freshness must still see the continuation while it waits" + ); + + let (arrival, trigger) = CodeIndexSchedulerRegistryV1::take_pending_arrival( + &pending, + CodeIndexCadenceTriggerV1::Mount, + ); + assert_eq!( + arrival, + CodeIndexArrivalV1::Unavailable, + "a worker continuation is not an external wake and must not publish \ + an event-to-ready sample" + ); + assert_eq!(trigger, CodeIndexCadenceTriggerV1::BusyFollowUp); + assert!( + !pending.has_pending_arrival(), + "claiming the continuation clears the slot" + ); + } + + #[test] + fn an_external_wake_replaces_an_unattributable_continuation() { + let pending = PendingWakeV1::default(); + let wake = tokio::sync::Notify::new(); + CodeIndexSchedulerRegistryV1::note_worker_continuation(&pending, &wake); + assert!( + CodeIndexSchedulerRegistryV1::note_wake_if_idle( + &pending, + &wake, + CodeIndexCadenceTriggerV1::QueryAdmission, + ), + "a real wake must replace the continuation placeholder" + ); + + let (arrival, trigger) = CodeIndexSchedulerRegistryV1::take_pending_arrival( + &pending, + CodeIndexCadenceTriggerV1::Mount, + ); + assert!( + matches!(arrival, CodeIndexArrivalV1::Observed { wake_micros } if wake_micros > 1), + "the receipt names the external wake, not the continuation slot: {arrival:?}" + ); + assert_eq!(trigger, CodeIndexCadenceTriggerV1::QueryAdmission); + } } #[cfg(test)] diff --git a/crates/tracedecay-rusqlite-runtime/src/repository/graph_publication/support.rs b/crates/tracedecay-rusqlite-runtime/src/repository/graph_publication/support.rs index c101e90f35..9b77210313 100644 --- a/crates/tracedecay-rusqlite-runtime/src/repository/graph_publication/support.rs +++ b/crates/tracedecay-rusqlite-runtime/src/repository/graph_publication/support.rs @@ -1,5 +1,5 @@ use std::collections::HashMap; -use std::time::Duration; +use std::time::{Duration, Instant}; use tracedecay_store::{ GraphDependencyGenerationIdentityV1, GraphGenerationIdV1, GraphNamespaceV1, @@ -28,8 +28,6 @@ use super::{ REPLAY_READER_ACQUIRE_SLICE, TOMBSTONE_COLUMNS, }; -const BEGIN_BUSY_ATTEMPT_BUDGET: u32 = 64; - /// Maximum owner sequences bound into one `IN (...)` dependency lookup. /// /// This is **not** `REFERENCED_ANCHOR_BATCH` from @@ -59,37 +57,64 @@ const BEGIN_BUSY_ATTEMPT_BUDGET: u32 = 64; /// exceeded the row cap in one chunk. const GRAPH_REPLAY_DEPENDENCY_BATCH: usize = 38; -#[hotpath::measure(label = "rusqlite.graph_publication.begin")] -pub(super) fn begin( - handle: &ExactSqlHandle, +/// Wall-clock budget for one begin acquisition, retries included. +/// +/// `ExactSqlError::Busy` answers two different questions with one variant: the +/// exact-SQL command queue was full (`map_writer_send_error`), or the writer's +/// own `EXACT_SQL_WRITE_LOCK_ACQUIRE_LIMIT` lock loop was exhausted. Only the +/// first is worth another attempt. A wall-clock budget separates them without a +/// second error variant: an exhausted lock attempt has already spent this whole +/// window inside `begin_immediate`, so it gets exactly one attempt and its +/// 64ms answer is never multiplied into seconds, while a queue refusal returns +/// at once and still has the window to drain in. +const BEGIN_ACQUIRE_BUDGET: Duration = Duration::from_millis(64); + +/// Pause between admission retries, matching the writer's own busy pause. +const BEGIN_BUSY_RETRY_PAUSE: Duration = Duration::from_millis(1); + +/// Runs `attempt` until it answers or the caller is interrupted. +/// `Ok(None)` means no value: the budget expired under `Busy`, or the attempt +/// failed outright. The caller owns what that means for its own operation. +fn acquire_within_begin_budget( context: &GraphPublicationOperationContextV1<'_>, -) -> GraphPublicationStoreResultV1 { - let mut busy_attempts = 0_u32; + mut attempt: impl FnMut() -> Result, +) -> GraphPublicationStoreResultV1> { + let deadline = Instant::now() + BEGIN_ACQUIRE_BUDGET; loop { ensure_not_interrupted(context)?; - match hotpath::measure_block!("rusqlite.graph_publication.begin_immediate", { - handle.begin_immediate() - }) { - Ok(transaction) => { + match attempt() { + Ok(value) => { ensure_not_interrupted(context)?; - return Ok(transaction); + return Ok(Some(value)); } Err(ExactSqlError::Busy) => { - busy_attempts = busy_attempts.saturating_add(1); - if busy_attempts >= BEGIN_BUSY_ATTEMPT_BUDGET { - return Err(GraphPublicationStoreErrorV1::Infrastructure); - } - std::thread::sleep(Duration::from_millis(1)); ensure_not_interrupted(context)?; + if Instant::now() >= deadline { + return Ok(None); + } + std::thread::sleep(BEGIN_BUSY_RETRY_PAUSE); } Err(_) => { ensure_not_interrupted(context)?; - return Err(GraphPublicationStoreErrorV1::Infrastructure); + return Ok(None); } } } } +#[hotpath::measure(label = "rusqlite.graph_publication.begin")] +pub(super) fn begin( + handle: &ExactSqlHandle, + context: &GraphPublicationOperationContextV1<'_>, +) -> GraphPublicationStoreResultV1 { + acquire_within_begin_budget(context, || { + hotpath::measure_block!("rusqlite.graph_publication.begin_immediate", { + handle.begin_immediate() + }) + })? + .ok_or(GraphPublicationStoreErrorV1::Infrastructure) +} + pub(super) fn ensure_owner( handle: &ExactSqlHandle, projection: &GraphProjectionIdentityV1, @@ -128,30 +153,16 @@ pub(super) fn begin_read( handle: &ExactSqlHandle, context: &GraphPublicationOperationContextV1<'_>, ) -> GraphPublicationStoreResultV1 { - let mut busy_attempts = 0_u32; - loop { - ensure_not_interrupted(context)?; - match hotpath::measure_block!("rusqlite.graph_publication.begin_read_snapshot", { + if let Some(snapshot) = acquire_within_begin_budget(context, || { + hotpath::measure_block!("rusqlite.graph_publication.begin_read_snapshot", { handle.begin_read_snapshot(REPLAY_READER_ACQUIRE_SLICE) - }) { - Ok(snapshot) => { - ensure_not_interrupted(context)?; - return Ok(ExactPublicationRead::Snapshot(snapshot)); - } - Err(ExactSqlError::Busy) => { - busy_attempts = busy_attempts.saturating_add(1); - if busy_attempts >= BEGIN_BUSY_ATTEMPT_BUDGET { - break; - } - std::thread::sleep(Duration::from_millis(1)); - ensure_not_interrupted(context)?; - } - Err(_) => { - ensure_not_interrupted(context)?; - break; - } - } + }) + })? { + return Ok(ExactPublicationRead::Snapshot(snapshot)); } + // The deferred fallback waits the writer without consulting `context`, so + // this is the last point that can answer a cancelled or expired caller. + ensure_not_interrupted(context)?; hotpath::measure_block!("rusqlite.graph_publication.begin_deferred", { handle .begin_deferred()