diff --git a/CHANGELOG.md b/CHANGELOG.md
index 92bb8d88..72666f5e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -5,6 +5,17 @@ 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.
+- **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
+
+- `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<u64>,
         lanes: DeliveryLanes,
         queue: QueueHandle,
+        epoch: u64,
     },
     Behind {
         client_id: String,
         message: PublishPacket,
         target_flow: Option<u64>,
         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<u64>,
     ) {
         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<PublishPacket> {
@@ -2640,6 +2661,93 @@ mod tests {
             .generation
     }
 
+    async fn stall_publish_behind_full_lane(
+        router: &Arc<MessageRouter>,
+        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..d7d48a52 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<AtomicU64>,
@@ -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,36 +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(), Arc::new(message));
+        self.release_evicted(appended)
+    }
+
+    pub fn push_in_epoch(&self, message: QueuedMessage, epoch: u64) -> Option<PushOutcome> {
+        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)
-        };
-        let mut self_evicted = false;
-        for old in evicted {
-            if old.seq == seq {
-                self_evicted = true;
-            } else {
-                self.enqueue_delete(old.seq);
+            if self.epoch() != epoch {
+                return None;
             }
-        }
-        if !self_evicted {
+            self.append(&mut inner, Arc::new(message))
+        };
+        Some(self.release_evicted(appended))
+    }
+
+    fn append(&self, inner: &mut QueueInner, body: Arc<QueuedMessage>) -> (u64, Vec<QueueEntry>) {
+        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);
+        if !evicted.iter().any(|old| old.seq == seq) {
             self.enqueue_write(seq, body);
         }
+        (seq, evicted)
+    }
+
+    fn release_evicted(&self, (seq, evicted): (u64, Vec<QueueEntry>)) -> PushOutcome {
+        let dropped = evicted.len();
+        for old in evicted.iter().filter(|old| old.seq != seq) {
+            self.enqueue_delete(old.seq);
+        }
         if dropped > 0 {
             warn!(
                 client_id = %self.client_id,
@@ -321,51 +339,46 @@ impl ClientQueue {
             return;
         }
         let bodies: Vec<Arc<QueuedMessage>> = messages.into_iter().map(Arc::new).collect();
-        let entries: Vec<(QueueEntry, Arc<QueuedMessage>)> = {
-            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<u64> = evicted.iter().map(|old| old.seq).collect();
-            let batch_seqs: Vec<u64> = 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<u64> = evicted.iter().map(|old| old.seq).collect();
+        let batch_seqs: Vec<u64> = 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::<Vec<_>>()
-        };
-        for (entry, body) in entries {
-            self.enqueue_write(entry.seq, body);
         }
     }
 
@@ -383,6 +396,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 {
@@ -714,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<u64, bool> = 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());
@@ -721,12 +787,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 {