@@ -9,9 +9,6 @@ use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
99use std::sync::Arc;
1010use std::time::{Duration, SystemTime, UNIX_EPOCH};
1111
12−use base64::prelude::*;
13−use base64::Engine;
14−
1512use crux_core::{
1613 macros::effect,
1714 render::{render, RenderOperation},
@@ -61,24 +58,6 @@ const MAX_CONTACTS: usize = 500;
6158/// the budget is dropped exactly like an undecryptable blob would be.
6259const MAX_POLL_TOTAL_BYTES: usize = 64 * 1024 * 1024;
6360
64−/// Cap on the number of events we will attempt to decrypt out of a single poll
65−/// page, independent of the byte budget. Minimum-size blobs (173 bytes decoded)
66−/// would otherwise permit ~388K events within the 64 MiB byte budget, each
67−/// costing a full scalar mult + SHA-512 + AEAD pass — enough to freeze the app
68−/// for tens of seconds per poll. An honest relay caps at 50 events
69−/// (`maxPollEvents`); this is a hostile-relay bound. Set above `MAX_ACK_IDS`
70−/// (1024) so a legitimate large page from one sender still fully processes.
71−const MAX_POLL_EVENTS: usize = 2000;
72−
73−/// Upper bound on a poll response body we will buffer and parse, in bytes.
74−///
75−/// Before the per-blob byte/count budgets in `ingest_poll` ever run, the client
76−/// reads the entire response body into a `Vec<u8>` and deserializes it. A
77−/// multi-gigabyte response from a hostile relay would OOM the app before
78−/// `ingest_poll`'s defenses are consulted. This cap (generously above what 50
79−/// max-size blobs + JSON envelope could cost) bounds the allocation.
80−const MAX_POLL_RESPONSE_BYTES: usize = 64 * 1024 * 1024;
81−
8261/// How long a poll may stay in flight before we assume the shell lost it.
8362///
8463/// `polling` gates the whole long-poll loop, and it is only cleared by a
@@ -108,12 +87,6 @@ const POLL_WATCHDOG_MS: u64 = 90_000;
10887/// one costs a redundant request.
10988const FLUSH_WATCHDOG_MS: u64 = 90_000;
11089
111−/// How long an auth round-trip may stay in flight before we assume the
112−/// shell lost it. Same rationale as `POLL_WATCHDOG_MS`: `authenticating`
113−/// gates the whole auth flow, and without a watchdog a dropped effect
114−/// leaves it stuck forever. Kept above the 70 s shell HTTP timeout.
115−const AUTH_WATCHDOG_MS: u64 = 90_000;
116−
11790/// Messages per `/messages` request.
11891///
11992/// The relay rejects a longer batch outright (`413 batch_too_large`,
@@ -147,18 +120,10 @@ const K_PROFILE: &str = "profile";
147120const K_CONTACTS: &str = "contacts";
148121const K_CURSOR: &str = "cursor";
149122const K_OUTBOX: &str = "outbox";
150−const K_SEEN_IDS: &str = "seen_ids";
151123fn k_messages(peer: &str) -> String {
152124 format!("messages:{peer}")
153125}
154126
155−/// N7: Cap on the per-peer `seen_ids` dedup set. Without a cap the set grows
156−/// without bound — a peer who sends thousands of messages accumulates
157−/// thousands of IDs, and the `seen_ids` kv blob is rewritten in full on every
158−/// incoming message. Kept well above `MAX_MESSAGES_PER_PEER` (1000) so a
159−/// trim-and-replay within the conversation window still hits the set.
160−const MAX_SEEN_IDS_PER_PEER: usize = 2000;
161−
162127// ---------------------------------------------------------------------------
163128// Effect
164129// ---------------------------------------------------------------------------
@@ -226,9 +191,6 @@ pub enum Event {
226191 LoadedOutbox(#[facet(opaque)] KvData),
227192 #[serde(skip)]
228193 #[facet(skip)]
229− LoadedSeenIds(#[facet(opaque)] KvData),
230− #[serde(skip)]
231− #[facet(skip)]
232194 LoadedMessages(String, #[facet(opaque)] KvData),
233195 #[serde(skip)]
234196 #[facet(skip)]
@@ -262,17 +224,11 @@ pub enum Event {
262224 #[serde(skip)]
263225 #[facet(skip)]
264226 StartFlush,
265− /// Fires `FLUSH_WATCHDOG_MS` after generation `n`'s flush is issued; a no-op
266− /// unless that exact send is somehow still in flight — same pattern as
267− /// `PollWatchdog`.
268− #[serde(skip)]
269− #[facet(skip)]
270− FlushWatchdog(u64),
271− /// Fires `AUTH_WATCHDOG_MS` after generation `n`'s auth round-trip is
272− /// issued; a no-op unless that exact auth is somehow still in flight.
227+ /// Fires `FLUSH_WATCHDOG_MS` after a flush is issued; a no-op unless the
228+ /// send is somehow still in flight — same pattern as `PollWatchdog`.
273229 #[serde(skip)]
274230 #[facet(skip)]
275− AuthWatchdog(u64),
231+ FlushWatchdog,
276232 #[serde(skip)]
277233 #[facet(skip)]
278234 ChallengeResult(u64, #[facet(opaque)] HttpResult),
@@ -284,7 +240,7 @@ pub enum Event {
284240 PollResult(u64, #[facet(opaque)] HttpResult),
285241 #[serde(skip)]
286242 #[facet(skip)]
287− SendResult(u64, #[facet(opaque)] HttpResult),
243+ SendResult(#[facet(opaque)] HttpResult),
288244}
289245
290246// ---------------------------------------------------------------------------
@@ -310,9 +266,9 @@ pub enum Page {
310266 EditProfile,
311267}
312268
313−/// The six kv keys loaded at startup (`settings`, `profile`, `contacts`,
314−/// `cursor`, `outbox`, `seen_ids`). `Connect` waits for all of them.
315−const STARTUP_LOADS: u8 = 6;
269+/// The five kv keys loaded at startup (`settings`, `profile`, `contacts`,
270+/// `cursor`, `outbox`). `Connect` waits for all of them.
271+const STARTUP_LOADS: u8 = 5;
316272
317273/// The batch a live `/messages` request is carrying: the first `count` items of
318274/// the outbox, all bound for `recipient`.
@@ -338,12 +294,6 @@ pub struct Model {
338294 profile: OwnProfile,
339295 contacts: BTreeMap<String, Contact>,
340296 messages: BTreeMap<String, Vec<StoredMessage>>,
341− /// Q18: Per-peer dedup ID set, independent of the trimmed conversation.
342− /// Without this, a replayed blob whose `id` was aged out by `trim_history`
343− /// (past the 1000-message cap) passes the dedup check and re-enters the
344− /// conversation. This set is loaded from and persisted to kv alongside
345− /// messages, so it survives restarts.
346− seen_ids: BTreeMap<String, HashSet<String>>,
347297 outbox: VecDeque<OutboxItem>,
348298 cursor: i64,
349299 token: Option<String>,
@@ -386,12 +336,6 @@ pub struct Model {
386336 /// items it sent off the head of the outbox, so two overlapping sends would
387337 /// pop each other's messages.
388338 flushing: bool,
389− /// Which flush the in-flight `SendResult` / `FlushWatchdog` is allowed to
390− /// belong to. Same pattern as `poll_gen`: bumped every time a flush is
391− /// issued and every time one is abandoned, so a `SendResult` or watchdog
392− /// from a superseded send cannot clear the guard or pop items belonging to
393− /// the replacement.
394− flush_gen: u64,
395339 /// What the in-flight send will pop off the head of the outbox when it
396340 /// succeeds — see `InFlight`. `None` while nothing is in flight, and also
397341 /// while a send is in flight whose items have since been removed from under
@@ -456,7 +400,6 @@ impl Default for Model {
456400 profile: OwnProfile::default(),
457401 contacts: BTreeMap::new(),
458402 messages: BTreeMap::new(),
459− seen_ids: BTreeMap::new(),
460403 outbox: VecDeque::new(),
461404 cursor: 0,
462405 token: None,
@@ -468,7 +411,6 @@ impl Default for Model {
468411 authenticating: false,
469412 auth_gen: 0,
470413 flushing: false,
471− flush_gen: 0,
472414 in_flight: None,
473415 send_batch_budget: None,
474416 loads_pending: 0,
@@ -520,24 +462,6 @@ impl Model {
520462 self.auth_gen
521463 }
522464
523− /// Open a new flush generation and return it. Same pattern as
524− /// `next_poll_gen`: a `SendResult` or `FlushWatchdog` whose generation no
525− /// longer matches is dropped, so a late result from a superseded send
526− /// cannot clear the guard or pop items belonging to the replacement.
527− fn next_flush_gen(&mut self) -> u64 {
528− self.flush_gen = self.flush_gen.wrapping_add(1);
529− self.flush_gen
530− }
531−
532− /// Abandon the in-flight flush, if any: release the guard and retire the
533− /// generation so a late `SendResult` cannot clear the *replacement* flush's
534− /// guard or pop its items.
535− fn abandon_flush(&mut self) {
536− self.flushing = false;
537− self.in_flight = None;
538− self.flush_gen = self.flush_gen.wrapping_add(1);
539− }
540−
541465 /// Abandon the in-flight auth round-trip (either leg).
542466 fn abandon_auth(&mut self) {
543467 self.authenticating = false;
@@ -584,7 +508,7 @@ fn now_ms() -> i64 {
584508 SystemTime::now()
585509 .duration_since(UNIX_EPOCH)
586510 .map(|d| d.as_millis() as i64)
587− .unwrap_or(1)
511+ .unwrap_or(0)
588512}
589513
590514/// Bare lowercased hostname of a server URL (no scheme/port/trailing dot).
@@ -672,16 +596,6 @@ struct PollEvent {
672596 #[serde(rename = "encryptedBlob")]
673597 encrypted_blob: String,
674598}
675−
676−/// Q17: Deferred delivery-ack accumulator. Collects ack_ids and the max `ts`
677−/// across all `delivery.ack` payloads from one sender in a single poll page,
678−/// so the staleness check and watermark update run once per sender after the
679−/// event loop — immune to relay page ordering.
680−#[derive(Default)]
681−struct PendingAck {
682− ts: i64,
683− ids: HashSet<String>,
684−}
685599#[derive(Serialize)]
686600struct SendBatch {
687601 messages: Vec<Envelope>,
@@ -725,7 +639,6 @@ impl App for Skrepka {
725639 KeyValue::get(K_CONTACTS).then_send(Event::LoadedContacts),
726640 KeyValue::get(K_CURSOR).then_send(Event::LoadedCursor),
727641 KeyValue::get(K_OUTBOX).then_send(Event::LoadedOutbox),
728− KeyValue::get(K_SEEN_IDS).then_send(Event::LoadedSeenIds),
729642 ])
730643 .and(render())
731644 }
@@ -803,51 +716,10 @@ impl App for Skrepka {
803716 }
804717 self.startup_done(model).and(render())
805718 }
806− Event::LoadedSeenIds(res) => {
807− match parse_kv::<BTreeMap<String, HashSet<String>>>(res) {
808− Ok(Some(map)) => model.seen_ids = map,
809− Ok(None) => {}
810− Err(()) => {
811− // Non-fatal: dedup falls back to the conversation-derived
812− // set, which is weaker but functional.
813− model.seen_ids = BTreeMap::new();
814− }
815− }
816− self.startup_done(model).and(render())
817− }
818719 Event::LoadedMessages(peer, res) => {
819720 match parse_kv::<Vec<StoredMessage>>(res) {
820721 Ok(Some(list)) => {
821− // D5: Add on-disk message IDs to seen_ids so future
822− // replays don't pass dedup. Without this, a seen_ids
823− // kv load failure leaves the set empty, and on-disk
824− // messages aren't protected against replay.
825− if let Some(id_set) = model.seen_ids.get_mut(&peer) {
826− id_set.extend(list.iter().map(|m| m.id.clone()));
827− }
828− // Merge: a poll that beat this lazy load may have already
829− // inserted messages for this peer. Overwriting with the
830− // on-disk list (which the GET read before the poll wrote)
831− // would silently destroy the freshly-ingested messages.
832− // Union by id, keeping both the on-disk history and any
833− // in-memory additions.
834− match model.messages.get_mut(&peer) {
835− Some(existing) => {
836− let existing_ids: HashSet<&String> =
837− existing.iter().map(|m| &m.id).collect();
838− let new_msgs: Vec<StoredMessage> = list
839− .into_iter()
840− .filter(|m| !existing_ids.contains(&m.id))
841− .collect();
842− existing.extend(new_msgs);
843− // Re-sort by ts to maintain the invariant.
844− existing.sort_by_key(|m| m.ts);
845− trim_history(existing);
846− }
847− None => {
848− model.messages.insert(peer, list);
849− }
850− }
722+ model.messages.insert(peer, list);
851723 }
852724 Ok(None) => {}
853725 Err(()) => {
@@ -957,20 +829,11 @@ impl App for Skrepka {
957829 // would then ack a sequence number B never issued, and B would
958830 // drop every message waiting for us.
959831 model.abandon_poll();
960− // ...and retire the send that is still in flight *to relay A*,
961− // for the same reason: its `SendResult` carries A's acceptance,
962− // and popping items here would eat whatever the new relay's
963− // send is about to carry. The generation check in `SendResult`
964− // is what makes the late result a no-op.
965− model.abandon_flush();
966832 // The old relay's rate limit was the old relay's opinion. A new one is
967833 // entitled to take our sends immediately, and carrying the pause over
968834 // would park the outbox against a limit that no longer applies.
969835 model.flush_paused_until = 0;
970836 model.send_retries = 0;
971− // D2: Reset poll retries too — the new relay's poll failures
972− // should start fresh, not inherit the old relay's backoff.
973− model.poll_retries = 0;
974837 let writes = if self.refuse_write(model) {
975838 Command::done()
976839 } else {
@@ -1077,7 +940,6 @@ impl App for Skrepka {
1077940 Event::DeleteContact { peer } => {
1078941 model.contacts.remove(&peer);
1079942 model.messages.remove(&peer);
1080− model.seen_ids.remove(&peer);
1081943 model.outbox.retain(|item| item.recipient != peer);
1082944 // The send in flight may be carrying the very items we just
1083945 // dropped; if so its result must pop nothing.
@@ -1094,8 +956,6 @@ impl App for Skrepka {
1094956 // message the user just asked us to forget. Delete the key.
1095957 KeyValue::delete(k_messages(&peer)).then_send(Event::Saved),
1096958 self.persist_outbox(model),
1097− // Q18: persist the seen_ids map without the removed peer.
1098− self.persist_seen_ids(model),
1099959 ])
1100960 .and(render())
1101961 }
@@ -1123,13 +983,6 @@ impl App for Skrepka {
1123983 model.error = Some("photo too large".into());
1124984 return render();
1125985 }
1126− // S6: Validate the photo is valid base64 before broadcasting —
1127− // the recipient's parse_payload rejects invalid base64, so an
1128− // invalid photo would be silently dropped by every recipient.
1129− if photo.as_ref().is_some_and(|p| BASE64_STANDARD.decode(p).is_err()) {
1130− model.error = Some("photo is not valid base64".into());
1131− return render();
1132− }
1133986 model.profile = OwnProfile {
1134987 display_name,
1135988 bio,
@@ -1138,8 +991,8 @@ impl App for Skrepka {
1138991 model.page = Page::Conversations;
1139992 let ts = now_ms();
1140993 let payload = Payload::Profile {
1141− display_name: Some(model.profile.display_name.clone()),
1142− bio: Some(model.profile.bio.clone()),
994+ display_name: model.profile.display_name.clone(),
995+ bio: model.profile.bio.clone(),
1143996 photo: model.profile.photo.clone(),
1144997 };
1145998 // Broadcast the new profile to every contact we haven't blocked.
@@ -1218,11 +1071,6 @@ impl App for Skrepka {
12181071 let gen = model.next_auth_gen();
12191072 req.build()
12201073 .then_send(move |r| Event::ChallengeResult(gen, r))
1221− .and(
1222− Time::notify_after(Duration::from_millis(AUTH_WATCHDOG_MS))
1223− .0
1224− .then_send(move |_| Event::AuthWatchdog(gen)),
1225− )
12261074 .and(render())
12271075 }
12281076 Err(_) => {
@@ -1401,90 +1249,35 @@ impl App for Skrepka {
14011249 return render();
14021250 }
14031251 let bytes = resp.take_body().unwrap_or_default();
1404− // Bound the response before parsing: a hostile relay can return
1405− // a multi-gigabyte body that OOMs the app before `ingest_poll`'s
1406− // per-blob budgets are even consulted.
1407− if bytes.len() > MAX_POLL_RESPONSE_BYTES {
1408− model.error = Some("relay sent an oversized poll response".into());
1409− return self.backoff_poll(model);
1410− }
1411− // Q25: A malformed or unparseable poll response must not be
1412− // treated as an empty page — `unwrap_or_default()` would adopt
1413− // cursor 0, forcing a re-fetch from the beginning and wasting
1414− // bandwidth on already-seen (deduped) messages. Back off instead.
1415− let parsed: PollResp = match serde_json::from_slice(&bytes) {
1416− Ok(p) => p,
1417− Err(_) => {
1418− model.error = Some("relay sent a malformed poll response".into());
1419− return self.backoff_poll(model);
1420− }
1421− };
1422− let cursor_before = model.cursor;
1252+ let parsed: PollResp = serde_json::from_slice(&bytes).unwrap_or_default();
14231253 let cmd = self.ingest_poll(model, parsed);
14241254 // The re-poll hangs off the *cursor write*, not off this event.
14251255 // Polling again is what acks the batch we just took — the server
14261256 // deletes everything up to `cursor` — so it must not happen until
14271257 // the batch is durable on our side. `SavedCursor` re-polls.
1428− //
1429− // Arm a watchdog alongside the cursor write: if the shell ever
1430− // drops the KV effect (suspension, bug), `SavedCursor` never fires,
1431− // `polling` stays `false`, and the loop is dead until the next
1432− // foreground `Connect`. The watchdog fires `Poll` if the loop is
1433− // still idle — same pattern as the poll/send HTTP watchdogs.
1434− //
1435− // Skip the cursor write when the cursor did not advance (an empty
1436− // or fully-deduped page): there is nothing new to persist, so
1437− // re-poll directly. The watchdog is still armed as a fallback.
1438− let cursor_advanced = model.cursor != cursor_before;
1439− cmd.and(if cursor_advanced {
1258+ cmd.and(
14401259 KeyValue::set(K_CURSOR, json_bytes(&model.cursor))
1441− .then_send(Event::SavedCursor)
1442− } else {
1443− Command::event(Event::Poll)
1444− })
1445− .and(
1446− Time::notify_after(Duration::from_millis(POLL_WATCHDOG_MS))
1447− .0
1448− .then_send(|_| Event::Poll),
1260+ .then_send(Event::SavedCursor),
14491261 )
14501262 .and(render())
14511263 }
14521264
14531265 // ---------------- outbox ----------------
14541266 Event::StartFlush => self.flush_next(model),
1455− Event::FlushWatchdog(gen) => {
1456− // Only the flush this watchdog was armed for, and only while it
1457− // is still the current one and still in flight. A watchdog whose
1458− // send already completed (or was superseded) is a stale timer and
1459− // must not touch a healthy successor — same pattern as
1460− // `PollWatchdog`.
1461− if gen != model.flush_gen || !model.flushing {
1267+ Event::FlushWatchdog => {
1268+ // The send request never came back — the shell dropped the HTTP
1269+ // effect (backgrounded app, bug), and `flushing` has been stuck
1270+ // ever since. Release the guard and retry the flush. The late
1271+ // `SendResult`, if it ever arrives, finds `in_flight` already
1272+ // cleared and pops nothing.
1273+ if !model.flushing {
14621274 return Command::done();
14631275 }
1464− // Retire it as we give up on it. The request itself is not
1465− // cancelled, so if it eventually resolves the generation check
1466− // in `SendResult` is what stops it from popping the replacement's
1467− // items.
1468− model.abandon_flush();
1276+ model.flushing = false;
1277+ model.in_flight = None;
14691278 self.flush_next(model)
14701279 }
1471− Event::AuthWatchdog(gen) => {
1472− // Only the auth this watchdog was armed for, and only while it is
1473− // still the current one and still in flight.
1474− if gen != model.auth_gen || !model.authenticating {
1475− return Command::done();
1476− }
1477− model.abandon_auth();
1478− Command::event(Event::Authenticate)
1479− }
1480− Event::SendResult(gen, Ok(resp)) => {
1481− // A result for a send we already abandoned (a watchdog gave up
1482− // on it, or a server change invalidated it). Dropping it is safe
1483− // and required: its items were re-queued, so popping by count
1484− // here would eat whichever messages slid into their place.
1485− if gen != model.flush_gen {
1486− return Command::done();
1487− }
1280+ Event::SendResult(Ok(resp)) => {
14881281 // The in-flight send is over; `flush_next` refuses to run while
14891282 // this is set, so it must be cleared before *any* branch below.
14901283 model.flushing = false;
@@ -1574,24 +1367,16 @@ impl App for Skrepka {
15741367 // later, under load that has nothing to do with this one — starts
15751368 // at whatever height the last one climbed to.
15761369 model.send_retries = 0;
1577− // D3-app: Don't pop if kv is broken — the persist would be
1578− // refused, and the items would be lost from memory but still
1579− // on disk, causing duplicate sends on relaunch.
1580− if !model.kv_load_failed {
1581− for _ in 0..batch.map_or(0, |b| b.count) {
1582− model.outbox.pop_front();
1583− }
1370+ for _ in 0..batch.map_or(0, |b| b.count) {
1371+ model.outbox.pop_front();
15841372 }
15851373 return self.persist_outbox(model).and(self.flush_next(model));
15861374 }
15871375 // A transient failure: a 5xx, a rate limit, a relay restarting.
15881376 self.retry_batch(model, batch)
15891377 }
1590− Event::SendResult(gen, Err(_)) => {
1378+ Event::SendResult(Err(_)) => {
15911379 // A transport error is the same transient failure as a 5xx.
1592− if gen != model.flush_gen {
1593− return Command::done();
1594− }
15951380 model.flushing = false;
15961381 let batch = model.in_flight.take();
15971382 self.retry_batch(model, batch)
@@ -1766,16 +1551,6 @@ impl Skrepka {
17661551 KeyValue::set(k_messages(peer), json_bytes(&list)).then_send(Event::Saved)
17671552 }
17681553
1769− /// Q18: Persist the per-peer dedup ID set. Written alongside messages so
1770− /// the set survives restarts and covers IDs that `trim_history` has aged
1771− /// out of the conversation.
1772− fn persist_seen_ids(&self, model: &mut Model) -> Command<Effect, Event> {
1773− if self.refuse_write(model) {
1774− return Command::done();
1775− }
1776− KeyValue::set(K_SEEN_IDS, json_bytes(&model.seen_ids)).then_send(Event::Saved)
1777− }
1778−
17791554 fn schedule_reconnect(&self, model: &mut Model) -> Command<Effect, Event> {
17801555 let delay = backoff_ms(model.poll_retries);
17811556 model.poll_retries = model.poll_retries.saturating_add(1);
@@ -1874,21 +1649,14 @@ impl Skrepka {
18741649 .insert(peer.clone(), Contact::new(peer.clone(), String::new(), ts));
18751650 }
18761651
1877− let text = Payload::Text { id, body };
1878− model.outbox.push_back(OutboxItem::new(
1879− peer.clone(),
1880− Arc::new(protocol::serialize_payload(&text, ts)),
1881− ));
1882−
1883− // On first contact, share our profile *after* the text. The text
1884− // auto-creates the contact on the recipient's side; the profile that
1885− // follows in the same batch then applies to the now-existing contact.
1886− // (Previously profile was queued before the text, but the recipient's
1887− // ingest drops profiles from non-contacts, so the profile was lost.)
1652+ // On first contact, share our profile *before* the text so the
1653+ // recipient's first impression is a name and avatar, not a truncated
1654+ // @p that gets replaced a moment later. Both are queued behind any
1655+ // in-flight items, so the ordering is preserved through the outbox.
18881656 if was_new {
18891657 let profile = Payload::Profile {
1890− display_name: Some(model.profile.display_name.clone()),
1891− bio: Some(model.profile.bio.clone()),
1658+ display_name: model.profile.display_name.clone(),
1659+ bio: model.profile.bio.clone(),
18921660 photo: model.profile.photo.clone(),
18931661 };
18941662 model.outbox.push_back(OutboxItem::profile(
@@ -1897,6 +1665,12 @@ impl Skrepka {
18971665 ));
18981666 }
18991667
1668+ let text = Payload::Text { id, body };
1669+ model.outbox.push_back(OutboxItem::new(
1670+ peer.clone(),
1671+ Arc::new(protocol::serialize_payload(&text, ts)),
1672+ ));
1673+
19001674 Command::all([
19011675 self.persist_messages(model, &peer),
19021676 self.persist_contacts(model),
@@ -2024,7 +1798,6 @@ impl Skrepka {
20241798 };
20251799
20261800 model.flushing = true;
2027− let gen = model.next_flush_gen();
20281801 model.in_flight = Some(InFlight {
20291802 recipient,
20301803 count: messages.len(),
@@ -2035,11 +1808,11 @@ impl Skrepka {
20351808 .header("authorization", format!("Bearer {token}"))
20361809 .body_json(&SendBatch { messages })
20371810 {
2038− Ok(req) => persist.and(req.build().then_send(move |r| Event::SendResult(gen, r)))
1811+ Ok(req) => persist.and(req.build().then_send(Event::SendResult))
20391812 .and(
20401813 Time::notify_after(Duration::from_millis(FLUSH_WATCHDOG_MS))
20411814 .0
2042− .then_send(move |_| Event::FlushWatchdog(gen)),
1815+ .then_send(|_| Event::FlushWatchdog),
20431816 ),
20441817 Err(_) => {
20451818 // Not a dead loop: the next StartFlush (from a poll page or a
@@ -2110,20 +1883,8 @@ impl Skrepka {
21101883 return Command::done();
21111884 }
21121885 let prev_cursor = model.cursor;
2113− // Clamp the relay-supplied cursor: honest cursors are wall-clock-seeded
2114− // sequence numbers, so anything far in the future is a hostile or buggy
2115− // relay. Adopting `i64::MAX` would ack (delete) all queued mail on the
2116− // next poll and persist that poison across relaunches.
2117− let clamped_cursor = page.cursor.min(now_ms().saturating_add(60_000));
2118− // Q16: Only advance the cursor when the page actually carried events.
2119− // An empty page with a cursor higher than ours is suspicious — a
2120− // compromised relay could poison the cursor to ack (delete) messages
2121− // the client hasn't seen. When there are no events, there is nothing
2122− // to acknowledge, so the cursor should not move. (An honest relay's
2123− // empty-page cursor equals what the client sent, so this is a no-op
2124− // in the normal case.)
2125− if !page.events.is_empty() && clamped_cursor > model.cursor {
2126− model.cursor = clamped_cursor;
1886+ if page.cursor > model.cursor {
1887+ model.cursor = page.cursor;
21271888 }
21281889 let Some(id) = model.identity() else {
21291890 return Command::done();
@@ -2132,24 +1893,12 @@ impl Skrepka {
21321893 let now = now_ms();
21331894 let mut touched_convos: HashSet<String> = HashSet::new();
21341895 let mut contacts_dirty = false;
2135− let mut outbox_dirty = false;
21361896 // sender hex -> message ids needing acks
21371897 let mut ack_targets: BTreeMap<String, Vec<String>> = BTreeMap::new();
2138− // Q17: Per-sender deferred ack accumulation. Collected during the event
2139− // loop, applied after it — so relay page ordering cannot cause an older
2140− // ack to be dropped as stale by a newer one processed first.
2141− let mut pending_acks: HashMap<String, PendingAck> = HashMap::new();
1898+ // Lazily-built per-sender sets of known message IDs for O(1) text dedup.
1899+ let mut seen_ids: HashMap<String, HashSet<String>> = HashMap::new();
21421900
21431901 let mut budget = MAX_POLL_TOTAL_BYTES;
2144− let mut events_processed = 0usize;
2145− // N9: Track whether any event was successfully decrypted and stored.
2146− // When the budget/count is exhausted mid-page, we ack the page (keep
2147− // the advanced cursor) only if at least one event was processed —
2148− // the unprocessed tail is from a hostile/misconfigured relay and
2149− // losing them is better than livelocking. If zero events were
2150− // processed (a completely hostile page), restore the cursor so we
2151− // don't ack mail we never saw.
2152− let mut events_stored = 0usize;
21531902 for ev in page.events {
21541903 // Reject an oversized blob on its hex length, before it costs us a
21551904 // decode. No conforming sender produces one (the relay caps blobs at
@@ -2157,37 +1906,19 @@ impl Skrepka {
21571906 if ev.encrypted_blob.len() > MAX_BLOB_LEN * 2 {
21581907 continue;
21591908 }
2160− // Event count cap: the byte budget alone permits ~388K minimum-size
2161− // events, each costing a full decrypt attempt. A hostile relay can
2162− // freeze the app for tens of seconds per poll with a page of small
2163− // junk blobs. Stop processing past this count.
2164− if events_processed >= MAX_POLL_EVENTS {
2165− // N9: If we stored at least one event, keep the advanced cursor
2166− // to ack the page and avoid a livelock. The unprocessed events
2167− // are hostile junk or beyond our capacity — we can't process
2168− // them anyway, and re-fetching them forever would wedge the
2169− // client.
2170− if events_stored == 0 {
2171− model.cursor = prev_cursor;
2172− }
2173− model.error = Some("relay sent too many poll events".into());
2174− break;
2175− }
2176− events_processed += 1;
21771909 match budget.checked_sub(ev.encrypted_blob.len() / 2) {
21781910 Some(left) => budget = left,
21791911 None => {
2180− // N9: Budget exhausted. If we stored at least one event,
2181− // keep the advanced cursor to ack the page — re-fetching
2182− // an oversized page forever is a livelock. The unprocessed
2183− // events are beyond our capacity and would be re-served
2184− // on every poll, wedging the client. Acking them breaks
2185− // the loop. If zero events were stored (completely hostile
2186− // page), restore the cursor to avoid acking mail we
2187− // never saw.
2188− if events_stored == 0 {
2189− model.cursor = prev_cursor;
2190− }
1912+ // A page this large is a misbehaving relay, not real mail.
1913+ // Stop here; the rest of the page is discarded exactly as an
1914+ // undecryptable blob would be.
1915+ //
1916+ // Restore the cursor: it was advanced to `page.cursor` above,
1917+ // but we have not processed the remaining events. If we keep
1918+ // the advanced cursor, the next poll acks the whole page to
1919+ // the relay — including every event we skipped — and the
1920+ // relay deletes them. Those messages are permanently lost.
1921+ model.cursor = prev_cursor;
21911922 model.error = Some("relay sent an oversized poll page".into());
21921923 break;
21931924 }
@@ -2201,10 +1932,6 @@ impl Skrepka {
22011932 let Some(parsed) = protocol::parse_payload(&dec.plaintext) else {
22021933 continue;
22031934 };
2204− // D1: Any event that was successfully decrypted AND parsed counts
2205− // as "stored" for cursor advancement — the client has processed it
2206− // even if it was a duplicate/stale/stranger that was dropped.
2207− events_stored += 1;
22081935 let sender = dec.sender_hex;
22091936
22101937 // Mail from ourselves. A relay cannot forge this — the sender identity
@@ -2256,35 +1983,12 @@ impl Skrepka {
22561983 Contact::new(sender.clone(), String::new(), ts),
22571984 );
22581985 contacts_dirty = true;
2259− // D6-app: Send our profile to the new contact so they
2260− // see our display name and avatar, not just our @p.
2261− let profile = Payload::Profile {
2262− display_name: Some(model.profile.display_name.clone()),
2263− bio: Some(model.profile.bio.clone()),
2264− photo: model.profile.photo.clone(),
2265− };
2266− model.outbox.push_back(OutboxItem::profile(
2267− sender.clone(),
2268− Arc::new(protocol::serialize_payload(&profile, ts)),
2269− ));
2270− outbox_dirty = true;
22711986 }
22721987 let convo = model.messages.entry(sender.clone()).or_default();
2273− // Q18: Use the persistent per-peer dedup set from
2274− // `model.seen_ids` (loaded from kv at startup) instead of
2275− // deriving it from the trimmed conversation. Without this,
2276− // a replayed blob whose `id` was aged out by `trim_history`
2277− // (past the 1000-message cap) passes the dedup check and
2278− // re-enters the conversation.
2279− let id_set = model
2280− .seen_ids
2281− .entry(sender.clone())
2282− .or_default();
2283− // Also seed from the conversation in case the kv load was
2284− // missing or partial — covers the upgrade path.
2285− if id_set.is_empty() {
2286− id_set.extend(convo.iter().map(|m| m.id.clone()));
2287− }
1988+ // Lazily build the per-sender ID set on first touch.
1989+ let id_set = seen_ids.entry(sender.clone()).or_insert_with(|| {
1990+ convo.iter().map(|m| m.id.clone()).collect()
1991+ });
22881992 // dedup by id — O(1) via HashSet
22891993 if !id_set.contains(&msg_id) {
22901994 // The conversation is *kept* sorted rather than re-sorted:
@@ -2302,29 +2006,6 @@ impl Skrepka {
23022006 },
23032007 );
23042008 id_set.insert(msg_id.clone());
2305− // N7: Cap the dedup set so it doesn't grow without
2306− // bound. The set is rewritten to kv on every message
2307− // receipt, so an unbounded set makes every receipt
2308− // slower. Keep the most recent IDs by draining excess
2309− // from the oldest. `HashSet` has no ordering, so when
2310− // the cap is exceeded we rebuild from the conversation
2311− // (which is trimmed to MAX_MESSAGES_PER_PEER and
2312− // sorted by ts) — the oldest IDs in the set that are
2313− // no longer in the conversation are the ones to drop.
2314− if id_set.len() > MAX_SEEN_IDS_PER_PEER {
2315− let convo_ids: HashSet<&String> =
2316− convo.iter().map(|m| &m.id).collect();
2317− let excess = id_set.len() - MAX_SEEN_IDS_PER_PEER;
2318− let to_remove: Vec<String> = id_set
2319− .iter()
2320− .filter(|id| !convo_ids.contains(*id))
2321− .take(excess)
2322− .cloned()
2323− .collect();
2324− for id in to_remove {
2325− id_set.remove(&id);
2326− }
2327− }
23282009 trim_history(convo);
23292010 touched_convos.insert(sender.clone());
23302011 // Only ack a message we actually stored — a duplicate id
@@ -2333,26 +2014,39 @@ impl Skrepka {
23332014 }
23342015 }
23352016 Payload::DeliveryAck { ack_ids } => {
2336− // Q17: Defer the staleness check until after the event loop.
2337− // Within a single page, a relay can reorder events so that a
2338− // newer ack (higher ts) is processed before an older one. If
2339− // we applied the watermark immediately, the older ack would be
2340− // dropped as stale — even if it acks different message IDs.
2341− // Instead, accumulate all ack_ids per sender and the max ts,
2342− // then apply them in a second pass below the loop.
2343− let prev_ts = pending_acks
2017+ // Staleness check: drop a replayed ack whose ts predates the
2018+ // last one we processed for this contact.
2019+ let stale = model
2020+ .contacts
23442021 .get(&sender)
2345− .map(|p| p.ts)
2346− .unwrap_or(0);
2347− if ts > prev_ts {
2348− let entry = pending_acks.entry(sender.clone()).or_default();
2349− entry.ts = ts;
2022+ .is_some_and(|c| ts < c.last_ack_ts);
2023+ if stale {
2024+ continue;
2025+ }
2026+ // The watermark is clamped to *now*, not merely into the skew
2027+ // window: `ts` is the peer's to choose, and `last_ack_ts` is
2028+ // what every later ack is measured against. A single ack at
2029+ // `ts = i64::MAX` would otherwise park the watermark a minute
2030+ // into the future, and every honest ack that followed — the
2031+ // ones that actually mark our messages delivered — would read
2032+ // as stale and be dropped.
2033+ if let Some(c) = model.contacts.get_mut(&sender) {
2034+ c.last_ack_ts = ts.min(now);
2035+ contacts_dirty = true;
2036+ }
2037+ let ack_set: HashSet<String> = ack_ids.iter().cloned().collect();
2038+ if let Some(convo) = model.messages.get_mut(&sender) {
2039+ let mut changed = false;
2040+ for m in convo.iter_mut() {
2041+ if m.outgoing && ack_set.contains(&m.id) && !m.delivered {
2042+ m.delivered = true;
2043+ changed = true;
2044+ }
2045+ }
2046+ if changed {
2047+ touched_convos.insert(sender);
2048+ }
23502049 }
2351− pending_acks
2352− .entry(sender.clone())
2353− .or_default()
2354− .ids
2355− .extend(ack_ids);
23562050 }
23572051 Payload::Profile {
23582052 display_name,
@@ -2379,17 +2073,9 @@ impl Skrepka {
23792073 // A replayed profile at the same `ts` overwrites with identical
23802074 // content, which is a no-op in practice.
23812075 if ts >= c.last_profile_ts {
2382− // D13: Only update fields that are present (Some).
2383− // An absent field (None) means "don't change".
2384− if let Some(name) = display_name {
2385− c.display_name = name;
2386− }
2387− if let Some(b) = bio {
2388− c.bio = b;
2389− }
2390− if let Some(p) = photo {
2391− c.photo = Some(p);
2392− }
2076+ c.display_name = display_name;
2077+ c.bio = bio;
2078+ c.photo = photo;
23932079 c.last_profile_ts = ts.min(now);
23942080 contacts_dirty = true;
23952081 }
@@ -2397,42 +2083,6 @@ impl Skrepka {
23972083 }
23982084 }
23992085
2400− // Q17: Apply deferred delivery acks. Collected during the event loop
2401− // above, now applied with a single staleness check per sender using the
2402− // highest `ts` seen — so relay page ordering cannot cause an older ack
2403− // to be dropped as stale by a newer one processed first.
2404− for (sender, pending) in &pending_acks {
2405− let stale = model
2406− .contacts
2407− .get(sender)
2408− .is_some_and(|c| pending.ts < c.last_ack_ts);
2409− if stale {
2410− continue;
2411− }
2412− // The watermark is clamped to *now*, not merely into the skew window:
2413− // `ts` is the peer's to choose, and `last_ack_ts` is what every later
2414− // ack is measured against. A single ack at `ts = i64::MAX` would
2415− // otherwise park the watermark a minute into the future, and every
2416− // honest ack that followed — the ones that actually mark our messages
2417− // delivered — would read as stale and be dropped.
2418− if let Some(c) = model.contacts.get_mut(sender) {
2419− c.last_ack_ts = pending.ts.min(now);
2420− contacts_dirty = true;
2421− }
2422− if let Some(convo) = model.messages.get_mut(sender) {
2423− let mut changed = false;
2424− for m in convo.iter_mut() {
2425− if m.outgoing && pending.ids.contains(&m.id) && !m.delivered {
2426− m.delivered = true;
2427− changed = true;
2428− }
2429− }
2430− if changed {
2431− touched_convos.insert(sender.clone());
2432− }
2433− }
2434− }
2435−
24362086 // Queue delivery acks, in batches the recipient will actually accept.
24372087 //
24382088 // `parse_payload` drops a `delivery.ack` carrying more than `MAX_ACK_IDS`
@@ -2455,7 +2105,6 @@ impl Skrepka {
24552105 Arc::new(protocol::serialize_payload(&payload, ack_ts)),
24562106 ));
24572107 }
2458− outbox_dirty = true;
24592108 }
24602109
24612110 // Persist everything touched.
@@ -2463,16 +2112,10 @@ impl Skrepka {
24632112 for peer in &touched_convos {
24642113 cmds.push(self.persist_messages(model, peer));
24652114 }
2466− // Q18: Persist the dedup ID set whenever we touched conversations.
2467− if !touched_convos.is_empty() {
2468− cmds.push(self.persist_seen_ids(model));
2469− }
24702115 if contacts_dirty {
24712116 cmds.push(self.persist_contacts(model));
24722117 }
2473− if outbox_dirty {
2474− cmds.push(self.persist_outbox(model));
2475− }
2118+ cmds.push(self.persist_outbox(model));
24762119 cmds.push(Command::event(Event::StartFlush));
24772120 Command::all(cmds)
24782121 }
@@ -2545,7 +2188,7 @@ mod tests {
25452188 use crate::model::{MAX_OUTBOX_RETRIES, OUTBOX_TTL_MS};
25462189
25472190 fn with_identity() -> Model {
2548− let id = Identity::from_seed(&[3u8; 32]).unwrap();
2191+ let id = Identity::from_seed(&[3u8; 32]);
25492192 Model {
25502193 secret_key: Some(Zeroizing::new(id.secret_key.to_vec())),
25512194 my_pubkey: id.public_key_hex(),
@@ -2554,7 +2197,7 @@ mod tests {
25542197 }
25552198
25562199 fn peer_hex(seed: u8) -> String {
2557− Identity::from_seed(&[seed; 32]).unwrap().public_key_hex()
2200+ Identity::from_seed(&[seed; 32]).public_key_hex()
25582201 }
25592202
25602203 /// `Command::event` *queues* an event; it does not run it. The runtime drains
@@ -2577,7 +2220,7 @@ mod tests {
25772220
25782221 /// A poll page carrying one payload, encrypted from `sender_seed` to `me`.
25792222 fn page_from(sender_seed: u8, me: &Model, payload: &Payload, ts: i64) -> PollResp {
2580− let sender = Identity::from_seed(&[sender_seed; 32]).unwrap();
2223+ let sender = Identity::from_seed(&[sender_seed; 32]);
25812224 let recipient = Identity::from_secret_bytes(me.secret_key.as_ref().unwrap()).unwrap();
25822225 let json = protocol::serialize_payload(payload, ts);
25832226 let blob = crate::crypto::encrypt(
@@ -2666,8 +2309,7 @@ mod tests {
26662309 /// One full send round-trip: flush the outbox, then answer with `status`.
26672310 fn send_and_answer(app: &Skrepka, m: &mut Model, status: u16) {
26682311 let _ = app.update(Event::StartFlush, m);
2669− let gen = m.flush_gen;
2670− let _ = app.update(Event::SendResult(gen, Ok(ok_response(status))), m);
2312+ let _ = app.update(Event::SendResult(Ok(ok_response(status))), m);
26712313 }
26722314
26732315 // -----------------------------------------------------------------------
@@ -2696,7 +2338,7 @@ mod tests {
26962338 // The last one exhausts the budget: the head goes, and what was stuck
26972339 // behind it goes out.
26982340 let _ = app.update(Event::StartFlush, &mut m);
2699− let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(503))), &mut m);
2341+ let mut cmd = app.update(Event::SendResult(Ok(ok_response(503))), &mut m);
27002342
27012343 assert_eq!(m.outbox.len(), 1, "the undeliverable item is dropped");
27022344 assert_eq!(m.outbox[0].recipient, behind, "and the queue moves on");
@@ -2815,7 +2457,7 @@ mod tests {
28152457
28162458 let _ = app.update(Event::StartFlush, &mut m);
28172459 let mut cmd = app.update(
2818− Event::SendResult(m.flush_gen, Ok(response_with_header(429, "retry-after", "2"))),
2460+ Event::SendResult(Ok(response_with_header(429, "retry-after", "2"))),
28192461 &mut m,
28202462 );
28212463
@@ -2846,7 +2488,7 @@ mod tests {
28462488 m.poll_retries = 0;
28472489
28482490 let _ = app.update(Event::StartFlush, &mut m);
2849− let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(429))), &mut m);
2491+ let mut cmd = app.update(Event::SendResult(Ok(ok_response(429))), &mut m);
28502492
28512493 assert_eq!(m.outbox[0].retries, 0, "still not the message's fault");
28522494 assert_eq!(
@@ -2859,7 +2501,7 @@ mod tests {
28592501 // A second 429 climbs, rather than restarting at 3 s.
28602502 m.flush_paused_until = 0;
28612503 let _ = app.update(Event::StartFlush, &mut m);
2862− let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(429))), &mut m);
2504+ let mut cmd = app.update(Event::SendResult(Ok(ok_response(429))), &mut m);
28632505 assert_eq!(scheduled_delays(&mut cmd), vec![backoff_ms(1)], "it climbs");
28642506 assert_eq!(m.send_retries, 2);
28652507 }
@@ -2877,7 +2519,7 @@ mod tests {
28772519
28782520 let _ = app.update(Event::StartFlush, &mut m);
28792521 let _ = app.update(
2880− Event::SendResult(m.flush_gen, Ok(response_with_header(429, "retry-after", "2"))),
2522+ Event::SendResult(Ok(response_with_header(429, "retry-after", "2"))),
28812523 &mut m,
28822524 );
28832525 assert!(
@@ -2933,7 +2575,7 @@ mod tests {
29332575
29342576 let _ = app.update(Event::StartFlush, &mut m);
29352577 let mut cmd = app.update(
2936− Event::SendResult(m.flush_gen, Ok(response_with_header(429, "retry-after", "999999999"))),
2578+ Event::SendResult(Ok(response_with_header(429, "retry-after", "999999999"))),
29372579 &mut m,
29382580 );
29392581
@@ -2968,7 +2610,7 @@ mod tests {
29682610 assert!(batches[0].iter().all(|e| e.to == peer));
29692611
29702612 // ...and its success clears the whole batch, not just the head.
2971− let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
2613+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
29722614 assert!(m.outbox.is_empty());
29732615 }
29742616
@@ -2993,7 +2635,7 @@ mod tests {
29932635 assert_eq!(batches[0].len(), 2, "only the run at the head");
29942636 assert!(batches[0].iter().all(|e| e.to == a));
29952637
2996− let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
2638+ let mut cmd = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
29972639 assert_eq!(m.outbox.len(), 2);
29982640 let batches = sent_batches(&mut cmd);
29992641 assert_eq!(batches[0].len(), 1, "then b's single message");
@@ -3016,7 +2658,7 @@ mod tests {
30162658 let batches = sent_batches(&mut cmd);
30172659 assert_eq!(batches[0].len(), MAX_SEND_BATCH);
30182660
3019− let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
2661+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
30202662 assert_eq!(m.outbox.len(), 20, "the rest stays queued for the next batch");
30212663 }
30222664
@@ -3078,7 +2720,7 @@ mod tests {
30782720 let sent_bytes: usize = first[0].iter().map(|e| e.encrypted_blob.len()).sum();
30792721
30802722 // ...and the relay says that was too big.
3081− let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(413))), &mut m);
2723+ let mut cmd = app.update(Event::SendResult(Ok(ok_response(413))), &mut m);
30822724
30832725 assert_eq!(
30842726 m.send_batch_budget,
@@ -3099,7 +2741,7 @@ mod tests {
30992741 assert!(shrunk > 0, "but not an empty one");
31002742
31012743 // The relay takes the smaller batch, and exactly it is popped.
3102− let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
2744+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
31032745 assert_eq!(m.outbox.len(), 5 - shrunk, "exactly the batch that landed is gone");
31042746 assert_eq!(
31052747 m.send_batch_budget,
@@ -3204,7 +2846,7 @@ mod tests {
32042846 let parsed = protocol::parse_payload(item.envelope_json.as_bytes()).unwrap();
32052847 match parsed.payload {
32062848 Payload::Profile { display_name, .. } => {
3207− assert_eq!(display_name, Some("Three".to_string()), "and it is the latest one");
2849+ assert_eq!(display_name, "Three", "and it is the latest one");
32082850 }
32092851 _ => panic!("expected a profile"),
32102852 }
@@ -3247,12 +2889,12 @@ mod tests {
32472889 );
32482890 assert_eq!(m.outbox.len(), 2, "the in-flight item is left alone");
32492891
3250− let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
2892+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
32512893 assert_eq!(m.outbox.len(), 1, "the sent one is popped");
32522894 let parsed = protocol::parse_payload(m.outbox[0].envelope_json.as_bytes()).unwrap();
32532895 match parsed.payload {
32542896 Payload::Profile { display_name, .. } => assert_eq!(
3255− display_name, Some("New".to_string()),
2897+ display_name, "New",
32562898 "and the newer profile survives to be sent"
32572899 ),
32582900 _ => panic!("expected a profile"),
@@ -3293,7 +2935,7 @@ mod tests {
32932935 assert_eq!(m.outbox.len(), 1);
32942936
32952937 // The send lands. It must pop nothing.
3296− let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
2938+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
32972939 assert_eq!(m.outbox.len(), 1, "the friend's message is untouched");
32982940 assert_eq!(m.outbox[0].recipient, friend);
32992941 }
@@ -3434,8 +3076,8 @@ mod tests {
34343076 .insert(peer.clone(), Contact::new(peer.clone(), String::new(), 0));
34353077
34363078 let profile = |name: &str| Payload::Profile {
3437− display_name: Some(name.to_string()),
3438− bio: Some(String::new()),
3079+ display_name: name.into(),
3080+ bio: String::new(),
34393081 photo: None,
34403082 };
34413083 let page = page_from(9, &m, &profile("Far"), i64::MAX);
@@ -3532,11 +3174,8 @@ mod tests {
35323174 let (m, peer) = ingest_one_text_from_an_unblocked_peer();
35333175 assert_eq!(m.messages[&peer].len(), 1);
35343176 assert_eq!(m.messages[&peer][0].body, "hi");
3535− // D6-app: A profile broadcast is queued alongside the delivery.ack
3536− // so the new contact sees our display name and avatar.
3537− assert_eq!(m.outbox.len(), 2, "one delivery.ack + one profile queued");
3177+ assert_eq!(m.outbox.len(), 1, "one delivery.ack is queued");
35383178 assert_eq!(m.outbox[0].recipient, peer);
3539− assert_eq!(m.outbox[1].recipient, peer);
35403179 }
35413180
35423181 /// A peer sending `ts = i64::MAX` used to pin itself to the top of the
@@ -3973,7 +3612,7 @@ mod tests {
39733612 assert!(m.flushing);
39743613
39753614 let resp = crux_http::testing::ResponseBuilder::ok().body(Vec::new()).build();
3976− let _ = app.update(Event::SendResult(m.flush_gen, Ok(resp)), &mut m);
3615+ let _ = app.update(Event::SendResult(Ok(resp)), &mut m);
39773616
39783617 assert!(!m.flushing, "flushing must be cleared on the success path");
39793618 assert!(m.outbox.is_empty(), "the sent item is dropped from the outbox");
@@ -3995,7 +3634,7 @@ mod tests {
39953634
39963635 // Its success must both drop the item and let the *next* one go out.
39973636 let resp = crux_http::testing::ResponseBuilder::ok().body(Vec::new()).build();
3998− let _ = app.update(Event::SendResult(m.flush_gen, Ok(resp)), &mut m);
3637+ let _ = app.update(Event::SendResult(Ok(resp)), &mut m);
39993638 assert_eq!(m.outbox.len(), 1);
40003639 assert!(m.flushing, "the second item is now in flight");
40013640 }
@@ -4138,7 +3777,6 @@ mod tests {
41383777 Event::LoadedProfile(Ok(None)),
41393778 Event::LoadedContacts(Ok(None)),
41403779 Event::LoadedCursor(Ok(None)),
4141− Event::LoadedSeenIds(Ok(None)),
41423780 ]
41433781 .into_iter()
41443782 .enumerate()
@@ -4152,7 +3790,7 @@ mod tests {
41523790 assert_eq!(connects(&mut cmd), 0, "and none of them connects");
41533791 }
41543792
4155− // The sixth — and only the sixth — emits Connect.
3793+ // The fifth — and only the fifth — emits Connect.
41563794 let mut cmd = app.update(Event::LoadedOutbox(Ok(None)), &mut m);
41573795 assert_eq!(m.loads_pending, 0);
41583796 assert_eq!(connects(&mut cmd), 1, "the last load connects, exactly once");
@@ -4242,7 +3880,7 @@ mod tests {
42423880 );
42433881
42443882 // The recovery is a relaunch, which re-runs the loads and clears the latch.
4245− let sk = Identity::from_seed(&[3u8; 32]).unwrap().secret_key.to_vec();
3883+ let sk = Identity::from_seed(&[3u8; 32]).secret_key.to_vec();
42463884 let _ = app.update(Event::IdentityLoaded(sk), &mut m);
42473885 assert!(
42483886 !m.kv_load_failed,
@@ -4349,8 +3987,8 @@ mod tests {
43493987 );
43503988
43513989 let profile = Payload::Profile {
4352− display_name: Some("Mallory".into()),
4353− bio: Some("spam".into()),
3990+ display_name: "Mallory".into(),
3991+ bio: "spam".into(),
43543992 photo: Some("aGk=".into()),
43553993 };
43563994 let page = page_from(9, &m, &profile, now_ms());
@@ -4431,8 +4069,8 @@ mod tests {
44314069 let peer = peer_hex(9);
44324070
44334071 let profile = Payload::Profile {
4434− display_name: Some("Mallory".into()),
4435− bio: Some(String::new()),
4072+ display_name: "Mallory".into(),
4073+ bio: String::new(),
44364074 photo: Some("aGk=".into()),
44374075 };
44384076 let page = page_from(9, &m, &profile, now_ms());
@@ -4644,7 +4282,7 @@ mod tests {
46444282
46454283 // One page holding more messages from one sender than fits in a single ack.
46464284 let count = protocol::MAX_ACK_IDS + 5;
4647− let sender = Identity::from_seed(&[9u8; 32]).unwrap();
4285+ let sender = Identity::from_seed(&[9u8; 32]);
46484286 let me = Identity::from_secret_bytes(m.secret_key.as_ref().unwrap()).unwrap();
46494287 let events = (0..count)
46504288 .map(|i| {
@@ -4964,7 +4602,7 @@ mod tests {
49644602
49654603 // The send never comes back. The watchdog fires, clears the stuck guard,
49664604 // and retries the flush — the message goes out again.
4967− let mut cmd = app.update(Event::FlushWatchdog(m.flush_gen), &mut m);
4605+ let mut cmd = app.update(Event::FlushWatchdog, &mut m);
49684606
49694607 assert!(m.flushing, "a fresh send is in flight (the retry)");
49704608 assert!(m.in_flight.is_some(), "in_flight is set for the retry");
@@ -4981,23 +4619,22 @@ mod tests {
49814619 m.outbox.push_back(queued(&peer_hex(9), "hi"));
49824620
49834621 let _ = app.update(Event::StartFlush, &mut m);
4984− let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
4622+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
49854623 assert!(!m.flushing);
49864624 assert!(m.outbox.is_empty());
49874625
49884626 // The watchdog for the completed send fires late.
4989− let _ = app.update(Event::FlushWatchdog(m.flush_gen), &mut m);
4627+ let _ = app.update(Event::FlushWatchdog, &mut m);
49904628 assert!(!m.flushing, "a completed send is not re-armed");
49914629 assert!(m.outbox.is_empty());
49924630 }
49934631
4994− /// A late `SendResult` from a superseded send is dropped by the generation
4995− /// check. The watchdog abandoned the first send and retried under a new
4996− /// generation; the late result's generation no longer matches, so it cannot
4997− /// pop the retry's items or clear its guard. The retry's own `SendResult`
4998− /// is what eventually pops them.
4632+ /// A late `SendResult` after the watchdog has already retried sees the
4633+ /// retry's `in_flight` and pops its items. This is correct: the retry sent
4634+ /// the same items, and a 200 means they were accepted — whether by the
4635+ /// original request or the retry is irrelevant. The items are delivered.
49994636 #[test]
5000− fn a_late_send_result_after_the_watchdog_is_dropped_by_generation() {
4637+ fn a_late_send_result_after_the_watchdog_pops_the_retrys_items() {
50014638 let app = Skrepka;
50024639 let mut m = with_identity();
50034640 m.token = Some("t".into());
@@ -5005,23 +4642,15 @@ mod tests {
50054642 m.outbox.push_back(queued(&peer, "stuck"));
50064643
50074644 let _ = app.update(Event::StartFlush, &mut m);
5008− let first_gen = m.flush_gen;
5009− // The watchdog fires and retries the flush under a new generation.
5010− let _ = app.update(Event::FlushWatchdog(first_gen), &mut m);
4645+ // The watchdog fires and retries the flush.
4646+ let _ = app.update(Event::FlushWatchdog, &mut m);
50114647 assert_eq!(m.outbox.len(), 1, "the item is still queued (re-sent)");
5012− assert_ne!(m.flush_gen, first_gen, "the generation was bumped");
5013− assert!(m.flushing, "the retry is in flight");
5014−
5015− // The original send's result arrives late. Its generation no longer
5016− // matches, so it is dropped — it cannot pop the retry's items or
5017− // clear the retry's guard.
5018− let _ = app.update(Event::SendResult(first_gen, Ok(ok_response(200))), &mut m);
5019− assert!(m.flushing, "the retry's guard is untouched");
5020− assert_eq!(m.outbox.len(), 1, "the late result popped nothing");
5021−
5022− // The retry's own result arrives and pops the items.
5023− let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
5024− assert!(m.outbox.is_empty(), "the retry's result pops the items");
4648+
4649+ // Now the original send's result arrives late. The retry's in_flight is
4650+ // set, so a 200 pops the item — which is correct, the message was
4651+ // accepted by one of the two sends.
4652+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
4653+ assert!(m.outbox.is_empty(), "the late 200 pops the retry's items — the message was accepted");
50254654 assert!(!m.flushing, "and the guard is released");
50264655 }
50274656}