From 89cf4fb05869a7613f9b07d2c350b33e14979156 Mon Sep 17 00:00:00 2001 From: Jeff Repanich Date: Thu, 3 Sep 2026 11:02:06 -0400 Subject: [PATCH] fix: emit one Queue terminal response --- .../domain_frame_dispatcher.rs | 13 +++ .../tests/domain_backpressure.rs | 85 +++++++++++++++++++ src/domains/queue/sink/ingress.rs | 8 +- src/domains/queue/sink/responses.rs | 8 ++ src/runtime/envelope.rs | 72 +++++++++++++++- 5 files changed, 179 insertions(+), 7 deletions(-) diff --git a/src/api/runtime_ingress/domain_frame_dispatcher.rs b/src/api/runtime_ingress/domain_frame_dispatcher.rs index c4d67d8e..6986e315 100644 --- a/src/api/runtime_ingress/domain_frame_dispatcher.rs +++ b/src/api/runtime_ingress/domain_frame_dispatcher.rs @@ -460,8 +460,19 @@ impl DomainFrameDispatcher<'_> { domain: DispatchDomain, router: &crate::runtime::Router, error: &crate::runtime::router::RouteError, + reply_claim: &crate::runtime::envelope::ReplyClaim, ) -> IngressDecision { obs::counter_inc(obs::METRIC_INGRESS_DOMAIN_DISPATCH_TIMEOUTS); + if domain == DispatchDomain::Queue && !reply_claim.try_claim() { + warn!( + session_id = session_id, + domain = domain.as_str(), + error = %error, + outcome = "domain-response-won", + "Ingress: domain dispatch timed out after its terminal response was claimed" + ); + return IngressDecision::Accept; + } warn!( session_id = session_id, domain = domain.as_str(), @@ -618,6 +629,7 @@ impl DomainFrameDispatcher<'_> { source: source.clone(), destination: addr.clone(), }); + let reply_claim = envelope.reply_claim(); debug!( session_id = session_id, @@ -671,6 +683,7 @@ impl DomainFrameDispatcher<'_> { domain, router, &error, + &reply_claim, )); } Err(error) => { diff --git a/src/api/runtime_ingress/tests/domain_backpressure.rs b/src/api/runtime_ingress/tests/domain_backpressure.rs index 1e6aed96..68da21df 100644 --- a/src/api/runtime_ingress/tests/domain_backpressure.rs +++ b/src/api/runtime_ingress/tests/domain_backpressure.rs @@ -216,6 +216,91 @@ impl MailboxSink for AlwaysTimingOutSink { } } +struct QueueReplyThenTimeoutSink { + router: Arc, + session_id: u64, +} + +impl MailboxSink for QueueReplyThenTimeoutSink { + fn deliver(&self, envelope: Envelope) -> Result<(), DeliveryError> { + assert!(envelope.try_claim_reply(), "queue response claim"); + let response = envelope + .try_reply_to(FrameContext::new( + self.session_id, + ChannelId::Pub, + crate::protocol::tlv::MessageType::new(200), + Bytes::from_static(&[0]), + RouteFamily::new(1), + )) + .expect("queue response envelope"); + self.router.route(response).expect("route queue response"); + Err(DeliveryError::Timeout) + } + + fn deliver_high_priority(&self, envelope: Envelope) -> Result<(), DeliveryError> { + self.deliver(envelope) + } +} + +#[test] +fn should_not_emit_a_second_queue_terminal_response_after_the_domain_replied() { + // Arrange + // A Queue command can finish at the same instant its mailbox reply wait + // expires. Its response and ingress' indeterminate timeout compete for one + // terminal-response slot; emitting both shifts the client's per-type FIFO + // and can make a later accepted enqueue look retryably rejected. + let rt = tokio::runtime::Runtime::new().unwrap(); + let router = Arc::new(crate::runtime::Router::new()); + let session_id = 6_500; + let client_frames = Arc::new(Mutex::new(Vec::::new())); + router.register( + crate::runtime::routing::RouteAddress::new( + RouteFamily::new(1), + crate::runtime::routing::Route::new(format!("inbox://session/{session_id}")), + ), + Arc::new(CapturingInboxSink { + frames: client_frames.clone(), + }) as Arc, + ); + router.register_domain_pattern( + "queue", + Arc::new(QueueReplyThenTimeoutSink { + router: router.clone(), + session_id, + }), + ); + let ingress = RuntimeIngress::new(false).with_router(router); + let (_, payload) = crate::benchkit::extract_single_tlv_field( + &crate::benchkit::build_queue_enqueue("queue://test/app/jobs", b"job"), + ); + + // Act + let decision = rt.block_on(async { + ingress + .on_open(make_session_info(session_id, TransportKind::Tcp)) + .await + .unwrap(); + ingress + .on_frame( + session_id, + ChannelId::Pub, + crate::protocol::tlv::MessageType::new(200), + payload, + ) + .await + }); + + // Assert + assert_eq!(decision, IngressDecision::Accept); + let frames = client_frames.lock().unwrap(); + assert_eq!( + frames.len(), + 1, + "one request must produce exactly one terminal response" + ); + assert_eq!(frames[0].payload.as_ref(), &[0]); +} + #[test] fn should_surface_sustained_high_lane_domain_mailbox_backpressure_for_each_domain() { // Arrange diff --git a/src/domains/queue/sink/ingress.rs b/src/domains/queue/sink/ingress.rs index 160f9b11..8175137a 100644 --- a/src/domains/queue/sink/ingress.rs +++ b/src/domains/queue/sink/ingress.rs @@ -217,7 +217,7 @@ impl QueueDomainCore { crate::domains::queue::QueueResponse::ReceivedRouted { messages } if messages.is_empty() ) { if let Some(wait_seconds) = wait_seconds.filter(|seconds| *seconds > 0) { - if let Some(source) = envelope.source() { + if envelope.source().is_some() { let mut message = pending_message; if let crate::domains::queue::protocol::QueueMessage::Receive { wait_seconds, @@ -230,11 +230,7 @@ impl QueueDomainCore { .checked_add(Duration::from_secs(wait_seconds)) .unwrap_or_else(Instant::now); self.pending_reserves.lock().push_back(PendingQueueReserve { - envelope: Envelope::from_route( - source.clone(), - envelope.destination().clone(), - (), - ), + envelope: envelope.clone_for_deferred_reply(), meta, request_started, message, diff --git a/src/domains/queue/sink/responses.rs b/src/domains/queue/sink/responses.rs index 5141dee1..b862d6cd 100644 --- a/src/domains/queue/sink/responses.rs +++ b/src/domains/queue/sink/responses.rs @@ -19,6 +19,14 @@ impl QueueDomainCore { meta: crate::runtime::ClientFrameMeta, response: &crate::domains::queue::QueueResponse, ) -> bool { + if request_envelope.source().is_none() || !request_envelope.try_claim_reply() { + tracing::debug!( + domain = "queue", + session = meta.session_id, + "Suppressed Queue response after another terminal response won" + ); + return false; + } let delivered; #[cfg(test)] { diff --git a/src/runtime/envelope.rs b/src/runtime/envelope.rs index 39ca31e9..93f36588 100644 --- a/src/runtime/envelope.rs +++ b/src/runtime/envelope.rs @@ -32,7 +32,8 @@ use crate::runtime::routing::RouteAddress; use std::any::Any; use std::fmt; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::Arc; use std::time::Instant; /// Envelope metadata without the payload (for zero-copy causation tracking) @@ -116,6 +117,9 @@ pub struct Envelope { /// Mailbox enqueue time, set when the envelope is accepted by a mailbox. queued_at: Option, + /// Shared one-shot claim for the terminal response to this request. + reply_claim: ReplyClaim, + /// Type-erased message payload (must be Send + Sync) payload: Box, } @@ -130,6 +134,7 @@ impl Envelope { causation: None, deadline: None, queued_at: None, + reply_claim: ReplyClaim::default(), payload: Box::new(payload), } } @@ -147,6 +152,7 @@ impl Envelope { causation: None, deadline: None, queued_at: None, + reply_claim: ReplyClaim::default(), payload: Box::new(payload), } } @@ -193,6 +199,7 @@ impl Envelope { causation: Some(self.id), deadline: self.deadline, queued_at: None, + reply_claim: ReplyClaim::default(), payload: Box::new(payload), } } @@ -211,10 +218,38 @@ impl Envelope { causation: Some(self.id), deadline: self.deadline, queued_at: None, + reply_claim: ReplyClaim::default(), payload: Box::new(payload), }) } + /// Share this request's one-shot terminal-response claim with its owner. + #[inline] + pub(crate) fn reply_claim(&self) -> ReplyClaim { + self.reply_claim.clone() + } + + /// Claim the right to emit this request's terminal response. + #[inline] + pub(crate) fn try_claim_reply(&self) -> bool { + self.reply_claim.try_claim() + } + + /// Retain this request's routing and response identity for a deferred reply. + #[must_use] + pub(crate) fn clone_for_deferred_reply(&self) -> Self { + Self { + id: self.id, + source: self.source.clone(), + destination: self.destination.clone(), + causation: self.causation, + deadline: self.deadline, + queued_at: None, + reply_claim: self.reply_claim.clone(), + payload: Box::new(()), + } + } + /// Get the message ID #[inline] #[must_use] @@ -321,6 +356,20 @@ impl Envelope { } } +/// Coordinates the single terminal response for a request that may outlive +/// the ingress dispatch deadline. +#[derive(Clone, Debug, Default)] +pub(crate) struct ReplyClaim(Arc); + +impl ReplyClaim { + #[inline] + pub(crate) fn try_claim(&self) -> bool { + self.0 + .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) + .is_ok() + } +} + impl fmt::Debug for Envelope { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { f.debug_struct("Envelope") @@ -330,6 +379,7 @@ impl fmt::Debug for Envelope { .field("causation", &self.causation) .field("deadline", &self.deadline) .field("queued_at", &self.queued_at) + .field("reply_claim", &self.reply_claim) .field("payload", &"") .finish() } @@ -514,4 +564,24 @@ mod tests { // Assert assert_eq!(reply.deadline(), Some(deadline)); } + + #[test] + fn should_share_one_terminal_reply_claim_with_deferred_context() { + // Arrange + let original = Envelope::from_route( + test_address(1, "/test/source"), + test_address(1, "/test/destination"), + "request", + ); + let deferred = original.clone_for_deferred_reply(); + let owner = original.reply_claim(); + + // Act + let first = deferred.try_claim_reply(); + let second = owner.try_claim(); + + // Assert + assert!(first); + assert!(!second, "the request may emit only one terminal response"); + } }