Trace writer lifetimes and yield replication verdict pages

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"] }

crates/smallclaims/src/profile.rs

@@ -942,6 +942,7 @@ pub struct WriterWait {
     holder: Option<Arc<str>>,
     // Keep queued jobs small beside always-on holders; allocate only for opt-in phase traces.
     trace: Option<Box<WriterTrace>>,
+    writer_operation: Option<crate::sqlite::telemetry::WriterOperation>,
 }
 
 struct WriterTrace {
@@ -959,6 +960,7 @@ pub fn writer_waiting() -> Option<WriterWait> {
             .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<Self>) -> Option<Self> {
+        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<crate::sqlite::telemetry::WriterOperation> {
+        self.writer_operation.take()
+    }
+
     pub(crate) fn trace_holder(&self) -> Option<crate::sqlite::telemetry::Origin> {
         self.trace.as_ref().and_then(|trace| trace.holder.clone())
     }

crates/smallclaims/src/sqlite.rs

@@ -457,6 +457,7 @@ pub struct WriterGuard<'a> {
     mutation_observer: Option<&'a Arc<writer_observer::MutationState>>,
     /// When profiling, when this thread took the writer.
     pub acquired: Option<std::time::Instant>,
+    writer_operation: Option<telemetry::WriterOperation>,
     /// 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<WriterJob>, std::time::Duration),
     committed_index: &AtomicU64,
@@ -959,6 +972,10 @@ fn run_write_batch(
 ) -> Option<WriterJob> {
     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());
     }
 }
 

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<Arc<WriterWait>>,
+    acquired: Option<std::time::Instant>,
+}
+
+impl WriterOperation {
+    #[track_caller]
+    pub(crate) fn start(queue: &'static str, operation: &'static str) -> Option<Self> {
+        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(&current_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(&current_request().unwrap(), &request));
+    }
+
     #[test]
     fn checkpoint_histogram_exports_operation_and_report_outcome() {
         use opentelemetry::metrics::MeterProvider as _;