From dbad259ae20c4e97b38172d2f7454f595c3e2cba Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fabr=C3=ADcio=20Bracht?= Date: Sun, 27 Sep 2026 19:29:15 -0300 Subject: [PATCH 1/2] drop stale deliveries routed to a discarded session --- CHANGELOG.md | 10 ++ crates/mqtt5/Cargo.toml | 2 +- crates/mqtt5/src/broker/router.rs | 154 +++++++++++++++--- .../mqtt5/src/broker/storage/client_queue.rs | 77 +++++++-- 4 files changed, 205 insertions(+), 38 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 92bb8d88..9cf1781a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,16 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [mqtt5 0.43.1] - 2026-09-27 + +### Fixed + +- **A clean-start reconnect no longer receives a message routed to the session it replaced.** A publish routed to a client just before that client reconnected with `clean_start=1` could land in the new session's queue after the queue had been cleared, delivering a message for a subscription the new session never made (#150). Each client queue now counts its full clears, and a message routed before the latest one is dropped. A resumed session still receives it. + +### Added + +- `ClientQueue::epoch` and `ClientQueue::push_in_epoch`. + ## [mqtt5 0.43.0] - 2026-09-27 ### Breaking diff --git a/crates/mqtt5/Cargo.toml b/crates/mqtt5/Cargo.toml index 50a0b4e7..5cf4977b 100644 --- a/crates/mqtt5/Cargo.toml +++ b/crates/mqtt5/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "mqtt5" -version = "0.43.0" +version = "0.43.1" edition.workspace = true rust-version.workspace = true authors.workspace = true diff --git a/crates/mqtt5/src/broker/router.rs b/crates/mqtt5/src/broker/router.rs index 15713b02..84428df9 100644 --- a/crates/mqtt5/src/broker/router.rs +++ b/crates/mqtt5/src/broker/router.rs @@ -326,12 +326,14 @@ enum DeliveryPlan { target_flow: Option, lanes: DeliveryLanes, queue: QueueHandle, + epoch: u64, }, Behind { client_id: String, message: PublishPacket, target_flow: Option, queue: QueueHandle, + epoch: u64, }, } @@ -1418,12 +1420,7 @@ impl MessageRouter { if self.storage.is_some() && sub.qos != QoS::AtMostOnce { let mut message = publish.clone(); message.qos = sub.qos; - plans.push(DeliveryPlan::Behind { - client_id: sub.client_id.clone(), - message, - target_flow: sub.flow_id, - queue: self.queue_handle(&sub.client_id), - }); + plans.push(self.plan_behind(sub, message)); } } } @@ -1478,15 +1475,21 @@ impl MessageRouter { fn queue_behind( queue: &QueueHandle, + epoch: u64, message: PublishPacket, client_id: &str, target_flow: Option, ) { let qos = message.qos; - let outcome = queue.push( - QueuedMessage::new(message, client_id.to_string(), qos, None) - .with_target_flow(target_flow), - ); + let queued = QueuedMessage::new(message, client_id.to_string(), qos, None) + .with_target_flow(target_flow); + let Some(outcome) = queue.push_in_epoch(queued, epoch) else { + debug!( + client_id, + "Dropped message routed to a session that has since been discarded" + ); + return; + }; queue.notify(); trace!( client_id, @@ -1507,13 +1510,15 @@ impl MessageRouter { message, target_flow, queue, - } => Self::queue_behind(&queue, message, &client_id, target_flow), + epoch, + } => Self::queue_behind(&queue, epoch, message, &client_id, target_flow), DeliveryPlan::Online { client_id, message, target_flow, lanes, queue, + epoch, } => { let routable = RoutableMessage { publish: message, @@ -1526,7 +1531,13 @@ impl MessageRouter { return; } if queue.behind() { - Self::queue_behind(&queue, routable.publish, &client_id, routable.target_flow); + Self::queue_behind( + &queue, + epoch, + routable.publish, + &client_id, + routable.target_flow, + ); return; } let routable = match lanes.qos1_tx.try_send(routable) { @@ -1534,6 +1545,7 @@ impl MessageRouter { Err(mpsc::error::TrySendError::Closed(rejected)) => { Self::queue_behind( &queue, + epoch, rejected.publish, &client_id, rejected.target_flow, @@ -1543,7 +1555,13 @@ impl MessageRouter { Err(mpsc::error::TrySendError::Full(rejected)) => rejected, }; if publishing_client_id == Some(client_id.as_str()) { - Self::queue_behind(&queue, routable.publish, &client_id, routable.target_flow); + Self::queue_behind( + &queue, + epoch, + routable.publish, + &client_id, + routable.target_flow, + ); return; } let permit = match deadline { @@ -1557,6 +1575,7 @@ impl MessageRouter { Some(permit) => permit.send(routable), None => Self::queue_behind( &queue, + epoch, routable.publish, &client_id, routable.target_flow, @@ -1621,12 +1640,7 @@ impl MessageRouter { if qos == QoS::AtMostOnce || self.storage.is_none() { return None; } - return Some(DeliveryPlan::Behind { - client_id: sub.client_id.clone(), - message: Self::prepare_message(publish, sub, qos), - target_flow: sub.flow_id, - queue: self.queue_handle(&sub.client_id), - }); + return Some(self.plan_behind(sub, Self::prepare_message(publish, sub, qos))); } } @@ -1649,6 +1663,7 @@ impl MessageRouter { qos0_tx: client_info.qos0_tx.clone(), }, queue: Arc::clone(&client_info.queue), + epoch: client_info.queue.epoch(), }); } @@ -1662,12 +1677,18 @@ impl MessageRouter { ); return None; } - Some(DeliveryPlan::Behind { + Some(self.plan_behind(sub, Self::prepare_message(publish, sub, sub.qos))) + } + + fn plan_behind(&self, sub: &Subscription, message: PublishPacket) -> DeliveryPlan { + let queue = self.queue_handle(&sub.client_id); + DeliveryPlan::Behind { client_id: sub.client_id.clone(), - message: Self::prepare_message(publish, sub, sub.qos), + message, target_flow: sub.flow_id, - queue: self.queue_handle(&sub.client_id), - }) + epoch: queue.epoch(), + queue, + } } pub async fn get_retained_messages(&self, topic_filter: &str) -> Vec { @@ -2640,6 +2661,93 @@ mod tests { .generation } + async fn stall_publish_behind_full_lane( + router: &Arc, + client_id: &str, + lanes: &TestLanes, + ) -> tokio::task::JoinHandle<()> { + register(router, client_id, lanes).await; + router + .subscribe(SubscriptionRequest::new(client_id, "t", QoS::AtLeastOnce)) + .await + .unwrap(); + let first = PublishPacket::new("t", &b"first"[..], QoS::AtLeastOnce); + router.route_message(&first, None).await; + + let stalled = Arc::clone(router); + let handle = tokio::spawn(async move { + let second = PublishPacket::new("t", &b"second"[..], QoS::AtLeastOnce); + stalled + .route_message_with_deadline( + &second, + None, + Instant::now() + Duration::from_secs(30), + ) + .await; + }); + for _ in 0..10 { + tokio::task::yield_now().await; + } + assert!( + !handle.is_finished(), + "the publish must wait on the full lane" + ); + handle + } + + #[tokio::test] + async fn clean_start_drops_a_publish_planned_for_the_previous_session() { + let router = Arc::new(MessageRouter::new()); + let previous = TestLanes::new(1); + let stalled = stall_publish_behind_full_lane(&router, "c", &previous).await; + + let current = TestLanes::new(1); + let queue = router.queue_handle("c"); + let (dtx, _drx) = tokio::sync::oneshot::channel(); + router + .register_session( + "c".to_string(), + current.lanes(), + Arc::clone(&queue), + dtx, + true, + ) + .await; + queue.clear(None); + drop(previous); + stalled.await.unwrap(); + + assert_eq!( + queue.count(), + 0, + "a publish routed under the previous session must not reach the clean session" + ); + } + + #[tokio::test] + async fn resumed_session_keeps_a_publish_planned_for_the_previous_connection() { + let router = Arc::new(MessageRouter::new()); + let previous = TestLanes::new(1); + let stalled = stall_publish_behind_full_lane(&router, "c", &previous).await; + + let current = TestLanes::new(1); + let queue = router.queue_handle("c"); + let (dtx, _drx) = tokio::sync::oneshot::channel(); + router + .register_session( + "c".to_string(), + current.lanes(), + Arc::clone(&queue), + dtx, + false, + ) + .await; + drop(previous); + stalled.await.unwrap(); + + assert_eq!(queue.count(), 1, "a resumed session must still receive it"); + } + #[tokio::test] async fn armed_will_is_claimed_exactly_once() { let router = MessageRouter::new(); diff --git a/crates/mqtt5/src/broker/storage/client_queue.rs b/crates/mqtt5/src/broker/storage/client_queue.rs index ca634b47..6155fba8 100644 --- a/crates/mqtt5/src/broker/storage/client_queue.rs +++ b/crates/mqtt5/src/broker/storage/client_queue.rs @@ -96,6 +96,7 @@ pub struct ClientQueue { count: AtomicUsize, handoffs: AtomicUsize, draining: AtomicBool, + epoch: AtomicU64, notify: Notify, handoff_done: Notify, seq: Arc, @@ -123,6 +124,7 @@ impl ClientQueue { count: AtomicUsize::new(0), handoffs: AtomicUsize::new(0), draining: AtomicBool::new(false), + epoch: AtomicU64::new(0), notify: Notify::new(), handoff_done: Notify::new(), seq, @@ -222,25 +224,52 @@ impl ClientQueue { self.seq.load(Ordering::Acquire) } + #[must_use] + pub fn epoch(&self) -> u64 { + self.epoch.load(Ordering::Acquire) + } + /// Appends a message. Drops the oldest entries first when a limit is exceeded. pub fn push(&self, message: QueuedMessage) -> PushOutcome { let body = Arc::new(message); - let (seq, dropped, evicted) = { + let appended = self.append(&mut self.inner.lock(), &body); + self.persist_append(appended, body) + } + + pub fn push_in_epoch(&self, message: QueuedMessage, epoch: u64) -> Option { + let body = Arc::new(message); + let appended = { let mut inner = self.inner.lock(); - let seq = self.seq.fetch_add(1, Ordering::AcqRel); - let entry = QueueEntry { - seq, - bytes: body.payload.len(), - expires_at: body.expires_at, - body: Some(Arc::clone(&body)), - path: None, - }; - inner.bytes += entry.bytes; - inner.entries.push_back(entry); - let evicted = self.enforce_limits(&mut inner); - self.count.store(inner.entries.len(), Ordering::Release); - (seq, evicted.len(), evicted) + if self.epoch() != epoch { + return None; + } + self.append(&mut inner, &body) + }; + Some(self.persist_append(appended, body)) + } + + fn append(&self, inner: &mut QueueInner, body: &Arc) -> (u64, Vec) { + let seq = self.seq.fetch_add(1, Ordering::AcqRel); + let entry = QueueEntry { + seq, + bytes: body.payload.len(), + expires_at: body.expires_at, + body: Some(Arc::clone(body)), + path: None, }; + inner.bytes += entry.bytes; + inner.entries.push_back(entry); + let evicted = self.enforce_limits(inner); + self.count.store(inner.entries.len(), Ordering::Release); + (seq, evicted) + } + + fn persist_append( + &self, + (seq, evicted): (u64, Vec), + body: Arc, + ) -> PushOutcome { + let dropped = evicted.len(); let mut self_evicted = false; for old in evicted { if old.seq == seq { @@ -383,6 +412,9 @@ impl ClientQueue { }); inner.bytes = inner.entries.iter().map(|entry| entry.bytes).sum(); self.count.store(inner.entries.len(), Ordering::Release); + if cutoff.is_none() { + self.epoch.fetch_add(1, Ordering::AcqRel); + } removed }; for seq in &removed { @@ -721,12 +753,29 @@ mod tests { queue.push(message("c", "before1")); queue.push(message("c", "before2")); let cutoff = queue.next_seq(); + let epoch = queue.epoch(); queue.push(message("c", "after")); assert_eq!(queue.clear(Some(cutoff)), 2); + assert_eq!(queue.epoch(), epoch); assert_eq!(queue.count(), 1); assert_eq!(queue.take(5).await[0].topic, "t/after"); } + #[tokio::test] + async fn push_in_epoch_refuses_a_message_routed_before_a_clear() { + let registry = registry(QueueLimits::default()); + let queue = registry.handle("c"); + let stale = queue.epoch(); + queue.push(message("c", "old")); + assert_eq!(queue.clear(None), 1); + assert!(queue.push_in_epoch(message("c", "stale"), stale).is_none()); + assert!(queue + .push_in_epoch(message("c", "fresh"), queue.epoch()) + .is_some()); + assert_eq!(queue.count(), 1); + assert_eq!(queue.take(5).await[0].topic, "t/fresh"); + } + #[tokio::test] async fn push_drops_oldest_when_the_message_limit_is_exceeded() { let registry = registry(QueueLimits { From a0dffa643c20bda45471f5877b90705de6e24043 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Fabr=C3=ADcio=20Bracht?= Date: Sun, 27 Sep 2026 22:09:41 -0300 Subject: [PATCH 2/2] persist queued messages before they become visible to delivery --- CHANGELOG.md | 1 + .../mqtt5/src/broker/storage/client_queue.rs | 166 +++++++++++------- 2 files changed, 101 insertions(+), 66 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 9cf1781a..72666f5e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed - **A clean-start reconnect no longer receives a message routed to the session it replaced.** A publish routed to a client just before that client reconnected with `clean_start=1` could land in the new session's queue after the queue had been cleared, delivering a message for a subscription the new session never made (#150). Each client queue now counts its full clears, and a message routed before the latest one is dropped. A resumed session still receives it. +- **The file backend no longer writes a queued message to disk after it was delivered or cleared.** A message delivered or cleared at the same moment it was queued could have its delete reach the storage writer before its write, leaving the file on disk; after a restart it was loaded again and redelivered. Writes are now handed to the storage writer before the entry becomes visible to delivery or clearing. ### Added diff --git a/crates/mqtt5/src/broker/storage/client_queue.rs b/crates/mqtt5/src/broker/storage/client_queue.rs index 6155fba8..d7d48a52 100644 --- a/crates/mqtt5/src/broker/storage/client_queue.rs +++ b/crates/mqtt5/src/broker/storage/client_queue.rs @@ -231,55 +231,44 @@ impl ClientQueue { /// Appends a message. Drops the oldest entries first when a limit is exceeded. pub fn push(&self, message: QueuedMessage) -> PushOutcome { - let body = Arc::new(message); - let appended = self.append(&mut self.inner.lock(), &body); - self.persist_append(appended, body) + let appended = self.append(&mut self.inner.lock(), Arc::new(message)); + self.release_evicted(appended) } pub fn push_in_epoch(&self, message: QueuedMessage, epoch: u64) -> Option { - let body = Arc::new(message); let appended = { let mut inner = self.inner.lock(); if self.epoch() != epoch { return None; } - self.append(&mut inner, &body) + self.append(&mut inner, Arc::new(message)) }; - Some(self.persist_append(appended, body)) + Some(self.release_evicted(appended)) } - fn append(&self, inner: &mut QueueInner, body: &Arc) -> (u64, Vec) { + fn append(&self, inner: &mut QueueInner, body: Arc) -> (u64, Vec) { let seq = self.seq.fetch_add(1, Ordering::AcqRel); let entry = QueueEntry { seq, bytes: body.payload.len(), expires_at: body.expires_at, - body: Some(Arc::clone(body)), + body: Some(Arc::clone(&body)), path: None, }; inner.bytes += entry.bytes; inner.entries.push_back(entry); let evicted = self.enforce_limits(inner); self.count.store(inner.entries.len(), Ordering::Release); + if !evicted.iter().any(|old| old.seq == seq) { + self.enqueue_write(seq, body); + } (seq, evicted) } - fn persist_append( - &self, - (seq, evicted): (u64, Vec), - body: Arc, - ) -> PushOutcome { + fn release_evicted(&self, (seq, evicted): (u64, Vec)) -> PushOutcome { let dropped = evicted.len(); - let mut self_evicted = false; - for old in evicted { - if old.seq == seq { - self_evicted = true; - } else { - self.enqueue_delete(old.seq); - } - } - if !self_evicted { - self.enqueue_write(seq, body); + for old in evicted.iter().filter(|old| old.seq != seq) { + self.enqueue_delete(old.seq); } if dropped > 0 { warn!( @@ -350,51 +339,46 @@ impl ClientQueue { return; } let bodies: Vec> = messages.into_iter().map(Arc::new).collect(); - let entries: Vec<(QueueEntry, Arc)> = { - let mut inner = self.inner.lock(); - let base = inner - .entries - .front() - .map_or_else(|| self.seq.load(Ordering::Acquire), |front| front.seq); - let len = bodies.len() as u64; - let first = base.checked_sub(len).unwrap_or_else(|| { - warn!( - client_id = %self.client_id, - base, - len, - "Sequence space exhausted below the queue head; re-queued order is best effort" - ); - 0 + let mut inner = self.inner.lock(); + let base = inner + .entries + .front() + .map_or_else(|| self.seq.load(Ordering::Acquire), |front| front.seq); + let len = bodies.len() as u64; + let first = base.checked_sub(len).unwrap_or_else(|| { + warn!( + client_id = %self.client_id, + base, + len, + "Sequence space exhausted below the queue head; re-queued order is best effort" + ); + 0 + }); + let mut entries = Vec::with_capacity(bodies.len()); + for (index, body) in bodies.iter().enumerate() { + entries.push(QueueEntry { + seq: first + index as u64, + bytes: body.payload.len(), + expires_at: body.expires_at, + body: Some(Arc::clone(body)), + path: None, }); - let mut entries = Vec::with_capacity(bodies.len()); - for (index, body) in bodies.iter().enumerate() { - entries.push(QueueEntry { - seq: first + index as u64, - bytes: body.payload.len(), - expires_at: body.expires_at, - body: Some(Arc::clone(body)), - path: None, - }); - } - for entry in entries.iter().rev() { - inner.bytes += entry.bytes; - inner.entries.push_front(entry.clone()); - } - let evicted = self.enforce_limits_dir(&mut inner, true); - self.count.store(inner.entries.len(), Ordering::Release); - let evicted_seqs: Vec = evicted.iter().map(|old| old.seq).collect(); - let batch_seqs: Vec = entries.iter().map(|entry| entry.seq).collect(); - for old in evicted_seqs.iter().filter(|seq| !batch_seqs.contains(seq)) { - self.enqueue_delete(*old); + } + for entry in entries.iter().rev() { + inner.bytes += entry.bytes; + inner.entries.push_front(entry.clone()); + } + let evicted = self.enforce_limits_dir(&mut inner, true); + self.count.store(inner.entries.len(), Ordering::Release); + let evicted_seqs: Vec = evicted.iter().map(|old| old.seq).collect(); + let batch_seqs: Vec = entries.iter().map(|entry| entry.seq).collect(); + for old in evicted_seqs.iter().filter(|seq| !batch_seqs.contains(seq)) { + self.enqueue_delete(*old); + } + for (entry, body) in entries.iter().zip(bodies) { + if !evicted_seqs.contains(&entry.seq) { + self.enqueue_write(entry.seq, body); } - entries - .into_iter() - .zip(bodies) - .filter(|(entry, _)| !evicted_seqs.contains(&entry.seq)) - .collect::>() - }; - for (entry, body) in entries { - self.enqueue_write(entry.seq, body); } } @@ -746,6 +730,56 @@ mod tests { ); } + #[test] + fn writer_never_sees_a_write_after_the_delete_of_the_same_entry() { + let (writer, mut ops) = mpsc::unbounded_channel(); + let registry = QueueRegistry::new(QueueLimits::default(), Some(writer)); + let queue = registry.handle("c"); + let pushers: Vec<_> = (0..4) + .map(|_| { + let queue = Arc::clone(&queue); + std::thread::spawn(move || { + for _ in 0..50_000 { + queue.push(message("c", "m")); + queue.requeue_front(vec![message("c", "r")]); + } + }) + }) + .collect(); + let clearers: Vec<_> = (0..4) + .map(|_| { + let queue = Arc::clone(&queue); + std::thread::spawn(move || { + for _ in 0..50_000 { + queue.clear(None); + } + }) + }) + .collect(); + for handle in pushers.into_iter().chain(clearers) { + handle.join().unwrap(); + } + queue.clear(None); + + let mut last_op: std::collections::HashMap = std::collections::HashMap::new(); + while let Ok(op) = ops.try_recv() { + match op { + QueueOp::Write { seq, .. } => { + last_op.insert(seq, true); + } + QueueOp::Delete { seq, .. } => { + last_op.insert(seq, false); + } + _ => {} + } + } + let resurrected = last_op.values().filter(|written| **written).count(); + assert_eq!( + resurrected, 0, + "every cleared entry must end deleted, not rewritten" + ); + } + #[tokio::test] async fn clear_with_cutoff_keeps_messages_queued_after_the_cutoff() { let registry = registry(QueueLimits::default());