From 9676f5400c08293c91a22a1db756e41c40d95bfe Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 10 Nov 2026 22:54:40 +0100 Subject: [PATCH 1/2] Trace writer queue wait and hold lifetimes per request --- Cargo.lock | 1 + crates/smallclaims/Cargo.toml | 1 + crates/smallclaims/src/profile.rs | 26 +++- crates/smallclaims/src/sqlite.rs | 32 ++++- crates/smallclaims/src/sqlite/telemetry.rs | 139 +++++++++++++++++++++ docs/when-something-is-wrong.md | 19 +++ 6 files changed, 214 insertions(+), 4 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 69958628352..3dc59295ee3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5900,6 +5900,7 @@ dependencies = [ "toml 0.9.12+spec-1.1.0", "tower", "tracing", + "tracing-opentelemetry", "uuid", ] diff --git a/crates/smallclaims/Cargo.toml b/crates/smallclaims/Cargo.toml index 46fed302d03..edd6add8a14 100644 --- a/crates/smallclaims/Cargo.toml +++ b/crates/smallclaims/Cargo.toml @@ -34,6 +34,7 @@ sha2 = "0.10" tokio = { version = "1", features = ["fs", "io-util", "macros", "net", "process", "rt", "sync", "time"] } toml = "0.9" tracing = "0.1" +tracing-opentelemetry = "0.31" tempfile = { version = "3", optional = true } tower = { version = "0.5", features = ["util"] } uuid = { version = "1", features = ["v7"] } diff --git a/crates/smallclaims/src/profile.rs b/crates/smallclaims/src/profile.rs index dbecba7a84a..a19206124bb 100644 --- a/crates/smallclaims/src/profile.rs +++ b/crates/smallclaims/src/profile.rs @@ -942,6 +942,7 @@ pub struct WriterWait { holder: Option>, // Keep queued jobs small beside always-on holders; allocate only for opt-in phase traces. trace: Option>, + writer_operation: Option, } struct WriterTrace { @@ -959,6 +960,7 @@ pub fn writer_waiting() -> Option { .unwrap_or_else(PoisonError::into_inner) .clone(), trace: None, + writer_operation: None, }) } @@ -968,13 +970,35 @@ pub(crate) fn writer_waiting_for( trace_origin: crate::sqlite::telemetry::Origin, ) -> WriterWait { let mut wait = wait.unwrap_or_else(|| WriterWait { - started: Instant::now(), holder: None, trace: None, + started: Instant::now(), holder: None, trace: None, writer_operation: None, }); wait.trace = Some(Box::new(WriterTrace { holder: trace_holder, origin: trace_origin })); wait } impl WriterWait { + pub(crate) fn writer_operation_acquired(&mut self, waited: Duration) { + if let Some(operation) = &mut self.writer_operation { + operation.acquired(waited); + } + } + + #[track_caller] + pub(crate) fn with_writer_operation(wait: Option) -> Option { + if wait.as_ref().is_some_and(|wait| wait.writer_operation.is_some()) { return wait; } + let operation = crate::sqlite::telemetry::WriterOperation::start("foreground", "write.batched"); + if wait.is_none() && operation.is_none() { return None; } + let mut wait = wait.unwrap_or_else(|| Self { + started: Instant::now(), holder: None, trace: None, writer_operation: None, + }); + wait.writer_operation = operation; + Some(wait) + } + + pub(crate) fn take_writer_operation(&mut self) -> Option { + self.writer_operation.take() + } + pub(crate) fn trace_holder(&self) -> Option { self.trace.as_ref().and_then(|trace| trace.holder.clone()) } diff --git a/crates/smallclaims/src/sqlite.rs b/crates/smallclaims/src/sqlite.rs index 9419cc134e7..5bbc338f2b6 100644 --- a/crates/smallclaims/src/sqlite.rs +++ b/crates/smallclaims/src/sqlite.rs @@ -457,6 +457,7 @@ pub struct WriterGuard<'a> { mutation_observer: Option<&'a Arc>, /// When profiling, when this thread took the writer. pub acquired: Option, + writer_operation: Option, /// The connection's changed-row count when it was lent, so the rows this thread changed are /// noted for it when it gives the connection back; see `touched::writes`. pub changes_at_lend: u64, @@ -654,7 +655,11 @@ impl WriterConnection { self.finalizers.install_runtime(callback) } + #[track_caller] pub fn send(&self, mut job: WriterJob) { + if let WriterJob::Batched { wait, .. } = &mut job { + *wait = crate::profile::WriterWait::with_writer_operation(wait.take()); + } if let (Some(writer), WriterJob::Batched { wait, .. }) = (&self.telemetry, &mut job) { *wait = Some(crate::profile::writer_waiting_for(wait.take(), writer.holder(), telemetry::current_origin().unwrap_or_else(telemetry::Origin::other))); @@ -711,6 +716,10 @@ impl WriterConnection { let (give_back, returned) = std::sync::mpsc::sync_channel(1); let enqueued = std::time::Instant::now(); let origin = telemetry::current_origin(); + let mut writer_operation = telemetry::WriterOperation::start( + if matches!(class, writer_queue::LoanClass::Background) { "background" } else { "foreground" }, + "write.lend", + ); match class { writer_queue::LoanClass::Foreground => self.send(WriterJob::Lend { lent, returned, holder, enqueued, origin }), writer_queue::LoanClass::Background => { @@ -748,6 +757,9 @@ impl WriterConnection { .expect("the writer thread lends its connection"); drop(waiting); telemetry::acknowledged(enqueued.elapsed(), false); + if let Some(operation) = &mut writer_operation { + operation.acquired(enqueued.elapsed()); + } WriterGuard { changes_at_lend: connection.total_changes(), connection: Some(connection), @@ -758,6 +770,7 @@ impl WriterConnection { handler_policy: &self.handler_policy, mutation_observer: self.mutation_observer.as_ref(), acquired: crate::profile::writer_acquired(wait), + writer_operation, } } @@ -944,7 +957,7 @@ fn write_queue( /// writer), fails, or meets a lending write, which it returns to run next. fn run_write_batch( connection: &mut Connection, - first: WriterJob, + mut first: WriterJob, // Where the batch takes more writes from, and for how long. intake: (&std::sync::mpsc::Receiver, std::time::Duration), committed_index: &AtomicU64, @@ -959,6 +972,10 @@ fn run_write_batch( ) -> Option { let (queue, window) = intake; let (observers, finalizers, mutation_observer, writer_telemetry) = callbacks; + // The first caller owns transaction setup too; later jobs acquire at their savepoint. + if let WriterJob::Batched { wait: Some(wait), enqueued, .. } = &mut first { + wait.writer_operation_acquired(enqueued.elapsed()); + } let first_origin = match &first { WriterJob::Batched { wait, .. } => wait.as_ref().and_then(crate::profile::WriterWait::trace_origin), _ => None, @@ -1022,6 +1039,10 @@ fn run_write_batch( } } let origin = wait.as_mut().map(crate::profile::WriterWait::take_trace_origin); + let mut writer_operation = wait.as_mut().and_then(crate::profile::WriterWait::take_writer_operation); + if let Some(operation) = &mut writer_operation { + operation.acquired(enqueued.elapsed()); + } if let Some(writer) = writer_telemetry { writer.set_holder(origin.clone()); } @@ -1031,6 +1052,7 @@ fn run_write_batch( } match (&transaction, &failure) { (Some(transaction), None) => { + let _request = writer_operation.as_ref().map(telemetry::WriterOperation::enter); let operation_started = attribution.as_ref().map(|_| std::time::Instant::now()); // The caller's operation waited until now, holds the writer while its write // runs, and shares the commit below with the rest of the batch. @@ -1066,7 +1088,7 @@ fn run_write_batch( // A batch that failed to begin answers its first write with the error. _ => drop(run), } - answers.push(done); + answers.push((done, writer_operation)); if failure.is_none() && answers.len() < WRITE_BATCH_LIMIT && started.elapsed() < window @@ -1135,7 +1157,9 @@ fn run_write_batch( // delivery too: channel backpressure is no longer counted as writer-held time. drop(hold); drop(_holder); - for done in answers { + for (done, writer_operation) in answers { + // Include shared finalizers, COMMIT and observers, but not ACK backpressure. + drop(writer_operation); let _ = done.send(committed.clone()); } lend @@ -1208,6 +1232,8 @@ impl Drop for WriterGuard<'_> { ); crate::profile::writer_released(self.acquired.take()); let _ = self.give_back.send(connection); + // The loan remains held through finalization, observer notification and return. + drop(self.writer_operation.take()); } } diff --git a/crates/smallclaims/src/sqlite/telemetry.rs b/crates/smallclaims/src/sqlite/telemetry.rs index bc6f25b5de4..d80cf3c9df0 100644 --- a/crates/smallclaims/src/sqlite/telemetry.rs +++ b/crates/smallclaims/src/sqlite/telemetry.rs @@ -23,6 +23,69 @@ pub fn phase_spans_enabled() -> bool { EXPORT_ENABLED.load(Ordering::Relaxed) } +/// One caller's writer lifetime, independent of opt-in slow-batch root spans. +/// The disabled path keeps no span, request handle, or caller-site allocation. +pub(crate) struct WriterOperation { + context: opentelemetry::Context, + request: Option>, + acquired: Option, +} + +impl WriterOperation { + #[track_caller] + pub(crate) fn start(queue: &'static str, operation: &'static str) -> Option { + use opentelemetry::trace::{TraceContextExt as _, Tracer as _}; + use tracing_opentelemetry::OpenTelemetrySpanExt as _; + let mut parent = opentelemetry::Context::current(); + if !parent.span().span_context().is_valid() { + let current = tracing::Span::current(); + if current.is_disabled() { return None; } + parent = current.context(); + } + let span_context = parent.span().span_context().clone(); + if !span_context.is_valid() || !span_context.is_sampled() { return None; } + let caller = std::panic::Location::caller(); + let tracer = opentelemetry::global::tracer("smallclaims.sqlite"); + let span = tracer.span_builder("st.db.writer") + .with_attributes([ + KeyValue::new("span.label", operation), + KeyValue::new("db.system.name", "sqlite"), + KeyValue::new("db.operation.name", operation), + KeyValue::new("queue", queue), + KeyValue::new("code.file.path", caller.file()), + KeyValue::new("code.line.number", i64::from(caller.line())), + KeyValue::new("code.column.number", i64::from(caller.column())), + ]) + .start_with_context(&tracer, &parent); + Some(Self { + context: parent.with_span(span), + request: current_request(), + acquired: None, + }) + } + + pub(crate) fn acquired(&mut self, waited: Duration) { + use opentelemetry::trace::TraceContextExt as _; + if self.acquired.is_some() { return; } + self.acquired = Some(std::time::Instant::now()); + self.context.span().set_attribute(KeyValue::new("writer_wait_ms", waited.as_secs_f64() * 1000.)); + } + + pub(crate) fn enter(&self) -> (opentelemetry::ContextGuard, RequestScope) { + (self.context.clone().attach(), enter_request(self.request.clone())) + } +} + +impl Drop for WriterOperation { + fn drop(&mut self) { + use opentelemetry::trace::TraceContextExt as _; + if let Some(acquired) = self.acquired { + self.context.span().set_attribute(KeyValue::new("writer_hold_ms", acquired.elapsed().as_secs_f64() * 1000.)); + } + self.context.span().end(); + } +} + #[derive(Clone, PartialEq, Eq, Hash)] pub struct Origin(OriginLabel); @@ -688,6 +751,82 @@ mod tests { use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData}; use opentelemetry_sdk::metrics::{InMemoryMetricExporter, PeriodicReader, SdkMeterProvider}; + #[test] + fn writer_children_keep_request_context_and_complete_after_finalization() { + // Isolate the global tracer provider and opt-in phase flag from sibling tests. + const CHILD: &str = "ST3_WRITER_CHILD_SPANS_TEST_CHILD"; + if std::env::var_os(CHILD).is_none() { + let status = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", "sqlite::telemetry::tests::writer_children_keep_request_context_and_complete_after_finalization", "--nocapture"]) + .env(CHILD, "1") + .status().unwrap(); + assert!(status.success()); + return; + } + use opentelemetry::trace::{Span as _, TraceContextExt as _, Tracer as _, TracerProvider as _}; + let exporter = opentelemetry_sdk::trace::InMemorySpanExporter::default(); + let provider = opentelemetry_sdk::trace::SdkTracerProvider::builder() + .with_simple_exporter(exporter.clone()).build(); + opentelemetry::global::set_tracer_provider(provider.clone()); + assert!(!phase_spans_enabled()); + assert!(WriterOperation::start("foreground", "write.lend").is_none()); + let request_span = provider.tracer("writer.test").start("request"); + let parent_id = request_span.span_context().span_id(); + let context = opentelemetry::Context::current_with_span(request_span); + let _attached = context.attach(); + let request = Arc::new(WriterWait::default()); + let _request = enter_request(Some(request.clone())); + let writer = super::super::WriterConnection::new( + rusqlite::Connection::open_in_memory().unwrap(), + Arc::new(AtomicU64::new(0)), + ); + let finalized = Arc::new(std::sync::atomic::AtomicBool::new(false)); + let finalized_in_hook = finalized.clone(); + let finalizer_exporter = exporter.clone(); + writer.install_transaction_maintenance(move |_| { + assert!(!finalizer_exporter.get_finished_spans().unwrap().iter() + .any(|span| span.name == "st.db.writer" + && span.attributes.contains(&KeyValue::new("db.operation.name", "write.batched")))); + finalized_in_hook.store(true, Ordering::Release); + Ok(()) + }).unwrap(); + // Installing hooks takes a foreground fence loan; inspect only the calls below. + exporter.reset(); + let expected_request = request.clone(); + let batch_line = line!() + 1; + writer.batched(move |_| -> Result<(), ()> { + assert!(Arc::ptr_eq(¤t_request().unwrap(), &expected_request)); + let current = opentelemetry::Context::current(); + assert_ne!(current.span().span_context().span_id(), parent_id); + assert!(current.span().span_context().is_valid()); + Ok(()) + }).unwrap().unwrap(); + assert!(finalized.load(Ordering::Acquire)); + assert_eq!(exporter.get_finished_spans().unwrap().len(), 1); + let foreground_line = line!() + 1; + drop(writer.write()); + let background_line = line!() + 1; + drop(writer.write_background()); + let spans = exporter.get_finished_spans().unwrap(); + assert_eq!(spans.len(), 3); + for (span, (queue, line)) in spans.iter().zip([ + ("foreground", batch_line), ("foreground", foreground_line), ("background", background_line), + ]) { + assert_eq!(span.name, "st.db.writer"); + assert_eq!(span.parent_span_id, parent_id); + assert!(span.attributes.contains(&KeyValue::new("queue", queue))); + assert!(span.attributes.contains(&KeyValue::new("code.file.path", file!()))); + assert!(span.attributes.contains(&KeyValue::new("code.line.number", i64::from(line)))); + for key in ["writer_wait_ms", "writer_hold_ms"] { + assert!(span.attributes.iter().any(|attribute| { + attribute.key.as_str() == key + && matches!(attribute.value, opentelemetry::Value::F64(value) if value >= 0.) + })); + } + } + assert!(Arc::ptr_eq(¤t_request().unwrap(), &request)); + } + #[test] fn checkpoint_histogram_exports_operation_and_report_outcome() { use opentelemetry::metrics::MeterProvider as _; diff --git a/docs/when-something-is-wrong.md b/docs/when-something-is-wrong.md index 678ae806431..3c5a44d1c33 100644 --- a/docs/when-something-is-wrong.md +++ b/docs/when-something-is-wrong.md @@ -39,6 +39,25 @@ fractal terminals peek agent/garden/worker The last terminal screen can explain a login, crash, or provider limit; the graph says whether the stop was requested, the owner ended, a suspension is active, or the runtime failed. A stopped agent can retain a historical terminal, and an agent with no recorded terminal may have nothing to peek. +## Writer queue waits + +When an interactive request is slow during replication, distinguish time queued +for the store writer from time holding it. Request-parented `st.db.writer` spans +report `writer_wait_ms`, `writer_hold_ms`, the `queue` class (`foreground` or +`background`), and the static `code.file.path`/`code.line.number` caller site. +These children require a sampled trace context, not `ST3_STORAGE_PHASE_SPANS`. +Caller sites are trace attributes, not metric dimensions. + +Foreground batched writes and foreground loans retain arrival order. Background +loans yield to pending foreground work, with one background turn after at most +eight foreground turns to prevent starvation. This bounds dispatch turns, not +wall time: an active envelope, transaction commit, and its finalization are not +preemptible. A long background hold can therefore still delay an interactive +request even when scheduling priority is correct. + +Compare the caller site and hold stages before attributing a slow request to +queue priority or changing admission chunk sizes. + ## Linux PID namespaces and procfs The runtime requires `/proc` to describe its own PID namespace. Entering a child From b179a140a8e5e352512657fca3a8568c6d3d129f Mon Sep 17 00:00:00 2001 From: schickling-assistant <261620128+schickling-assistant@users.noreply.github.com> Date: Tue, 10 Nov 2026 22:54:57 +0100 Subject: [PATCH 2/2] Schedule replication verdict pages as background writer loans --- .../src/sqlite/writer_queue/tests.rs | 74 +++++++++++++++++++ crates/smallclaims/src/store/principals.rs | 43 ++++++++++- docs/when-something-is-wrong.md | 3 + 3 files changed, 119 insertions(+), 1 deletion(-) diff --git a/crates/smallclaims/src/sqlite/writer_queue/tests.rs b/crates/smallclaims/src/sqlite/writer_queue/tests.rs index f752c258343..15104171a6f 100644 --- a/crates/smallclaims/src/sqlite/writer_queue/tests.rs +++ b/crates/smallclaims/src/sqlite/writer_queue/tests.rs @@ -313,6 +313,80 @@ fn foreground_only_fifo_and_batched_durable_answers_are_unchanged() { last.release(connection); } +#[test] +fn foreground_batched_write_gets_a_bounded_turn_during_a_long_verdict_loop() { + let store = Arc::new(crate::store::Store::open_memory( + "studio", + Arc::new(crate::store::runtime::Plain), + ).unwrap()); + let total = 1_024_u64; + { + let mut connection = store.connection.write(); + let tx = connection.transaction().unwrap(); + for i in 0..total { + tx.execute( + "INSERT INTO claim_verdict_queue(claim_id) VALUES (?1)", + [format!("removed-claim/{i:05}")], + ).unwrap(); + } + tx.commit().unwrap(); + } + let writer = &store.connection; + let held = writer.write(); + // Intercept admission, not execution timing: the real verdict loop must opt into + // background scheduling before any contender can run. + let (proxy, incoming) = mpsc::channel(); + let original = writer.jobs.lock().unwrap().replace(proxy).unwrap(); + let judging = store.clone(); + let thread = std::thread::spawn(move || judging.judge_claims(false).unwrap()); + let pending = incoming.recv_timeout(Duration::from_secs(10)).unwrap(); + *writer.jobs.lock().unwrap() = Some(original.clone()); + assert!( + matches!(pending, WriterJob::BackgroundReady(_)), + "replication verdict chunks must be classified as background loans", + ); + original.send(pending).unwrap(); + let front = selected_loans(writer, FOREGROUND_TURNS); + let (recorded, remaining) = mpsc::sync_channel(1); + let (done, answer) = mpsc::sync_channel(1); + writer.send(WriterJob::Batched { + run: Box::new(move |tx| { + let queued: u64 = tx.query_row( + "SELECT COUNT(*) FROM claim_verdict_queue", [], |row| row.get(0), + ).unwrap(); + recorded.send(queued).unwrap(); + tx.execute( + "INSERT INTO meta(key,value) VALUES ('verdict-fairness-foreground','committed')", + [], + ).unwrap(); + true + }), + profile: None, + wait: None, + holder: crate::windows::holds::Holder::current(), + done, + enqueued: Instant::now(), + }); + drop(held); + for next in &front { + next.release(next.acquire()); + } + answer.recv_timeout(Duration::from_secs(10)).unwrap().unwrap(); + // Eight queued foreground turns permit one background chunk (32 claims), not a + // complete backlog drain, before the batched foreground write receives its turn. + assert_eq!(remaining.recv_timeout(Duration::from_secs(10)).unwrap(), total - 32); + thread.join().unwrap(); + assert!(!store.verdicts_pending().unwrap()); + assert_eq!( + writer.write().query_row( + "SELECT value FROM meta WHERE key='verdict-fairness-foreground'", + [], + |row| row.get::<_, String>(0), + ).unwrap(), + "committed", + ); +} + #[test] fn actual_worker_failure_disconnects_queued_background_and_foreground_borrowers() { let writer = writer(); diff --git a/crates/smallclaims/src/store/principals.rs b/crates/smallclaims/src/store/principals.rs index 7a32fb44b8e..378d46a4d76 100644 --- a/crates/smallclaims/src/store/principals.rs +++ b/crates/smallclaims/src/store/principals.rs @@ -1741,7 +1741,7 @@ impl Store { let mut ordinary_claims = BTreeSet::new(); loop { loans += 1; - let mut connection = self.connection.write(); + let mut connection = self.connection.write_background(); let transaction = connection.transaction_with_behavior(rusqlite::TransactionBehavior::Immediate)?; let roots = self.trust_roots(&transaction, &self.origin)?; @@ -2398,6 +2398,47 @@ mod paced_verdict_tests { assert_eq!(count(&store, "SELECT COUNT(*) FROM claim_verdict_queue"), 0); } + #[test] + fn bootstrap_verdict_loans_revalidate_changed_roots_before_activation() { + let store = root_linked(REQUEUE_PAGE + 1); + store.connection.write().execute( + "UPDATE claim_signature_migration SET phase='judging' WHERE id=1", + [], + ).unwrap(); + // Skip the optional root-check preflight: bootstrap must revalidate inside each + // bounded loan, including the transaction that eventually activates the fold. + store.judge_claims_inner(false, Some(1), false).unwrap(); + let before = fleet_meta(&store.readers.get(), VERDICT_ROOTS).unwrap().unwrap(); + assert_eq!(cursor(&store, "root").as_deref(), Some("claim/01023")); + assert_eq!(store.signature_migration().unwrap().phase, "judging"); + { + let connection = store.connection.write(); + set_meta_tx(&connection, OWN_NODE_KEY, "changed-between-verdict-loans").unwrap(); + } + let expected = { + let connection = store.readers.get(); + crate::hash::canonical_hash(&store.trust_roots(&connection, &store.origin).unwrap()) + .unwrap() + }; + assert_ne!(before, expected); + store.judge_claims_inner(false, Some(1), false).unwrap(); + assert_eq!( + fleet_meta(&store.readers.get(), VERDICT_ROOTS).unwrap(), + Some(expected.clone()), + ); + // The changed roots restart dependency paging rather than continue past the first + // page under a digest that no longer describes the current authority. + assert_eq!(cursor(&store, "root").as_deref(), Some("claim/01023")); + assert_eq!(store.signature_migration().unwrap().phase, "judging"); + store.judge_claims_inner(false, None, false).unwrap(); + assert_eq!(store.signature_migration().unwrap().phase, "done"); + assert_eq!(cursor(&store, "root"), None); + assert_eq!( + fleet_meta(&store.readers.get(), VERDICT_ROOTS).unwrap(), + Some(expected), + ); + } + #[test] fn a_claim_sealed_after_the_startup_pass_is_handed_to_the_paced_job_and_judged() { // A single node with no fleet: nothing exchanges, so nothing else judges. Its node key diff --git a/docs/when-something-is-wrong.md b/docs/when-something-is-wrong.md index 3c5a44d1c33..ae7c1d4c91a 100644 --- a/docs/when-something-is-wrong.md +++ b/docs/when-something-is-wrong.md @@ -55,6 +55,9 @@ wall time: an active envelope, transaction commit, and its finalization are not preemptible. A long background hold can therefore still delay an interactive request even when scheduling priority is correct. +Replication verdict pages use background loans and revalidate trust roots after +each acquisition, including the transaction that activates the authority fold. + Compare the caller site and hold stages before attributing a slow request to queue priority or changing admission chunk sizes.