From 56f2410eea5a5c200869d3ae673d519ee3d98d2e Mon Sep 17 00:00:00 2001 From: forkwright Date: Fri, 21 Aug 2026 01:14:04 -0500 Subject: [PATCH 1/3] fix(kerykeion): bound the outbound pending queue max_inflight bounded only what awaits an ACK. The pending queue behind it had no bound at all, so anything enqueuing faster than the radio drains -- a chatty caller, or a store-and-forward backlog arriving at once -- grew memory without limit. This was #229's last unbounded-growth clause. enqueue now refuses past max_pending and hands the message back rather than dropping it, because the two callers want different things and neither can act on a message the queue has already discarded. send reports the refusal through the existing QueueFull error. The caller asked to send; a queue that silently swallowed the message would look exactly like one that accepted it. The store-and-forward drain is the case worth care. drain_for removes everything it returns, so enqueuing until the queue refused would have destroyed the remainder. It now takes only what fits and returns the rest to store-and-forward, so a saturated queue delays delivery instead of losing it. The fourteen test call sites go through a helper that asserts acceptance rather than discarding the new Result. Discarding at each site would have hidden a regression that makes the queue refuse everything -- which is exactly what the anti-vacuity test exists to catch. Refs #229 --- crates/kerykeion/src/config.rs | 11 +++ crates/kerykeion/src/outbound.rs | 163 ++++++++++++++++++++++++------- crates/kerykeion/src/router.rs | 64 +++++++++--- 3 files changed, 190 insertions(+), 48 deletions(-) diff --git a/crates/kerykeion/src/config.rs b/crates/kerykeion/src/config.rs index c809d65..cc6d195 100644 --- a/crates/kerykeion/src/config.rs +++ b/crates/kerykeion/src/config.rs @@ -281,6 +281,13 @@ impl BridgeConfig { pub struct OutboundConfig { /// Maximum number of concurrent inflight (awaiting-ACK) messages. pub max_inflight: usize, + /// Maximum number of messages waiting to be sent. + /// + /// WHY(#229): `max_inflight` bounds only what is awaiting an ACK. The + /// pending queue behind it had no bound at all, so anything that enqueues + /// faster than the radio drains — a chatty caller, or a store-and-forward + /// backlog arriving at once — grew memory without limit. + pub max_pending: usize, /// Maximum retry attempts per message before declaring delivery failure. pub max_retries: u8, /// Default ACK timeout for inflight messages, in seconds. @@ -301,6 +308,10 @@ impl Default for OutboundConfig { fn default() -> Self { Self { max_inflight: 8, + // WHY this size: well above any burst the radio can clear in a + // session, and small enough that a saturated queue is bounded + // memory rather than a leak. + max_pending: 1024, max_retries: 5, ack_timeout_secs: 30, store_forward_ttl_secs: 3600, diff --git a/crates/kerykeion/src/outbound.rs b/crates/kerykeion/src/outbound.rs index 17e8d5c..971d22f 100644 --- a/crates/kerykeion/src/outbound.rs +++ b/crates/kerykeion/src/outbound.rs @@ -82,6 +82,7 @@ pub struct OutboundQueue { pending: VecDeque, inflight: HashMap, max_inflight: usize, + max_pending: usize, max_retries: u8, } @@ -99,6 +100,7 @@ impl OutboundQueue { pending: VecDeque::new(), inflight: HashMap::new(), max_inflight: config.max_inflight, + max_pending: config.max_pending, max_retries: config.max_retries, } } @@ -110,6 +112,7 @@ impl OutboundQueue { pending: VecDeque::new(), inflight: HashMap::new(), max_inflight, + max_pending: OutboundConfig::default().max_pending, max_retries: OutboundConfig::default().max_retries, } } @@ -126,14 +129,33 @@ impl OutboundQueue { self.max_retries } + /// Room left in the pending queue. + #[must_use] + pub fn remaining_capacity(&self) -> usize { + self.max_pending.saturating_sub(self.pending.len()) + } + /// Insert a message by priority (higher priority first). - pub fn enqueue(&mut self, msg: PendingMessage) { + /// + /// # Errors + /// + /// Returns the message back when the pending queue is at + /// [`OutboundConfig::max_pending`]. WHY(#229) it is handed back rather than + /// dropped: the two callers want different things — one reports the refusal + /// to whoever asked to send, the other returns the message to + /// store-and-forward — and neither can do that with a message this queue + /// has already discarded. + pub fn enqueue(&mut self, msg: PendingMessage) -> Result<(), PendingMessage> { + if self.pending.len() >= self.max_pending { + return Err(msg); + } let insert_pos = self .pending .iter() .position(|existing| i32::from(existing.priority) < i32::from(msg.priority)) .unwrap_or(self.pending.len()); self.pending.insert(insert_pos, msg); + Ok(()) } /// Pop the highest-priority message that hasn't expired. @@ -275,6 +297,18 @@ impl Default for OutboundQueue { mod tests { use super::*; + /// Enqueue in a test, asserting the queue accepted it. + /// + /// WHY assert rather than discard: `enqueue` returns a `Result` now, and + /// `let _ =` at fourteen call sites would hide a regression that makes the + /// queue refuse everything. + fn enq(q: &mut OutboundQueue, msg: PendingMessage) { + assert!( + q.enqueue(msg).is_ok(), + "the queue should accept this message" + ); + } + const DEFAULT_ACK_TIMEOUT: Duration = Duration::from_secs(30); fn make_packet(id: u32, priority: Priority) -> MeshPacket { @@ -305,12 +339,66 @@ mod tests { } } + /// WHY(#229): `max_inflight` bounded only what awaits an ACK. The pending + /// queue behind it had no bound, so anything enqueuing faster than the radio + /// drains grew memory without limit. + #[test] + fn the_pending_queue_refuses_past_its_bound() { + let mut q = OutboundQueue::with_config(&OutboundConfig { + max_pending: 4, + ..OutboundConfig::default() + }); + + for id in 0..4 { + enq(&mut q, make_pending(id, Priority::Default)); + } + assert_eq!( + q.remaining_capacity(), + 0, + "four messages fill a bound of four" + ); + + let refused = q.enqueue(make_pending(99, Priority::Default)); + assert!(refused.is_err(), "the fifth must be refused, not accepted"); + } + + /// The refused message is handed back rather than dropped, which is what + /// lets a caller report it or return it to store-and-forward. + #[test] + fn a_refused_message_is_returned_to_its_caller() { + let mut q = OutboundQueue::with_config(&OutboundConfig { + max_pending: 1, + ..OutboundConfig::default() + }); + enq(&mut q, make_pending(1, Priority::Default)); + + let returned = q.enqueue(make_pending(7, Priority::Default)).err(); + assert_eq!( + returned.map(|msg| msg.packet.id), + Some(7), + "the caller must get back the message it offered" + ); + } + + /// Anti-vacuity: a queue within its bound must still accept, or the two + /// cases above would pass against a queue that refuses everything. + #[test] + fn a_queue_within_its_bound_still_accepts() { + let mut q = OutboundQueue::with_config(&OutboundConfig::default()); + assert!(q.remaining_capacity() > 0); + enq(&mut q, make_pending(1, Priority::Default)); + assert_eq!( + q.remaining_capacity(), + OutboundConfig::default().max_pending - 1 + ); + } + #[test] fn enqueue_orders_by_priority() { let mut q = OutboundQueue::new(); - q.enqueue(make_pending(1, Priority::Background)); - q.enqueue(make_pending(2, Priority::Reliable)); - q.enqueue(make_pending(3, Priority::Default)); + enq(&mut q, make_pending(1, Priority::Background)); + enq(&mut q, make_pending(2, Priority::Reliable)); + enq(&mut q, make_pending(3, Priority::Default)); #[expect(clippy::unwrap_used, reason = "test-only: queue has 3 items")] let first = q.next_to_send().unwrap(); @@ -328,20 +416,26 @@ mod tests { #[tokio::test(start_paused = true)] async fn expired_messages_skipped() { let mut q = OutboundQueue::new(); - q.enqueue(PendingMessage { - packet: make_packet(1, Priority::Default), - created: Instant::now(), - ttl: Duration::from_secs(1), - priority: Priority::Default, - retries: 0, - }); - q.enqueue(PendingMessage { - packet: make_packet(2, Priority::Default), - created: Instant::now(), - ttl: Duration::from_secs(3600), - priority: Priority::Default, - retries: 0, - }); + enq( + &mut q, + PendingMessage { + packet: make_packet(1, Priority::Default), + created: Instant::now(), + ttl: Duration::from_secs(1), + priority: Priority::Default, + retries: 0, + }, + ); + enq( + &mut q, + PendingMessage { + packet: make_packet(2, Priority::Default), + created: Instant::now(), + ttl: Duration::from_secs(3600), + priority: Priority::Default, + retries: 0, + }, + ); // Advance past the first message's TTL. tokio::time::advance(Duration::from_secs(2)).await; @@ -404,13 +498,16 @@ mod tests { // TTL; a short-lived message must still expire at its ORIGINAL // deadline after being retried. let mut q = OutboundQueue::new(); - q.enqueue(PendingMessage { - packet: make_packet(1, Priority::Default), - created: Instant::now(), - ttl: Duration::from_secs(60), - priority: Priority::Default, - retries: 0, - }); + enq( + &mut q, + PendingMessage { + packet: make_packet(1, Priority::Default), + created: Instant::now(), + ttl: Duration::from_secs(60), + priority: Priority::Default, + retries: 0, + }, + ); #[expect(clippy::unwrap_used, reason = "test-only: queue has 1 item")] let msg = q.next_to_send().unwrap(); @@ -461,9 +558,9 @@ mod tests { #[test] fn max_inflight_limits_sends() { let mut q = OutboundQueue::with_max_inflight(2); - q.enqueue(make_pending(1, Priority::Default)); - q.enqueue(make_pending(2, Priority::Default)); - q.enqueue(make_pending(3, Priority::Default)); + enq(&mut q, make_pending(1, Priority::Default)); + enq(&mut q, make_pending(2, Priority::Default)); + enq(&mut q, make_pending(3, Priority::Default)); // Pop and track two as inflight. for _ in 0..2 { @@ -491,7 +588,7 @@ mod tests { priority: Priority::Default, retries: 0, }); - q.enqueue(make_pending(2, Priority::Default)); + enq(&mut q, make_pending(2, Priority::Default)); q.drain_expired(); assert_eq!(q.pending_count(), 1, "expired message should be removed"); @@ -527,8 +624,8 @@ mod tests { ..OutboundConfig::default() }; let mut q = OutboundQueue::with_config(&cfg); - q.enqueue(make_pending(1, Priority::Default)); - q.enqueue(make_pending(2, Priority::Default)); + enq(&mut q, make_pending(1, Priority::Default)); + enq(&mut q, make_pending(2, Priority::Default)); #[expect(clippy::unwrap_used, reason = "test-only")] let msg = q.next_to_send().unwrap(); @@ -542,8 +639,8 @@ mod tests { #[test] fn alert_priority_sent_before_default() { let mut q = OutboundQueue::new(); - q.enqueue(make_pending(1, Priority::Default)); - q.enqueue(make_pending(2, Priority::Ack)); + enq(&mut q, make_pending(1, Priority::Default)); + enq(&mut q, make_pending(2, Priority::Ack)); #[expect(clippy::unwrap_used, reason = "test-only: queue has items")] let first = q.next_to_send().unwrap(); diff --git a/crates/kerykeion/src/router.rs b/crates/kerykeion/src/router.rs index aa4bf6e..2d934ee 100644 --- a/crates/kerykeion/src/router.rs +++ b/crates/kerykeion/src/router.rs @@ -13,6 +13,7 @@ use prost::Message as _; use crate::config::OutboundConfig; use crate::delivery::{DeliveryFailure, DeliveryTracker}; +use crate::error::QueueFullSnafu; use crate::outbound::{OutboundQueue, PendingMessage}; use crate::proto::MeshPacket; use crate::proto::mesh_packet::{PayloadVariant, Priority}; @@ -147,13 +148,22 @@ impl MeshRouter { let dest = packet.to; if reachable { - self.outbound.enqueue(PendingMessage { - packet, - created: tokio::time::Instant::now(), - ttl: Duration::from_secs(options.ttl_secs), - priority: options.priority, - retries: 0, - }); + // WHY(#229) the refusal is reported: the caller asked to send. A + // full queue that silently swallowed the message would look + // identical to one that accepted it. + if self + .outbound + .enqueue(PendingMessage { + packet, + created: tokio::time::Instant::now(), + ttl: Duration::from_secs(options.ttl_secs), + priority: options.priority, + retries: 0, + }) + .is_err() + { + return QueueFullSnafu { dest }.fail(); + } } else { let portnum = match &packet.payload_variant { Some(PayloadVariant::Decoded(data)) => data.portnum, @@ -262,7 +272,15 @@ impl MeshRouter { /// drives this from packet receipt (verified: only `MeshRouter`'s own /// tests call it); keep it that way. pub fn node_came_online(&mut self, dest: NodeNum) { - let stored = self.store_forward.drain_for(dest); + // WHY(#229) only as many as will fit are drained: `drain_for` removes + // everything it returns, so enqueuing until the queue refuses would + // destroy the remainder. Taking what fits and returning the rest to + // store-and-forward means a saturated outbound queue delays delivery + // rather than losing it. + let mut stored = self.store_forward.drain_for(dest); + let capacity = self.outbound.remaining_capacity(); + let deferred = stored.split_off(stored.len().min(capacity)); + for msg in stored { let priority = Priority::try_from(msg.priority).unwrap_or(Priority::Default); // WARNING: packet_bytes is only ever written by `send` on this @@ -271,13 +289,29 @@ impl MeshRouter { let Ok(packet) = MeshPacket::decode(msg.packet_bytes.as_slice()) else { continue; }; - self.outbound.enqueue(PendingMessage { - packet, - created: tokio::time::Instant::now(), - ttl: Duration::from_secs(msg.ttl_secs), - priority, - retries: 0, - }); + if self + .outbound + .enqueue(PendingMessage { + packet, + created: tokio::time::Instant::now(), + ttl: Duration::from_secs(msg.ttl_secs), + priority, + retries: 0, + }) + .is_err() + { + tracing::warn!( + dest = dest.0, + "outbound queue filled mid-drain; remaining messages stay stored" + ); + break; + } + } + + for msg in deferred { + if let Err(error) = self.store_forward.store(dest, msg) { + tracing::warn!(dest = dest.0, %error, "could not re-store a deferred message"); + } } } From 1fc90f8dc094338f18e1677accf2c201d13b559b Mon Sep 17 00:00:00 2001 From: forkwright Date: Fri, 21 Aug 2026 01:21:08 -0500 Subject: [PATCH 2/3] fix(kerykeion): honour the pending bound on retry, and complete the config fixture Two call sites the first pass missed. The retry requeue is inside outbound.rs itself, so the grep that found the router's callers excluded it -- searching for callers while filtering out the defining file hides the ones that live beside the definition. Its contract is 'will it be retried', so a refused requeue now returns false rather than promising an attempt that will not happen. One OutboundConfig literal spells out every field instead of using struct update, so it needed the new one. --- crates/kerykeion/src/config.rs | 1 + crates/kerykeion/src/outbound.rs | 8 ++++++-- 2 files changed, 7 insertions(+), 2 deletions(-) diff --git a/crates/kerykeion/src/config.rs b/crates/kerykeion/src/config.rs index cc6d195..70840ed 100644 --- a/crates/kerykeion/src/config.rs +++ b/crates/kerykeion/src/config.rs @@ -703,6 +703,7 @@ stale_node_timeout_secs = 7200 }, outbound: OutboundConfig { max_inflight: 2, + max_pending: 3, max_retries: 1, ack_timeout_secs: 7, store_forward_ttl_secs: 11, diff --git a/crates/kerykeion/src/outbound.rs b/crates/kerykeion/src/outbound.rs index 971d22f..14dde9c 100644 --- a/crates/kerykeion/src/outbound.rs +++ b/crates/kerykeion/src/outbound.rs @@ -257,14 +257,18 @@ impl OutboundQueue { // INVARIANT: `created`/`ttl` are the ORIGINAL enqueue time and configured // TTL, carried forward unchanged so the message expires at its originally // configured deadline regardless of how many retries it goes through. + // WHY(#229) the refusal is returned rather than ignored: this function's + // contract is "will it be retried", and a full queue means it will not. + // Reporting `true` after failing to requeue would promise a delivery + // attempt that never happens. self.enqueue(PendingMessage { packet: msg.packet, created: msg.created, ttl: msg.ttl, priority, retries: msg.retries, - }); - true + }) + .is_ok() } /// Remove messages past TTL FROM both pending and inflight. From 68c03657488f6613045c78a841b960cde1930665 Mon Sep 17 00:00:00 2001 From: forkwright Date: Fri, 21 Aug 2026 01:26:55 -0500 Subject: [PATCH 3/3] fix(kerykeion): box the refused message in enqueue's error clippy::result_large_err: PendingMessage carries a whole packet, and an Err variant that large is paid for by every Result in the call chain rather than only on the refusal. Boxing moves the cost onto the path that is already the exception. --- crates/kerykeion/src/outbound.rs | 9 +++++++-- 1 file changed, 7 insertions(+), 2 deletions(-) diff --git a/crates/kerykeion/src/outbound.rs b/crates/kerykeion/src/outbound.rs index 14dde9c..8e23100 100644 --- a/crates/kerykeion/src/outbound.rs +++ b/crates/kerykeion/src/outbound.rs @@ -145,9 +145,14 @@ impl OutboundQueue { /// to whoever asked to send, the other returns the message to /// store-and-forward — and neither can do that with a message this queue /// has already discarded. - pub fn enqueue(&mut self, msg: PendingMessage) -> Result<(), PendingMessage> { + /// + /// WHY boxed: a [`PendingMessage`] carries a whole packet, and an `Err` + /// variant that large is paid for by every `Result` in the call chain + /// rather than only on the refusal. The allocation happens on the path that + /// is already the exception. + pub fn enqueue(&mut self, msg: PendingMessage) -> Result<(), Box> { if self.pending.len() >= self.max_pending { - return Err(msg); + return Err(Box::new(msg)); } let insert_pos = self .pending