fix(lsp): admit acknowledgement receipts off the session lock

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())

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<tracedecay_domain::DeliveryDropReasonV1>,
-    ) -> 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<tracedecay_application::observability::BoundedDeliverySettlementRecorderV1>,
+    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<tracedecay_domain::DeliveryDropReasonV1>,
+    ) -> Result<LspDeliverySettlementClaim, LspDeliverySettlementAdmissionV1> {
+        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();
     }
 }

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<std::sync::Mutex<Vec<std::thread::ThreadId>>>,
+}
+
+impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> 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, &registry, "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(
+                &registry,
+                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;