From 3b855e0fbac927c2e7e7a979f437e67f26a10ecc Mon Sep 17 00:00:00 2001 From: Simon Hachey Date: Mon, 20 Jul 2026 23:14:46 +0000 Subject: [PATCH 1/5] feat(coordinator): add notif-steady/notif-batch channels + affinity-aware recv_for Purely additive priority-queue plumbing for long-running queues Task 1: - Channel enum (Sync/Async/NotifSteady/NotifBatch) replaces Priority - ServedQueue enum + from_str (unknown/"" -> Fast, legacy default) - PriorityQueueSender::send(action, Channel), len(Channel), send_notif_steady/send_notif_batch - PriorityQueueReceiver::recv_for(ServedQueue, metrics) with affinity-aware dispatch; recv() now delegates to recv_for(Fast, ..), preserving the existing biased sync>async behavior and async-queue-time metric recording No subscription routing into the new channels yet (later task). Existing send_sync/send_async/recv/len_sync/len_async callers are unaffected. --- osprey_coordinator/src/priority_queue.rs | 198 ++++++++++++++++++++--- 1 file changed, 179 insertions(+), 19 deletions(-) diff --git a/osprey_coordinator/src/priority_queue.rs b/osprey_coordinator/src/priority_queue.rs index de10a68bf..eae9ee2c3 100644 --- a/osprey_coordinator/src/priority_queue.rs +++ b/osprey_coordinator/src/priority_queue.rs @@ -80,55 +80,112 @@ impl ActionAcker { } } -pub enum Priority { +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum Channel { Sync, Async, + NotifSteady, + NotifBatch, +} + +/// Which channel class a worker connection serves, advertised in +/// `ClientDetails.served_queue`. `Fast` (the default / legacy value) serves the +/// existing `[sync, async]` biased path; the notif classes serve exactly one channel. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum ServedQueue { + Fast, + NotifSteady, + NotifBatch, +} + +impl ServedQueue { + pub fn from_str(s: &str) -> ServedQueue { + match s { + "notif_steady" => ServedQueue::NotifSteady, + "notif_batch" => ServedQueue::NotifBatch, + // "", "fast", or any unknown value => legacy fast path (fail safe). + _ => ServedQueue::Fast, + } + } } #[derive(Clone)] pub struct PriorityQueueSender { sync_sender: async_channel::Sender, async_sender: async_channel::Sender, + notif_steady_sender: async_channel::Sender, + notif_batch_sender: async_channel::Sender, } impl PriorityQueueSender { fn new( sync_sender: async_channel::Sender, async_sender: async_channel::Sender, + notif_steady_sender: async_channel::Sender, + notif_batch_sender: async_channel::Sender, ) -> PriorityQueueSender { PriorityQueueSender { sync_sender, async_sender, + notif_steady_sender, + notif_batch_sender, } } pub fn close(&self) { self.sync_sender.close(); self.async_sender.close(); + self.notif_steady_sender.close(); + self.notif_batch_sender.close(); } pub async fn send_sync( &self, ackable_action: AckableAction, ) -> Result<(), async_channel::SendError> { - self.send(ackable_action, Priority::Sync).await + self.send(ackable_action, Channel::Sync).await } pub async fn send_async( &self, ackable_action: AckableAction, ) -> Result<(), async_channel::SendError> { - self.send(ackable_action, Priority::Async).await + self.send(ackable_action, Channel::Async).await + } + + pub async fn send_notif_steady( + &self, + ackable_action: AckableAction, + ) -> Result<(), async_channel::SendError> { + self.send(ackable_action, Channel::NotifSteady).await } - async fn send( + pub async fn send_notif_batch( &self, ackable_action: AckableAction, - priority: Priority, + ) -> Result<(), async_channel::SendError> { + self.send(ackable_action, Channel::NotifBatch).await + } + + pub async fn send( + &self, + ackable_action: AckableAction, + channel: Channel, ) -> Result<(), async_channel::SendError> { ackable_action.increment_retry_count(); - match priority { - Priority::Sync => self.sync_sender.send(ackable_action).await, - Priority::Async => self.async_sender.send(ackable_action).await, + match channel { + Channel::Sync => self.sync_sender.send(ackable_action).await, + Channel::Async => self.async_sender.send(ackable_action).await, + Channel::NotifSteady => self.notif_steady_sender.send(ackable_action).await, + Channel::NotifBatch => self.notif_batch_sender.send(ackable_action).await, + } + } + + pub fn len(&self, channel: Channel) -> usize { + match channel { + Channel::Sync => self.sync_sender.len(), + Channel::Async => self.async_sender.len(), + Channel::NotifSteady => self.notif_steady_sender.len(), + Channel::NotifBatch => self.notif_batch_sender.len(), } } @@ -153,33 +210,54 @@ impl PriorityQueueSender { pub struct PriorityQueueReceiver { sync_receiver: async_channel::Receiver, async_receiver: async_channel::Receiver, + notif_steady_receiver: async_channel::Receiver, + notif_batch_receiver: async_channel::Receiver, } impl PriorityQueueReceiver { fn new( sync_receiver: async_channel::Receiver, async_receiver: async_channel::Receiver, + notif_steady_receiver: async_channel::Receiver, + notif_batch_receiver: async_channel::Receiver, ) -> PriorityQueueReceiver { PriorityQueueReceiver { sync_receiver, async_receiver, + notif_steady_receiver, + notif_batch_receiver, } } + pub async fn recv( &self, metrics: Arc, + ) -> Result { + self.recv_for(ServedQueue::Fast, metrics).await + } + + /// Affinity-aware receive: a connection only pulls from the channels its pool + /// serves. `Fast` preserves the existing biased sync>async behavior. + pub async fn recv_for( + &self, + served: ServedQueue, + metrics: Arc, ) -> Result { loop { - let result = tokio::select! { - biased; - result = self.sync_receiver.recv() => result, - result = self.async_receiver.recv() => match result { - Ok(ackable_action) => { - metrics.action_time_in_async_queue.record(Instant::now().duration_since(ackable_action.created_at)); - Ok(ackable_action) - } - Err(_) => self.sync_receiver.recv().await + let result = match served { + ServedQueue::Fast => tokio::select! { + biased; + result = self.sync_receiver.recv() => result, + result = self.async_receiver.recv() => match result { + Ok(ackable_action) => { + metrics.action_time_in_async_queue.record(Instant::now().duration_since(ackable_action.created_at)); + Ok(ackable_action) + } + Err(_) => self.sync_receiver.recv().await + }, }, + ServedQueue::NotifSteady => self.notif_steady_receiver.recv().await, + ServedQueue::NotifBatch => self.notif_batch_receiver.recv().await, }; match result { Ok(ackable_action) => { @@ -230,9 +308,21 @@ impl PriorityQueueReceiver { pub fn create_ackable_action_priority_queue() -> (PriorityQueueSender, PriorityQueueReceiver) { let (sync_sender, sync_receiver) = async_channel::unbounded(); let (async_sender, async_receiver) = async_channel::unbounded(); + let (notif_steady_sender, notif_steady_receiver) = async_channel::unbounded(); + let (notif_batch_sender, notif_batch_receiver) = async_channel::unbounded(); ( - PriorityQueueSender::new(sync_sender, async_sender), - PriorityQueueReceiver::new(sync_receiver, async_receiver), + PriorityQueueSender::new( + sync_sender, + async_sender, + notif_steady_sender, + notif_batch_sender, + ), + PriorityQueueReceiver::new( + sync_receiver, + async_receiver, + notif_steady_receiver, + notif_batch_receiver, + ), ) } @@ -263,3 +353,73 @@ pub fn spawn_priority_queue_metrics_worker( AbortOnDrop::new(join_handle) } + +#[cfg(test)] +mod tests { + use super::*; + use crate::coordinator_metrics::OspreyCoordinatorMetrics; + + fn ackable() -> (AckableAction, oneshot::Receiver) { + AckableAction::new(proto::OspreyCoordinatorAction::default()) + } + + #[tokio::test] + async fn served_queue_from_str_defaults_to_fast() { + assert_eq!( + ServedQueue::from_str("notif_steady"), + ServedQueue::NotifSteady + ); + assert_eq!( + ServedQueue::from_str("notif_batch"), + ServedQueue::NotifBatch + ); + assert_eq!(ServedQueue::from_str(""), ServedQueue::Fast); + assert_eq!(ServedQueue::from_str("fast"), ServedQueue::Fast); + assert_eq!(ServedQueue::from_str("garbage"), ServedQueue::Fast); + } + + #[tokio::test] + async fn notif_steady_pool_only_sees_notif_steady() { + let metrics = OspreyCoordinatorMetrics::new(); + let (tx, rx) = create_ackable_action_priority_queue(); + + // Put one action on async and one on notif-steady, with distinct names so + // we can tell which one the notif-steady pool actually received. + let (a_async, _r1) = AckableAction::new(proto::OspreyCoordinatorAction { + action_name: "async_action".into(), + ..Default::default() + }); + let (a_steady, _r2) = AckableAction::new(proto::OspreyCoordinatorAction { + action_name: "notif_steady_action".into(), + ..Default::default() + }); + tx.send(a_async, Channel::Async).await.unwrap(); + tx.send(a_steady, Channel::NotifSteady).await.unwrap(); + + // A NotifSteady stream must receive ONLY the notif-steady action, never the async one. + let got = rx + .recv_for(ServedQueue::NotifSteady, metrics.clone()) + .await + .unwrap(); + assert_eq!(got.action.action_name, "notif_steady_action"); + // async action is still queued (not drained by the notif-steady pool) + assert_eq!(tx.len(Channel::Async), 1); + assert_eq!(tx.len(Channel::NotifSteady), 0); + } + + #[tokio::test] + async fn fast_pool_prefers_sync_then_async() { + let metrics = OspreyCoordinatorMetrics::new(); + let (tx, rx) = create_ackable_action_priority_queue(); + let (a_async, _r1) = ackable(); + let (a_sync, _r2) = ackable(); + tx.send(a_async, Channel::Async).await.unwrap(); + tx.send(a_sync, Channel::Sync).await.unwrap(); + // biased: sync drains first + rx.recv_for(ServedQueue::Fast, metrics.clone()) + .await + .unwrap(); + assert_eq!(tx.len(Channel::Sync), 0); + assert_eq!(tx.len(Channel::Async), 1); + } +} From fb5f5ab10e95f799691741048e801faed0c4d809 Mon Sep 17 00:00:00 2001 From: Simon Hachey Date: Mon, 20 Jul 2026 23:23:54 +0000 Subject: [PATCH 2/5] feat(coordinator): consume multiple subscriptions, one channel each --- osprey_coordinator/src/consumer/pubsub.rs | 12 +-- osprey_coordinator/src/coordinator_metrics.rs | 2 + osprey_coordinator/src/main.rs | 83 +++++++++++++++++-- osprey_coordinator/src/priority_queue.rs | 6 ++ 4 files changed, 92 insertions(+), 11 deletions(-) diff --git a/osprey_coordinator/src/consumer/pubsub.rs b/osprey_coordinator/src/consumer/pubsub.rs index b89ac0e15..5fc76ed62 100644 --- a/osprey_coordinator/src/consumer/pubsub.rs +++ b/osprey_coordinator/src/consumer/pubsub.rs @@ -17,7 +17,9 @@ use crate::metrics::MetricsClientBuilder; use crate::{ consumer::message_decoder, coordinator_metrics::OspreyCoordinatorMetrics, - priority_queue::{AckOrNack, AckableAction, PriorityQueueSender}, + priority_queue::{ + AckOrNack, AckableAction, Channel as PriorityQueueChannel, PriorityQueueSender, + }, proto, pub_sub_streaming_pull::DetachedMessage, pub_sub_streaming_pull::{FlowControl, SpawnTaskPerMessageHandler, StreamingPullManager}, @@ -151,15 +153,14 @@ pub async fn start_pubsub_subscriber( snowflake_client: Arc, priority_queue_sender: PriorityQueueSender, metrics: Arc, + subscription_id: String, + channel: PriorityQueueChannel, ) -> Result<()> { let subscriber_client = create_pubsub_subscription_client().await; let subscription_name = { let project_id = std::env::var("PUBSUB_SUBSCRIPTION_PROJECT_ID").unwrap_or("osprey-dev".to_string()); - let subscription_id = std::env::var("PUBSUB_SUBSCRIPTION_ID") - .unwrap_or("osprey-coordinator-actions".to_string()); - PubSubSubscription::new(project_id, subscription_id) }; @@ -216,6 +217,7 @@ pub async fn start_pubsub_subscriber( let priority_queue_sender = priority_queue_sender.clone(); let snowflake_client = snowflake_client.clone(); let kms_envelope = kms_envelope.clone(); + let channel = channel; async move { let ack_id: u64 = rand::thread_rng().gen(); @@ -242,7 +244,7 @@ pub async fn start_pubsub_subscriber( let send_start_time = Instant::now(); match timeout( max_time_to_send_to_async_queue, - priority_queue_sender.send_async(ackable_action), + priority_queue_sender.send(ackable_action, channel), ) .await { diff --git a/osprey_coordinator/src/coordinator_metrics.rs b/osprey_coordinator/src/coordinator_metrics.rs index 2bb67a3a9..1e5d1c239 100644 --- a/osprey_coordinator/src/coordinator_metrics.rs +++ b/osprey_coordinator/src/coordinator_metrics.rs @@ -11,6 +11,8 @@ define_metrics!(OspreyCoordinatorMetrics, [ // How many messages are currently buffered in the priority queue priority_queue_size_sync => StaticGauge("priority_queue_size",["type" => "sync"]), priority_queue_size_async => StaticGauge("priority_queue_size",["type" => "async"]), + priority_queue_size_notif_steady => StaticGauge("priority_queue_size",["type" => "notif_steady"]), + priority_queue_size_notif_batch => StaticGauge("priority_queue_size",["type" => "notif_batch"]), // How many receivers are open for the priority queue // can be used as a proxy for number of connections open from the osprey worker diff --git a/osprey_coordinator/src/main.rs b/osprey_coordinator/src/main.rs index 0673d1df9..415cdcdd8 100644 --- a/osprey_coordinator/src/main.rs +++ b/osprey_coordinator/src/main.rs @@ -41,6 +41,64 @@ use tokio::join; use crate::osprey_bidirectional_stream::OspreyCoordinatorServer; use crate::proto::osprey_coordinator_service_server::OspreyCoordinatorServiceServer; +/// Parse `OSPREY_COORDINATOR_SUBSCRIPTIONS` = "sub_a:async,sub_b:notif_steady". +/// Falls back to `[(PUBSUB_SUBSCRIPTION_ID, Async)]` when unset (legacy behavior). +fn parse_subscriptions( + raw: Option, + legacy_sub: Option, +) -> Result> { + use priority_queue::Channel; + if let Some(raw) = raw.filter(|s| !s.trim().is_empty()) { + raw.split(',') + .map(|pair| { + let (sub, chan) = pair.split_once(':').ok_or_else(|| { + anyhow::anyhow!("bad subscription spec '{pair}', want 'sub:channel'") + })?; + let channel = match chan.trim() { + "async" => Channel::Async, + "notif_steady" => Channel::NotifSteady, + "notif_batch" => Channel::NotifBatch, + other => anyhow::bail!("unknown channel '{other}'"), + }; + Ok((sub.trim().to_string(), channel)) + }) + .collect() + } else { + let sub = legacy_sub.ok_or_else(|| { + anyhow::anyhow!( + "neither OSPREY_COORDINATOR_SUBSCRIPTIONS nor PUBSUB_SUBSCRIPTION_ID set" + ) + })?; + Ok(vec![(sub, Channel::Async)]) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use priority_queue::Channel; + + #[test] + fn parses_multi_subscription_spec() { + let got = parse_subscriptions(Some("s1:async,s2:notif_steady,s3:notif_batch".into()), None) + .unwrap(); + assert_eq!( + got, + vec![ + ("s1".into(), Channel::Async), + ("s2".into(), Channel::NotifSteady), + ("s3".into(), Channel::NotifBatch), + ] + ); + } + + #[test] + fn falls_back_to_legacy_single_sub() { + let got = parse_subscriptions(None, Some("legacy-sub".into())).unwrap(); + assert_eq!(got, vec![("legacy-sub".into(), Channel::Async)]); + } +} + #[derive(Debug, Parser)] struct CliOptions { #[arg( @@ -117,12 +175,25 @@ async fn main() -> Result<()> { as std::pin::Pin> + Send>> } Some("pubsub") => { - tracing::info!("starting PubSub subscriber"); - Box::pin(start_pubsub_subscriber( - snowflake_client.clone(), - priority_queue_sender.clone(), - metrics.clone(), - )) + let subs = parse_subscriptions( + std::env::var("OSPREY_COORDINATOR_SUBSCRIPTIONS").ok(), + std::env::var("PUBSUB_SUBSCRIPTION_ID").ok(), + )?; + tracing::info!(?subs, "starting PubSub subscribers"); + let mut futs = Vec::new(); + for (sub_id, channel) in subs { + futs.push(start_pubsub_subscriber( + snowflake_client.clone(), + priority_queue_sender.clone(), + metrics.clone(), + sub_id, + channel, + )); + } + Box::pin(async move { + futures::future::try_join_all(futs).await?; + Ok(()) + }) as std::pin::Pin> + Send>> } Some(invalid) => { diff --git a/osprey_coordinator/src/priority_queue.rs b/osprey_coordinator/src/priority_queue.rs index eae9ee2c3..27d5ecadb 100644 --- a/osprey_coordinator/src/priority_queue.rs +++ b/osprey_coordinator/src/priority_queue.rs @@ -342,6 +342,12 @@ pub fn spawn_priority_queue_metrics_worker( metrics .priority_queue_size_async .set(queue_sender.len_async() as u64); + metrics + .priority_queue_size_notif_steady + .set(queue_sender.len(Channel::NotifSteady) as u64); + metrics + .priority_queue_size_notif_batch + .set(queue_sender.len(Channel::NotifBatch) as u64); metrics .priority_queue_receiver_count_async .set(queue_sender.receiver_count_async() as u64); From fe6ce3e092a90c52c9ee3a6651d09158df3f4b9a Mon Sep 17 00:00:00 2001 From: Simon Hachey Date: Mon, 20 Jul 2026 23:29:43 +0000 Subject: [PATCH 3/5] test(coordinator): cover parse_subscriptions error when no subscription configured --- osprey_coordinator/src/main.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/osprey_coordinator/src/main.rs b/osprey_coordinator/src/main.rs index 415cdcdd8..048b4a396 100644 --- a/osprey_coordinator/src/main.rs +++ b/osprey_coordinator/src/main.rs @@ -97,6 +97,11 @@ mod tests { let got = parse_subscriptions(None, Some("legacy-sub".into())).unwrap(); assert_eq!(got, vec![("legacy-sub".into(), Channel::Async)]); } + + #[test] + fn errors_when_no_subscription_configured() { + assert!(parse_subscriptions(None, None).is_err()); + } } #[derive(Debug, Parser)] From e53c1915a17370119097354282276c587ad3e653 Mon Sep 17 00:00:00 2001 From: Simon Hachey Date: Tue, 21 Jul 2026 00:54:47 +0000 Subject: [PATCH 4/5] fix(coordinator): nack notif-steady/notif-batch channels on shutdown --- osprey_coordinator/src/priority_queue.rs | 37 ++++++++++++++++++++++ osprey_coordinator/src/shutdown_handler.rs | 4 ++- 2 files changed, 40 insertions(+), 1 deletion(-) diff --git a/osprey_coordinator/src/priority_queue.rs b/osprey_coordinator/src/priority_queue.rs index 27d5ecadb..da9327deb 100644 --- a/osprey_coordinator/src/priority_queue.rs +++ b/osprey_coordinator/src/priority_queue.rs @@ -289,6 +289,19 @@ impl PriorityQueueReceiver { Self::nack_all(&self.sync_receiver); } + /// Same as `nack_all_sync`/`nack_all_async`, but for the notif-steady + /// channel. Called on shutdown so queued-but-undispatched notif actions + /// (live traffic) are nacked promptly rather than waiting out the + /// pubsub lease (30-600s) for redelivery. + pub fn nack_all_notif_steady(&self) { + Self::nack_all(&self.notif_steady_receiver); + } + + /// Same as `nack_all_notif_steady`, but for the notif-batch channel. + pub fn nack_all_notif_batch(&self) { + Self::nack_all(&self.notif_batch_receiver); + } + fn nack_all(receiver: &async_channel::Receiver) { loop { match receiver.try_recv() { @@ -413,6 +426,30 @@ mod tests { assert_eq!(tx.len(Channel::NotifSteady), 0); } + #[tokio::test] + async fn nack_all_notif_steady_nacks_queued_action() { + let (tx, rx) = create_ackable_action_priority_queue(); + let (action, acking_receiver) = ackable(); + tx.send(action, Channel::NotifSteady).await.unwrap(); + + rx.nack_all_notif_steady(); + + assert!(matches!(acking_receiver.await.unwrap(), AckOrNack::Nack)); + assert_eq!(tx.len(Channel::NotifSteady), 0); + } + + #[tokio::test] + async fn nack_all_notif_batch_nacks_queued_action() { + let (tx, rx) = create_ackable_action_priority_queue(); + let (action, acking_receiver) = ackable(); + tx.send(action, Channel::NotifBatch).await.unwrap(); + + rx.nack_all_notif_batch(); + + assert!(matches!(acking_receiver.await.unwrap(), AckOrNack::Nack)); + assert_eq!(tx.len(Channel::NotifBatch), 0); + } + #[tokio::test] async fn fast_pool_prefers_sync_then_async() { let metrics = OspreyCoordinatorMetrics::new(); diff --git a/osprey_coordinator/src/shutdown_handler.rs b/osprey_coordinator/src/shutdown_handler.rs index e041ac64f..ff3833e4f 100644 --- a/osprey_coordinator/src/shutdown_handler.rs +++ b/osprey_coordinator/src/shutdown_handler.rs @@ -25,7 +25,9 @@ pub fn spawn_shutdown_handler( // pubsub redelivery rather than waiting for the lease to expire. priority_queue_receiver.nack_all_sync(); priority_queue_receiver.nack_all_async(); - tracing::info!("nacked all queued sync + async actions"); + priority_queue_receiver.nack_all_notif_steady(); + priority_queue_receiver.nack_all_notif_batch(); + tracing::info!("nacked all queued sync + async + notif-steady + notif-batch actions"); // Hold the channel open while workers ack dispatched-but-not-yet-acked // actions over bidi. At typical worker latencies of ~150ms p95, 30s // gives ~200x the processing window for in-flight actions to drain From 12abc806d82f07383b2ae385d77ac855e2bcb76a Mon Sep 17 00:00:00 2001 From: Simon Hachey Date: Wed, 22 Jul 2026 20:40:37 +0000 Subject: [PATCH 5/5] chore(coordinator): address review nits (docs + nack log hygiene) --- osprey_coordinator/src/consumer/pubsub.rs | 1 - osprey_coordinator/src/priority_queue.rs | 22 +++++++++++++--------- 2 files changed, 13 insertions(+), 10 deletions(-) diff --git a/osprey_coordinator/src/consumer/pubsub.rs b/osprey_coordinator/src/consumer/pubsub.rs index 5fc76ed62..1837d6166 100644 --- a/osprey_coordinator/src/consumer/pubsub.rs +++ b/osprey_coordinator/src/consumer/pubsub.rs @@ -217,7 +217,6 @@ pub async fn start_pubsub_subscriber( let priority_queue_sender = priority_queue_sender.clone(); let snowflake_client = snowflake_client.clone(); let kms_envelope = kms_envelope.clone(); - let channel = channel; async move { let ack_id: u64 = rand::thread_rng().gen(); diff --git a/osprey_coordinator/src/priority_queue.rs b/osprey_coordinator/src/priority_queue.rs index da9327deb..75450edf7 100644 --- a/osprey_coordinator/src/priority_queue.rs +++ b/osprey_coordinator/src/priority_queue.rs @@ -88,9 +88,10 @@ pub enum Channel { NotifBatch, } -/// Which channel class a worker connection serves, advertised in -/// `ClientDetails.served_queue`. `Fast` (the default / legacy value) serves the -/// existing `[sync, async]` biased path; the notif classes serve exactly one channel. +/// Which channel class a worker connection serves, which a worker WILL +/// advertise in `ClientDetails.served_queue` (added in a follow-up PR). `Fast` +/// (the default / legacy value) serves the existing `[sync, async]` biased +/// path; the notif classes serve exactly one channel. #[derive(Clone, Copy, Debug, PartialEq, Eq)] pub enum ServedQueue { Fast, @@ -290,9 +291,11 @@ impl PriorityQueueReceiver { } /// Same as `nack_all_sync`/`nack_all_async`, but for the notif-steady - /// channel. Called on shutdown so queued-but-undispatched notif actions - /// (live traffic) are nacked promptly rather than waiting out the - /// pubsub lease (30-600s) for redelivery. + /// channel. Without this, a queued-but-undispatched notif action is only + /// released when its in-flight pubsub handler's ack-await times out at + /// `max_acking_receiver_wait_time` (`MAX_ACKING_RECEIVER_WAIT_TIME_MS`, + /// default 60s) or the channel closes and drops the sender. Nacking here + /// releases it immediately on shutdown instead. pub fn nack_all_notif_steady(&self) { Self::nack_all(&self.notif_steady_receiver); } @@ -307,9 +310,10 @@ impl PriorityQueueReceiver { match receiver.try_recv() { Ok(action) => match action.acking_oneshot_sender.send(AckOrNack::Nack) { Ok(_) => (), - Err(_) => println!( - "tried to nack {:?} and the nacking receiver was dropped", - action.action + Err(_) => tracing::warn!( + action_id = action.action.action_id, + action_name = %action.action.action_name, + "tried to nack an action but the nacking receiver was dropped" ), }, Err(_) => return,