diff --git a/crates/tracedecay-daemon-service/src/invocation/lsp.rs b/crates/tracedecay-daemon-service/src/invocation/lsp.rs index 7c19fb938f7..b42c42072de 100644 --- a/crates/tracedecay-daemon-service/src/invocation/lsp.rs +++ b/crates/tracedecay-daemon-service/src/invocation/lsp.rs @@ -773,19 +773,31 @@ impl DaemonInvocationService { Ok(access) => access, Err(problem) => return DaemonInvocationResponse::problem(request_id, problem), }; - let mut sessions = self.lsp_sessions.lock().await; - let Some(session) = sessions.get_mut(access.session_id()) else { - return DaemonInvocationResponse::problem( - request_id, - DaemonInvocationProblem::NotFoundOrNotAuthorized, - ); + // The session map lock covers only the state transition. The receipt's + // durable admission is disk-latency work and runs after it releases. + let (acknowledged, delivery) = { + let mut sessions = self.lsp_sessions.lock().await; + let Some(session) = sessions.get_mut(access.session_id()) else { + return DaemonInvocationResponse::problem( + request_id, + DaemonInvocationProblem::NotFoundOrNotAuthorized, + ); + }; + let acknowledged = session.actor.acknowledge_outbound(); + let delivery = if acknowledged { + session + .claim_in_flight_delivery( + tracedecay_domain::DeliverySettlementOutcomeV1::Delivered, + None, + ) + .ok() + } else { + None + }; + (acknowledged, delivery) }; - let acknowledged = session.actor.acknowledge_outbound(); - if acknowledged { - let _ = session.settle_in_flight_delivery( - tracedecay_domain::DeliverySettlementOutcomeV1::Delivered, - None, - ); + if let Some(delivery) = delivery { + let _ = delivery.admit().await; } DaemonInvocationResponse::with_outcome( request_id, @@ -824,10 +836,12 @@ impl DaemonInvocationService { // The stdio bridge reaches this path after a write or flush failure. // Removing the actor discards its in-flight frame, so terminalize that // exact captured attempt before any later detach bookkeeping can fail. - let _ = session.settle_in_flight_delivery( + if let Ok(delivery) = session.claim_in_flight_delivery( tracedecay_domain::DeliverySettlementOutcomeV1::Dropped, Some(tracedecay_domain::DeliveryDropReasonV1::Disconnected), - ); + ) { + let _ = delivery.admit().await; + } let lease_cancelled = self .lsp_lease_tasks .cancel(access.session_id()) diff --git a/crates/tracedecay-daemon-service/src/invocation/lsp_delivery.rs b/crates/tracedecay-daemon-service/src/invocation/lsp_delivery.rs index 0d448043ee7..dd9be0797e0 100644 --- a/crates/tracedecay-daemon-service/src/invocation/lsp_delivery.rs +++ b/crates/tracedecay-daemon-service/src/invocation/lsp_delivery.rs @@ -89,25 +89,41 @@ pub enum LspDeliverySettlementAdmissionV1 { Refused(&'static str), } -impl RuntimeLspSession { - pub(super) fn settle_in_flight_delivery( - &mut self, - outcome: tracedecay_domain::DeliverySettlementOutcomeV1, - drop_reason: Option, - ) -> LspDeliverySettlementAdmissionV1 { - let Some(attempt) = self.in_flight_delivery_attempt.take() else { - return LspDeliverySettlementAdmissionV1::NoAttemptInFlight; - }; - let Some(recorder) = self.delivery_settlements.as_ref() else { - return LspDeliverySettlementAdmissionV1::RecorderUnavailable; - }; - let settlement = tracedecay_domain::DeliverySettlementV1 { - settled_at: now_micros().max(attempt.attempted_at), - attempt, - outcome, - drop_reason, - }; - match recorder.try_record(settlement) { +/// The terminal receipt for one session's in-flight outbound frame, claimed +/// out of the session so its durable admission can run without the session. +/// +/// Admission publishes the receipt to the recorder's spool with file and +/// directory fsyncs. That is disk-latency work: run under the daemon-wide +/// session map lock it stalls every LSP session and every invocation (each one +/// takes that lock to expire sessions), and run on an async worker it also +/// takes the worker. Claiming under the lock and admitting after it releases +/// both. +pub(super) struct LspDeliverySettlementClaim { + recorder: Arc, + settlement: tracedecay_domain::DeliverySettlementV1, +} + +impl LspDeliverySettlementClaim { + /// Admits the receipt on the blocking pool. The work is queued at this + /// call, so abandoning the returned future abandons only the wait: a + /// claimed receipt is never lost to a dropped connection. + pub(super) async fn admit(self) -> LspDeliverySettlementAdmissionV1 { + match tracedecay_runtime_core::native_runtime::spawn_blocking(move || self.admit_inline()) + .await + { + Ok(admission) => admission, + Err(error) => { + tracing::warn!(%error, "LSP delivery receipt admission could not run"); + LspDeliverySettlementAdmissionV1::Refused( + "lsp_delivery_settlement_blocking_pool_unavailable", + ) + } + } + } + + /// Admission on the calling thread, for `Drop`, which cannot await. + fn admit_inline(self) -> LspDeliverySettlementAdmissionV1 { + match self.recorder.try_record(self.settlement) { Ok(tracedecay_application::observability::DeliverySettlementRecordOutcomeV1::Enqueued) => { LspDeliverySettlementAdmissionV1::Enqueued } @@ -125,15 +141,44 @@ impl RuntimeLspSession { } } +impl RuntimeLspSession { + /// Takes the exact in-flight attempt for settlement. The attempt leaves the + /// session here, so a concurrent terminalization can never settle it twice. + pub(super) fn claim_in_flight_delivery( + &mut self, + outcome: tracedecay_domain::DeliverySettlementOutcomeV1, + drop_reason: Option, + ) -> Result { + let Some(attempt) = self.in_flight_delivery_attempt.take() else { + return Err(LspDeliverySettlementAdmissionV1::NoAttemptInFlight); + }; + let Some(recorder) = self.delivery_settlements.as_ref() else { + return Err(LspDeliverySettlementAdmissionV1::RecorderUnavailable); + }; + Ok(LspDeliverySettlementClaim { + recorder: Arc::clone(recorder), + settlement: tracedecay_domain::DeliverySettlementV1 { + settled_at: now_micros().max(attempt.attempted_at), + attempt, + outcome, + drop_reason, + }, + }) + } +} + impl Drop for RuntimeLspSession { fn drop(&mut self) { // Any removal without a protocol ACK discards the frame. This fallback // covers transport loss, TTL expiry, owner retirement, and shutdown; - // explicit paths may settle first with a more specific reason. - let _ = self.settle_in_flight_delivery( + // explicit paths may settle first with a more specific reason. Drop + // cannot await, so admission is inline here. + if let Ok(claim) = self.claim_in_flight_delivery( tracedecay_domain::DeliverySettlementOutcomeV1::Dropped, Some(tracedecay_domain::DeliveryDropReasonV1::Disconnected), - ); + ) { + let _ = claim.admit_inline(); + } self.actor.expire(); } } diff --git a/crates/tracedecay-daemon-service/src/invocation/tests/lsp_tests.rs b/crates/tracedecay-daemon-service/src/invocation/tests/lsp_tests.rs index 945361c7db0..f6ae482f046 100644 --- a/crates/tracedecay-daemon-service/src/invocation/tests/lsp_tests.rs +++ b/crates/tracedecay-daemon-service/src/invocation/tests/lsp_tests.rs @@ -575,6 +575,14 @@ async fn open_polled_lsp_delivery( } async fn assert_one_lsp_delivery_drop(fixture: LspDeliveryFixture) { + assert_one_lsp_delivery_settlement(fixture, 0, 1).await; +} + +async fn assert_one_lsp_delivery_settlement( + fixture: LspDeliveryFixture, + delivered: u16, + dropped: u16, +) { let summary = fixture .recorder .shutdown() @@ -613,11 +621,90 @@ async fn assert_one_lsp_delivery_drop(fixture: LspDeliveryFixture) { else { panic!("expected LSP delivery fanout observation"); }; - assert_eq!(fanout.delivered, 0); - assert_eq!(fanout.dropped, 1); + assert_eq!(fanout.delivered, delivered); + assert_eq!(fanout.dropped, dropped); assert_eq!(fanout.unknown, 0); } +/// Records the thread that opens each span of one name. +struct SpanThreads { + span_name: &'static str, + threads: Arc>>, +} + +impl tracing_subscriber::Layer for SpanThreads { + fn on_new_span( + &self, + attributes: &tracing::span::Attributes<'_>, + _id: &tracing::span::Id, + _context: tracing_subscriber::layer::Context<'_, S>, + ) { + if attributes.metadata().name() == self.span_name { + self.threads + .lock() + .expect("span thread log") + .push(std::thread::current().id()); + } + } +} + +/// An acknowledgement publishes its delivery receipt to the recorder's spool +/// with file and directory fsyncs. Measured on an idle macOS disk that is +/// 8 to 13 ms and all of the acknowledge span's busy time, 50 to 80 times any +/// other LSP operation, and it ran on the serving thread under the daemon-wide +/// session map lock. On a disk whose fsyncs stall, that parked the serving +/// worker and every invocation behind the lock until the client's own +/// deadline lapsed. The receipt must still be durable before the +/// acknowledgement is answered, but it must be admitted off the serving thread. +#[tokio::test] +async fn lsp_acknowledgement_admits_its_delivery_receipt_off_the_serving_thread() { + use tracing_subscriber::layer::SubscriberExt; + + let fixture = lsp_delivery_fixture().await; + let service = DaemonInvocationService::default(); + let registry = Arc::new(Mutex::new(LspSessionRegistry::default())); + let session = open_polled_lsp_delivery(&fixture, &service, ®istry, "ack").await; + + let spool_publishers = Arc::new(std::sync::Mutex::new(Vec::new())); + let acknowledged = { + let _capture = tracedecay_runtime_core::logging::set_tracing_capture( + tracing_subscriber::registry().with(SpanThreads { + span_name: "private_fs.framed_log.publish", + threads: Arc::clone(&spool_publishers), + }), + ); + service + .invoke( + ®istry, + None, + None, + None, + None, + DaemonInvocationRequest::lsp_acknowledge( + "request.lsp-delivery.ack.ack", + session, + bridge_lsp_deadline(), + bridge_lsp_cancellation(), + ), + ) + .await + }; + assert!(matches!( + acknowledged.outcome, + DaemonInvocationOutcome::LspAcknowledged { acknowledged: true } + )); + assert!( + !spool_publishers + .lock() + .expect("span thread log") + .contains(&std::thread::current().id()), + "the acknowledgement published its receipt on the serving thread" + ); + + drop(service); + assert_one_lsp_delivery_settlement(fixture, 1, 0).await; +} + #[tokio::test] async fn lsp_detach_after_unacknowledged_outbound_records_disconnected_drop() { let fixture = lsp_delivery_fixture().await;