diff --git a/CHANGELOG.md b/CHANGELOG.md index 7f33174a..9ff6f92d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,19 @@ archived by series under [docs/changelog/](docs/changelog/); see the ### Fixed +- **Confirmation probes no longer crowd real messages out of the outbox.** A + session the peer never confirms is probed every five seconds, and each probe + used to enter the outbox with its own retry ladder on top of the unanswered + ones before it. Only a relay verdict backed that off, so a mesh-only link + never did. An account with ten such contacts stacked about two probes a + second, and once the outbox hit its 500-entry cap, eviction failed the user's + own messages with `Outbox capacity exceeded`. Seen on a device test between + an Android and an iPhone, offline over BLE. Each new probe now supersedes + the last, quietly and without counting against the carrier, whether the + periodic scan or the Welcome fast path sent it. Probes are no longer written + to storage, and those an older build wrote are dropped at restore. A peer + holds one probe in the outbox and is still probed on the same cadence. + - **A message carried through a dense cluster no longer dies inside it** ([#510](https://github.com/Offline-Protocol/offline-protocol-sdk/issues/510)). A forwarder that received a second copy of a frame while its own forward @@ -488,6 +501,7 @@ archived by series under [docs/changelog/](docs/changelog/); see the unchanged. TypeScript's `SecurityWarningCode` union gains the code, so an exhaustive `switch` over it needs a new arm. + - **An old message no longer comes back as a push notification, again and again.** A direct message the relay had pushed stayed in the outbox well past a week, because each probe re-send refreshed its lifetime and so did each diff --git a/crates/offline-protocol/src/protocol/mod.rs b/crates/offline-protocol/src/protocol/mod.rs index 41d27679..07dd6e1d 100644 --- a/crates/offline-protocol/src/protocol/mod.rs +++ b/crates/offline-protocol/src/protocol/mod.rs @@ -709,6 +709,25 @@ pub struct OfflineProtocol { /// flap the connection. Reset on a reachability edge (`on_peer_presence`). confirmation_probe_unreachable_parks: HashMap, + /// The confirmation probe each pending peer last sent. At most one probe + /// per peer lives in the outbox: a new one supersedes the last. Stacking a + /// fresh probe (new id, full retry ladder) every + /// `CONFIRMATION_PROBE_INTERVAL_SECS` on top of the unanswered ones filled + /// the outbox whenever no relay verdict (`unreachable` or + /// `relay_pushed`) arrived to back the schedule off, as on a mesh-only + /// carrier. Capacity eviction then failed real messages. + confirmation_probe_outstanding: HashMap, + + /// The probe each pending peer's current one superseded. A relay verdict + /// finds its peer through the outbox entry, which the supersede removed, + /// so a verdict slower than `CONFIRMATION_PROBE_INTERVAL_SECS` would + /// otherwise be dropped and never back the schedule off, leaving the + /// peer probed every 5s against a relay that has said it is not there. + /// One id per peer, so this covers a verdict up to two intervals late; + /// one later than that is dropped, and the next probe's verdict backs + /// the schedule off instead. + confirmation_probe_superseded: HashMap, + /// Token bucket that caps how fast the RETRY/PROBE path /// (`process_retry_queue`) re-sends, so a large backlog of resends to /// unreachable peers cannot burst past the relay's per-connection rate limit @@ -1176,6 +1195,8 @@ impl OfflineProtocol { confirmation_retry_due_at: HashMap::new(), confirmation_probe_due_at: HashMap::new(), confirmation_probe_unreachable_parks: HashMap::new(), + confirmation_probe_outstanding: HashMap::new(), + confirmation_probe_superseded: HashMap::new(), retry_drain_tokens: 0.0, retry_drain_last: None, rekey_due_at: HashMap::new(), diff --git a/crates/offline-protocol/src/protocol/send.rs b/crates/offline-protocol/src/protocol/send.rs index 9ae73ef1..75c06e39 100644 --- a/crates/offline-protocol/src/protocol/send.rs +++ b/crates/offline-protocol/src/protocol/send.rs @@ -3277,12 +3277,28 @@ impl OfflineProtocol { // nothing, and keeping it would only grow the map. self.custody_receipts.remove(message_id); if let Some(entry) = self.outbox.remove(message_id) { - self.clear_outbox_entry_from_storage(message_id); + if !Self::is_confirmation_probe(&entry.message) { + self.clear_outbox_entry_from_storage(message_id); + } return Some(entry); } self.media_outbox.remove(message_id) } + /// Takes `message_id` out of the three places that can send it again: + /// the retry queue, the pending-ACK tracker and the outbox. One without + /// the others resurrects it (see [`Self::retire_undeliverable_message`]). + /// No failure is recorded against the carrier: that is the give-up + /// path's, and a message withdrawn for being superseded did not fail. + pub(super) fn forget_outbound_message( + &mut self, + message_id: &MessageId, + ) -> Option { + self.retry_queue.remove(&message_id.as_str()); + self.ack_manager.remove_ack(message_id); + self.remove_outbox_entry(message_id) + } + /// Takes a message this device has given up on out of everything that /// could send it again, returning its outbox entry if it still had one. /// @@ -3307,10 +3323,8 @@ impl OfflineProtocol { message_id: &MessageId, media_reason: &str, ) -> Option { - self.retry_queue.remove(&message_id.as_str()); - self.ack_manager.remove_ack(message_id); self.handle_outbound_media_chunk_failed(message_id, media_reason); - let entry = self.remove_outbox_entry(message_id); + let entry = self.forget_outbound_message(message_id); if let Some(transport) = entry.as_ref().and_then(|entry| entry.last_transport) { self.transport_manager.record_delivery_failure(transport); } @@ -3534,6 +3548,17 @@ impl OfflineProtocol { message.content_type == ContentType::FileChunk } + /// A session confirmation probe lives in memory only. It carries no user + /// data, and a restarted process does not know which probe was the last, + /// so a restored one could never be superseded; the next scan asks the + /// same question. Persisting it cost a secure-storage write per probe and + /// a delete per supersede, under the protocol lock. + pub(super) fn is_confirmation_probe(message: &Message) -> bool { + message + .content + .starts_with(internal_prefixes::SESSION_CONFIRM_PROBE) + } + // ======================================================================== // DELIVERY TRACKING // ======================================================================== @@ -4111,10 +4136,12 @@ impl OfflineProtocol { None => match self.media_outbox.get(&parsed_id) { Some(entry) => (entry, true), None => { - debug!( - message_id = %message_id, - "Unreachable verdict for message without outbox entry, dropping" - ); + if !self.note_superseded_probe_verdict(&parsed_id, carrier) { + debug!( + message_id = %message_id, + "Unreachable verdict for message without outbox entry, dropping" + ); + } return; } }, @@ -4270,6 +4297,7 @@ impl OfflineProtocol { return; }; if !self.is_parkable_plain_dm(&parsed_id) { + self.note_superseded_probe_verdict(&parsed_id, carrier); return; } let Some(entry) = self.outbox.get_mut(&parsed_id) else { diff --git a/crates/offline-protocol/src/protocol/session.rs b/crates/offline-protocol/src/protocol/session.rs index 6537286c..4b19ccd9 100644 --- a/crates/offline-protocol/src/protocol/session.rs +++ b/crates/offline-protocol/src/protocol/session.rs @@ -16,7 +16,7 @@ use crate::mls_observability::MlsOperationContext; use crate::protocol::reachability::{Claim, FactSource}; use crate::{Error, EstablishmentState, Event, Result, SessionStateError}; use chrono::{Duration as ChronoDuration, Utc}; -use offline_protocol_core::{Message, MessagePriority}; +use offline_protocol_core::{Message, MessageId, MessagePriority}; use offline_protocol_mls::{MlsManager, WelcomeMessage}; use offline_protocol_transport::TransportType; use std::collections::HashSet; @@ -449,6 +449,45 @@ impl OfflineProtocol { self.confirmation_retry_due_at.remove(peer_id); self.confirmation_probe_due_at.remove(peer_id); self.confirmation_probe_unreachable_parks.remove(peer_id); + // Withdrawn with the tracking: once the map forgets it, no later + // probe can supersede it, and it would keep its full retry ladder. + if let Some(probe) = self.confirmation_probe_outstanding.remove(peer_id) { + self.forget_outbound_message(&probe); + } + self.confirmation_probe_superseded.remove(peer_id); + } + + /// Applies a relay verdict on a probe that has since been superseded + /// (`confirmation_probe_superseded`) the way the verdict handlers apply + /// one on a queued probe: the reachability fact, then the probe back-off. + /// Consumed, so one probe's verdict escalates once, as it did while the + /// probe was still queued. A no-op for any other id; returns whether + /// the verdict was applied. + pub(super) fn note_superseded_probe_verdict( + &mut self, + message_id: &MessageId, + carrier: Option, + ) -> bool { + let Some(peer) = self + .confirmation_probe_superseded + .iter() + .find(|(_, probe)| *probe == message_id) + .map(|(peer, _)| peer.clone()) + else { + return false; + }; + self.confirmation_probe_superseded.remove(&peer); + if let Some(carrier) = carrier { + self.reachability.record( + &peer, + carrier, + Claim::Unreachable, + FactSource::GatewayVerdict, + Instant::now(), + ); + } + self.note_confirmation_probe_unreachable(&peer); + true } /// Fold a relay "recipient unreachable" verdict into the session-confirmation @@ -457,7 +496,7 @@ impl OfflineProtocol { /// [`Self::apply_recipient_unreachable_failure`]). Without this, a peer that /// established an MLS session then vanished before confirming is re-probed /// every `CONFIRMATION_PROBE_INTERVAL_SECS` (5s) indefinitely, because - /// `kick_pending_session_reconciliation` re-arms it every scan and + /// `send_session_confirmation_probe` re-arms it on every probe and /// `on_transport_send_failed` otherwise never consults this scheduler. A /// handful of such peers pushes aggregate relay traffic past the /// per-connection rate limit, which disconnects the socket on a loop. @@ -512,12 +551,44 @@ impl OfflineProtocol { } pub(super) fn send_session_confirmation_probe(&mut self, peer_id: &str, source_event: &str) { + // Superseded, not failed: the new probe asks the same question, so + // the old one's retry ladder would only stack onto it. Here rather + // than in the scan because the Welcome fast path sends one too, and + // a probe it replaced would keep its full ladder. Silent, and no + // delivery failure is recorded against the carrier. Before the send, + // not after: if the send errors the peer has no probe in flight + // until the next scan, but a new probe can never evict a message at + // outbox capacity to make room beside the one it replaces. The + // superseded id is kept so a relay verdict that arrives for it late + // still backs the schedule off (`note_superseded_probe_verdict`). + if let Some(previous) = self.confirmation_probe_outstanding.remove(peer_id) { + self.forget_outbound_message(&previous); + self.confirmation_probe_superseded + .insert(peer_id.to_string(), previous); + } + // The sender owns the cadence, so a fast-path probe is not superseded + // by the next scan before its ACK can arrive, and a relay verdict for + // it can back the schedule off (`note_confirmation_probe_unreachable` + // only escalates a peer that has a due time). Stamped before the send + // so a verdict delivered during it is not overwritten. + // + // A later due time is kept: the scan only sends once the due time has + // passed, but the fast path can fire while the schedule is backed off, + // and resetting it would undo the ladder a relay verdict built. + let next = Utc::now() + ChronoDuration::seconds(CONFIRMATION_PROBE_INTERVAL_SECS); + let due_at = self + .confirmation_probe_due_at + .entry(peer_id.to_string()) + .or_insert(next); + *due_at = (*due_at).max(next); match self.send_internal_message( peer_id, internal_prefixes::SESSION_CONFIRM_PROBE.to_string(), MessagePriority::High, ) { - Ok(_) => { + Ok(message_id) => { + self.confirmation_probe_outstanding + .insert(peer_id.to_string(), message_id); info!( event = "session_confirmation_probe_sent", session_or_group_id = %peer_id, @@ -526,6 +597,22 @@ impl OfflineProtocol { ); } Err(err) => { + // The send can fail after the probe entered the outbox (a + // full ACK tracker refuses it), and then no id comes back to + // track. Any probe still queued for this peer is such an + // orphan, since the tracked one was withdrawn above. + let orphans: Vec<_> = self + .outbox + .values() + .filter(|entry| { + entry.message.recipient.as_str() == peer_id + && Self::is_confirmation_probe(&entry.message) + }) + .map(|entry| entry.message.id.clone()) + .collect(); + for orphan in orphans { + self.forget_outbound_message(&orphan); + } warn!( event = "session_confirmation_probe_failed", session_or_group_id = %peer_id, @@ -596,6 +683,22 @@ impl OfflineProtocol { .retain(|peer, _| pending_set.contains(peer)); self.confirmation_probe_unreachable_parks .retain(|peer, _| pending_set.contains(peer)); + self.confirmation_probe_superseded + .retain(|peer, _| pending_set.contains(peer)); + // Not a plain `retain`: a probe whose peer left the pending set + // without passing through `clear_confirmation_recovery_tracking` + // would be forgotten while still queued. + let settled: Vec = self + .confirmation_probe_outstanding + .keys() + .filter(|peer| !pending_set.contains(*peer)) + .cloned() + .collect(); + for peer in settled { + if let Some(probe) = self.confirmation_probe_outstanding.remove(&peer) { + self.forget_outbound_message(&probe); + } + } for peer_id in pending_peers { let due_at = self @@ -606,12 +709,7 @@ impl OfflineProtocol { if due_at > now { continue; } - self.send_session_confirmation_probe(&peer_id, source_event); - self.confirmation_probe_due_at.insert( - peer_id, - now + ChronoDuration::seconds(CONFIRMATION_PROBE_INTERVAL_SECS), - ); } } diff --git a/crates/offline-protocol/src/protocol/storage.rs b/crates/offline-protocol/src/protocol/storage.rs index 4cc9f112..8509f5ae 100644 --- a/crates/offline-protocol/src/protocol/storage.rs +++ b/crates/offline-protocol/src/protocol/storage.rs @@ -882,6 +882,16 @@ const _: () = assert!( underflows to zero and it can never prune at all" ); +// The outbox walk charges a legacy confirmation probe's delete to its pool +// (`restore_outbox`). A build that stacked probes could leave a full outbox +// of them, and if the pool is smaller than that the walk breaks on probes and +// leaves real messages unrestored until a later launch. +const _: () = assert!( + MAX_RESTORE_PRUNE_DELETES >= MAX_OUTBOX_ENTRIES, + "an outbox full of legacy confirmation probes must clear in one launch, \ + or the walk breaks on them and defers restoring real messages" +); + /// A pool of durable restore-path deletes, drawn on by one or more walks. /// /// The unit the [`MAX_RESTORE_PRUNE_DELETES`] bound is actually about. A pool @@ -3800,7 +3810,9 @@ impl OfflineProtocol { let Some(storage) = &self.protocol_state_storage else { return; }; - if Self::is_media_outbox_message(&entry.message) { + if Self::is_media_outbox_message(&entry.message) + || Self::is_confirmation_probe(&entry.message) + { return; } match serde_json::to_vec(entry) { @@ -3941,6 +3953,7 @@ impl OfflineProtocol { // `PruneAllowance::pool`. let mut budget = allowance.counting(); let mut prune_bound_reached = false; + let mut legacy_probes_dropped = 0usize; for message_id in message_ids.into_iter().take(OUTBOX_RESTORE_KEY_CAP) { if budget.is_spent() { prune_bound_reached = true; @@ -3997,8 +4010,29 @@ impl OfflineProtocol { continue; } + // Probes are no longer persisted (`is_confirmation_probe`); drop + // any an older build wrote. Charged to the pool like any delete: + // the first launch after upgrading from the build that stacked + // them can spend most of it here, deferring the expiry and + // capacity prunes below to the next launch. That is the bound + // working, not a reason to make this delete unbudgeted. + if Self::is_confirmation_probe(&entry.message) { + budget.claim(); + self.delete_outbox_key(&message_id); + legacy_probes_dropped += 1; + continue; + } + restored.push(entry); } + if legacy_probes_dropped > 0 { + info!( + event = "outbox_entry_dropped", + repair_action = "legacy_confirmation_probe", + count = legacy_probes_dropped, + "outbox_entry_dropped" + ); + } // Drop entries past the absolute lifetime cap before anything else: // the carrier-relative refresh below would otherwise re-grant them a diff --git a/crates/offline-protocol/src/protocol/tests/mod.rs b/crates/offline-protocol/src/protocol/tests/mod.rs index 3b8436be..5ca4fcf4 100644 --- a/crates/offline-protocol/src/protocol/tests/mod.rs +++ b/crates/offline-protocol/src/protocol/tests/mod.rs @@ -40904,6 +40904,351 @@ fn test_confirmation_probe_unreachable_escalation() { ); } +/// Alice holds a session bob never confirms, running over a BLE mock. The +/// fixture for the confirmation-probe supersede tests below. +fn unconfirmed_probe_target() -> (OfflineProtocol, MockTransport) { + let mut alice_config = create_test_config_for_user("alice"); + alice_config.encryption.enabled = true; + let mut alice = OfflineProtocol::new(alice_config).unwrap(); + let mut bob = OfflineProtocol::new(create_test_config_for_user("bob")).unwrap(); + alice + .initialize_mls_for_test(Arc::new(InMemoryStorage::new())) + .unwrap(); + bob.initialize_mls_for_test(Arc::new(InMemoryStorage::new())) + .unwrap(); + + let bob_key_package = { + let manager = bob.mls_manager.as_ref().unwrap().read().unwrap(); + manager.get_or_create_key_package().unwrap() + }; + { + let manager = alice.mls_manager.as_ref().unwrap().read().unwrap(); + manager + .import_key_package(&id("bob"), &bob_key_package.key_package_data) + .unwrap(); + manager.create_session(&id("bob")).unwrap(); + } + alice + .ensure_session_state_entry(&id("bob"), "test_setup") + .unwrap(); + + let transport = MockTransport::new(TransportType::BLE); + transport.start().unwrap(); + let handle = transport.clone(); + alice + .transport_manager_mut() + .add_transport(TransportType::BLE, Box::new(transport)); + alice.start().unwrap(); + handle.clear_sent_messages(); + (alice, handle) +} + +fn sent_confirmation_probes(handle: &MockTransport) -> Vec { + handle + .sent_messages() + .into_iter() + .filter(|m| { + m.content + .starts_with(internal_prefixes::SESSION_CONFIRM_PROBE) + }) + .map(|m| m.id) + .collect() +} + +/// A peer whose session never confirms, and that no `unreachable` verdict +/// backs off (a relay that pushes to offline users, or a mesh-only link), +/// holds one probe in the outbox, however many intervals pass. Each due scan +/// still probes, so a lost probe is replaced on the next one, but the new +/// probe supersedes the last instead of stacking a fresh retry ladder on it: +/// stacking filled the outbox at one probe per such peer per 5s, and capacity +/// eviction failed real messages. +#[test] +fn an_unanswered_confirmation_probe_supersedes_the_last_one() { + let (mut alice, handle) = unconfirmed_probe_target(); + let bob = id("bob").to_string(); + + let events: Arc>> = Arc::new(Mutex::new(Vec::new())); + let events_handle = Arc::clone(&events); + alice.on_event(move |event| events_handle.lock().unwrap().push(event)); + + // The scan and the Welcome fast path (`on_transport_send_confirmed`) + // both probe, interleaved: either may supersede the other's probe. + for round in 0..6 { + if round % 2 == 0 { + alice + .confirmation_probe_due_at + .insert(bob.clone(), Utc::now() - ChronoDuration::seconds(1)); + alice.kick_pending_session_reconciliation("test"); + } else { + alice.send_session_confirmation_probe(&bob, "transport_confirmed"); + } + } + let sent = sent_confirmation_probes(&handle); + assert_eq!(sent.len(), 6, "every due scan and fast path still probes"); + + let outstanding: Vec<_> = alice.outbox.keys().filter(|id| sent.contains(id)).collect(); + assert_eq!( + outstanding, + vec![sent.last().unwrap()], + "only the latest probe waits in the outbox; each superseded the last" + ); + for superseded in &sent[..sent.len() - 1] { + assert!( + !alice.retry_queue.contains(&superseded.as_str()), + "a superseded probe keeps no retry ladder" + ); + assert!( + !alice.ack_manager.is_waiting_for_ack(superseded), + "a superseded probe's ACK timeout would re-queue it" + ); + } + assert_eq!( + alice + .transport_manager + .observed_failure_count(TransportType::BLE), + 0, + "superseding is not a delivery failure against the carrier" + ); + assert!( + !events + .lock() + .unwrap() + .iter() + .any(|e| matches!(e, Event::MessageFailed { .. })), + "a superseded probe is withdrawn silently" + ); + + // The sender stamps the cadence, so the next scan does not supersede a + // fast-path probe before its ACK can arrive. + assert!(alice.confirmation_probe_due_at[&bob] > Utc::now()); + + // A probe whose send failed waits in the retry queue; the next probe + // takes it out of there too. + handle.set_fail_next_sends(1); + alice.send_session_confirmation_probe(&bob, "test"); + let deferred = alice.confirmation_probe_outstanding[&bob].clone(); + assert!(alice.retry_queue.contains(&deferred.as_str())); + alice.send_session_confirmation_probe(&bob, "test"); + assert!(!alice.retry_queue.contains(&deferred.as_str())); + assert!(!alice.outbox.contains_key(&deferred)); +} + +/// A relay verdict slower than the probe interval names a probe already +/// superseded. It still records the fact and backs the schedule off, once per +/// probe, for either verdict. One id per peer is kept, so a verdict two +/// intervals late finds nothing and changes nothing. +#[test] +fn a_late_verdict_on_a_superseded_probe_still_backs_off() { + let (mut alice, _handle) = unconfirmed_probe_target(); + let bob = id("bob").to_string(); + + // Two intervals late: the oldest of three probes is no longer remembered. + alice.send_session_confirmation_probe(&bob, "test"); + let too_late = alice.confirmation_probe_outstanding[&bob].clone(); + alice.send_session_confirmation_probe(&bob, "test"); + alice.send_session_confirmation_probe(&bob, "test"); + assert!(!alice.note_superseded_probe_verdict(&too_late, Some(TransportType::Internet))); + assert_eq!( + alice + .reachability + .claim_for(&bob, TransportType::Internet, std::time::Instant::now()), + None, + "a verdict past the remembered probe records nothing" + ); + assert!(!alice + .confirmation_probe_unreachable_parks + .contains_key(&bob)); + + for verdict in ["recipient_unreachable: User is offline", "relay_pushed"] { + alice.send_session_confirmation_probe(&bob, "test"); + let late = alice.confirmation_probe_superseded[&bob].clone(); + let parks = alice + .confirmation_probe_unreachable_parks + .get(&bob) + .copied() + .unwrap_or(0); + for _ in 0..2 { + alice + .on_transport_send_failed_via( + &late.as_str(), + Some(verdict.to_string()), + Some(TransportType::Internet), + ) + .unwrap(); + } + assert_eq!( + alice + .reachability + .claim_for(&bob, TransportType::Internet, std::time::Instant::now()), + Some(crate::protocol::reachability::Claim::Unreachable), + "a late {verdict} verdict is recorded as a fact too" + ); + assert_eq!( + alice + .confirmation_probe_unreachable_parks + .get(&bob) + .copied(), + Some(parks + 1), + "a late {verdict} verdict escalates once" + ); + assert!( + alice.confirmation_probe_due_at[&bob] + > Utc::now() + ChronoDuration::seconds(CONFIRMATION_PROBE_INTERVAL_SECS) + ); + // A fast-path probe while backed off keeps the back-off. + let backed_off = alice.confirmation_probe_due_at[&bob]; + alice.send_session_confirmation_probe(&bob, "transport_confirmed"); + assert_eq!(alice.confirmation_probe_due_at[&bob], backed_off); + } +} + +/// A probe the tracking no longer names could never be superseded and would +/// keep its full retry ladder, so it is withdrawn whenever the tracking lets +/// go of it: the session confirms, the session is deleted under the engine, +/// or the send fails after the probe was queued. +#[test] +fn a_confirmation_probe_is_withdrawn_when_nothing_could_supersede_it() { + let (mut alice, handle) = unconfirmed_probe_target(); + let bob = id("bob").to_string(); + + // Confirming the session withdraws the last probe. + alice.send_session_confirmation_probe(&bob, "test"); + let last = alice.confirmation_probe_outstanding[&bob].clone(); + alice.clear_confirmation_recovery_tracking(&bob); + assert!(!alice.outbox.contains_key(&last)); + assert!(!alice.retry_queue.contains(&last.as_str())); + assert!(!alice.ack_manager.is_waiting_for_ack(&last)); + + // A peer that leaves the pending set without that call (its session + // deleted under the engine) has its probe withdrawn by the next scan, + // and the superseded id it kept for a late verdict is pruned too. + alice.send_session_confirmation_probe(&bob, "test"); + alice.send_session_confirmation_probe(&bob, "test"); + assert!(alice.confirmation_probe_superseded.contains_key(&bob)); + let orphan = sent_confirmation_probes(&handle).last().unwrap().clone(); + assert!(alice.outbox.contains_key(&orphan)); + { + let manager = alice.mls_manager.as_ref().unwrap().read().unwrap(); + manager.delete_session(&id("bob")).unwrap(); + } + alice.kick_pending_session_reconciliation("test"); + assert!(!alice.outbox.contains_key(&orphan)); + assert!(!alice.retry_queue.contains(&orphan.as_str())); + assert!(!alice.ack_manager.is_waiting_for_ack(&orphan)); + assert!(!alice.confirmation_probe_outstanding.contains_key(&bob)); + assert!(!alice.confirmation_probe_superseded.contains_key(&bob)); + + // A full ACK tracker fails the send after the probe entered the outbox, + // so no id comes back to supersede next time. The failed send clears it. + for _ in 0..100_000 { + if alice + .ack_manager + .register_pending_ack_with_priority(MessageId::new(), None, MessagePriority::High) + .is_err() + { + break; + } + } + alice.send_session_confirmation_probe(&bob, "test"); + alice.send_session_confirmation_probe(&bob, "test"); + assert_eq!( + alice + .outbox + .values() + .filter(|e| OfflineProtocol::is_confirmation_probe(&e.message)) + .count(), + 0, + "a probe whose send failed is not left queued" + ); +} + +/// A probe persisted before a restart is not in +/// `confirmation_probe_outstanding`, so no later probe could supersede it. +/// Restore drops it: it carries no user data, and the next scan asks again. +#[test] +fn a_persisted_confirmation_probe_is_not_restored() { + let storage = Arc::new(InMemoryStorage::new()); + let entry = |message: Message| OutboxEntry { + message, + attempt_count: 1, + first_sent_at: chrono::Utc::now(), + last_sent_at: chrono::Utc::now(), + last_transport: None, + reseal: None, + relay_pushed: false, + }; + let probe = test_message("bob", internal_prefixes::SESSION_CONFIRM_PROBE); + let real = test_message("bob", "hello"); + let real_id = real.id.clone(); + store_outbox_entry(&storage, &entry(probe)); + store_outbox_entry(&storage, &entry(real)); + + let mut protocol = OfflineProtocol::new(create_test_config()).unwrap(); + protocol + .enable_message_persistence_for_test(storage.clone()) + .unwrap(); + + let restored: Vec<_> = protocol.outbox_messages().map(|m| m.id.clone()).collect(); + assert_eq!(restored, vec![real_id.clone()]); + assert_eq!( + storage.list_keys(storage_keys::OUTBOX).unwrap(), + vec![real_id.as_str().to_string()], + "the probe's record is deleted, not left to be restored next launch" + ); + + // Nor is a probe written in the first place. + let probe = test_message("bob", internal_prefixes::SESSION_CONFIRM_PROBE); + protocol.persist_outbox_entry(&entry(probe)); + assert_eq!(storage.list_keys(storage_keys::OUTBOX).unwrap().len(), 1); +} + +/// The first launch after upgrading from the build that stacked probes can +/// find an outbox full of them. Their deletes are charged to the walk's pool, +/// so the pool has to cover a full outbox, or the walk breaks on probes and a +/// real message behind them waits for a later launch to be restored. +#[test] +fn an_outbox_full_of_legacy_probes_clears_in_one_launch() { + let storage = Arc::new(InMemoryStorage::new()); + let entry = |message: Message| OutboxEntry { + message, + attempt_count: 1, + first_sent_at: chrono::Utc::now(), + last_sent_at: chrono::Utc::now(), + last_transport: None, + reseal: None, + relay_pushed: false, + }; + for _ in 0..crate::constants::MAX_OUTBOX_ENTRIES { + store_outbox_entry( + &storage, + &entry(test_message( + "bob", + internal_prefixes::SESSION_CONFIRM_PROBE, + )), + ); + } + let real = test_message("bob", "hello"); + let real_id = real.id.clone(); + store_outbox_entry(&storage, &entry(real)); + + let mut protocol = OfflineProtocol::new(create_test_config()).unwrap(); + protocol + .enable_message_persistence_for_test(storage.clone()) + .unwrap(); + + let restored: Vec<_> = protocol.outbox_messages().map(|m| m.id.clone()).collect(); + assert_eq!( + restored, + vec![real_id.clone()], + "the real message is restored" + ); + assert_eq!( + storage.list_keys(storage_keys::OUTBOX).unwrap(), + vec![real_id.as_str().to_string()], + "every legacy probe is deleted in the same launch" + ); +} + /// Pins component (D) of the flap fix: the opt-in edge-driven parking gate. /// With `edge_driven_unreachable_dm` on, a durably-unreachable DM keeps being /// timed-probed up to `DM_UNREACHABLE_PROBE_LIMIT` times, then stops (rests in diff --git a/crates/offline-protocol/src/transport_manager.rs b/crates/offline-protocol/src/transport_manager.rs index 054fb1a1..a64fc6c4 100644 --- a/crates/offline-protocol/src/transport_manager.rs +++ b/crates/offline-protocol/src/transport_manager.rs @@ -1506,6 +1506,13 @@ impl TransportManager { stats.record_failure(); } + #[cfg(test)] + pub(crate) fn observed_failure_count(&self, transport_type: TransportType) -> u32 { + self.observations + .get(&transport_type) + .map_or(0, |stats| stats.failure_count) + } + /// Updates the DORS selector configuration at runtime, preserving /// accumulated state (transport history, retry counts, signal tracking). pub fn update_selector_config(&mut self, config: DorsConfig) { diff --git a/docs/state-machines/outbox-and-retries.md b/docs/state-machines/outbox-and-retries.md index 417e6060..5f685266 100644 --- a/docs/state-machines/outbox-and-retries.md +++ b/docs/state-machines/outbox-and-retries.md @@ -92,7 +92,12 @@ doing. the outbox together.** Capacity eviction, lifetime expiry, retry exhaustion and an acknowledgement timeout that finds no outbox entry all go through one function (`retire_undeliverable_message`) for that reason, and the event each -one owes the application is emitted by the path itself. +one owes the application is emitted by the path itself. A session confirmation +probe leaves through the same three-way teardown (`forget_outbound_message`) +when it is superseded by a newer one or withdrawn (its session confirmed or torn +down, or its send failed after it was queued), without the delivery failure a +give-up records against the carrier: it did not fail, it was replaced or is no +longer needed. Removing the outbox entry alone is not enough, because the retry and flush paths re-create a missing entry before they resend, with fresh timestamps. An id left diff --git a/docs/state-machines/session-lifecycle.md b/docs/state-machines/session-lifecycle.md index 7847c659..f750e803 100644 --- a/docs/state-machines/session-lifecycle.md +++ b/docs/state-machines/session-lifecycle.md @@ -93,6 +93,37 @@ re-enters the desync path. Repeating the reset flag would tear down a replacement session whose Welcome is in flight, so it belongs with the split-session work rather than here. +### An unconfirmed session holds one probe + +An unconfirmed session is probed on a fixed cadence, and **a peer holds at most +one probe in the outbox**: each new probe supersedes the last rather than +stacking its own retry ladder on top. The cadence only backs off on a relay +verdict (`unreachable` or `relay_pushed`), which a mesh-only carrier never +produces, so stacked probes once filled the outbox and capacity eviction failed +real messages. + +The supersede lives in the one function that sends a probe, because two paths +call it: the periodic scan and the fast path taken when the transport confirms +a Welcome went out. A supersede in the scan alone left the fast path's probe +with its full ladder. The same function sets the next due time, so the scan +does not supersede a fast-path probe before its acknowledgement can arrive. + +Superseding is not a delivery failure and is not counted against the carrier. +It does not discard the relay's answer either: a verdict finds its peer through +the outbox entry, so the superseded id is remembered, and a verdict slower than +one probe interval (but faster than two) still backs the cadence off. Dropping +it left a peer the relay had refused probed every five seconds, which is the +rate that flaps the relay connection. Only the last superseded id is kept: a +verdict that misses two intervals finds neither, and the cadence backs off on +the next probe's verdict instead. Relay verdicts arrive well inside one. + +A probe is not persisted, and one an older build persisted is dropped at +restore: the new process does not know which probe was the last, so a restored +one could never be superseded, and the next scan asks the same question. For +the same reason a probe is withdrawn with the rest of the confirmation tracking +when the session confirms or is torn down: a probe the tracking no longer names +could never be superseded, and would keep its ladder. + ### Both-create convergence Both peers can create a session simultaneously. The tiebreaker orders the two