From f528034599026d5eeae9bd9b067321f399e4a876 Mon Sep 17 00:00:00 2001 From: Zack Jackson <25274700+ScriptedAlchemy@users.noreply.github.com> Date: Tue, 10 Nov 2026 00:13:17 -0800 Subject: [PATCH] fix(lsp): admit acknowledgement receipts off the session lock An LSP frame acknowledgement published its delivery receipt to the recorder's spool (file and directory fsyncs) inline, on the serving thread, while holding the daemon-wide LSP session map lock. Every invocation takes that lock to expire sessions, so each acknowledgement's fsyncs serialized every daemon request behind them and parked a serving worker. Measured with TRACEDECAY_SPAN_TIMINGS on the production LSP journey (runtime_surface_acceptance::production_lsp_negotiates_and_projects_ canonical_context, idle macOS), the acknowledge span was 8-13 ms busy with ~2 us idle, all of it in private_fs.framed_log.publish, against ~160 us for a poll. After the change it is 25-50 us busy; the wait is an await on the blocking pool with no lock held. The receipt is still durable before the acknowledgement is answered. Claim the in-flight attempt under the lock and admit it afterwards on the blocking pool, for acknowledgement and detach. Drop keeps inline admission because it cannot await. Co-Authored-By: Claude Sonnet 5.5 --- .../src/invocation/lsp.rs | 42 ++++++--- .../src/invocation/lsp_delivery.rs | 89 +++++++++++++----- .../src/invocation/tests/lsp_tests.rs | 91 ++++++++++++++++++- 3 files changed, 184 insertions(+), 38 deletions(-) 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;