@@ -9,6 +9,9 @@ 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+
1215use crux_core::{
1316 macros::effect,
1417 render::{render, RenderOperation},
@@ -58,6 +61,24 @@ const MAX_CONTACTS: usize = 500;
5861/// the budget is dropped exactly like an undecryptable blob would be.
5962const MAX_POLL_TOTAL_BYTES: usize = 64 * 1024 * 1024;
6063
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+
6182/// How long a poll may stay in flight before we assume the shell lost it.
6283///
6384/// `polling` gates the whole long-poll loop, and it is only cleared by a
@@ -87,6 +108,12 @@ const POLL_WATCHDOG_MS: u64 = 90_000;
87108/// one costs a redundant request.
88109const FLUSH_WATCHDOG_MS: u64 = 90_000;
89110
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+
90117/// Messages per `/messages` request.
91118///
92119/// The relay rejects a longer batch outright (`413 batch_too_large`,
@@ -120,10 +147,18 @@ const K_PROFILE: &str = "profile";
120147const K_CONTACTS: &str = "contacts";
121148const K_CURSOR: &str = "cursor";
122149const K_OUTBOX: &str = "outbox";
150+const K_SEEN_IDS: &str = "seen_ids";
123151fn k_messages(peer: &str) -> String {
124152 format!("messages:{peer}")
125153}
126154
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+
127162// ---------------------------------------------------------------------------
128163// Effect
129164// ---------------------------------------------------------------------------
@@ -191,6 +226,9 @@ pub enum Event {
191226 LoadedOutbox(#[facet(opaque)] KvData),
192227 #[serde(skip)]
193228 #[facet(skip)]
229+ LoadedSeenIds(#[facet(opaque)] KvData),
230+ #[serde(skip)]
231+ #[facet(skip)]
194232 LoadedMessages(String, #[facet(opaque)] KvData),
195233 #[serde(skip)]
196234 #[facet(skip)]
@@ -224,11 +262,17 @@ pub enum Event {
224262 #[serde(skip)]
225263 #[facet(skip)]
226264 StartFlush,
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`.
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.
229273 #[serde(skip)]
230274 #[facet(skip)]
231− FlushWatchdog,
275+ AuthWatchdog(u64),
232276 #[serde(skip)]
233277 #[facet(skip)]
234278 ChallengeResult(u64, #[facet(opaque)] HttpResult),
@@ -240,7 +284,7 @@ pub enum Event {
240284 PollResult(u64, #[facet(opaque)] HttpResult),
241285 #[serde(skip)]
242286 #[facet(skip)]
243− SendResult(#[facet(opaque)] HttpResult),
287+ SendResult(u64, #[facet(opaque)] HttpResult),
244288}
245289
246290// ---------------------------------------------------------------------------
@@ -266,9 +310,9 @@ pub enum Page {
266310 EditProfile,
267311}
268312
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;
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;
272316
273317/// The batch a live `/messages` request is carrying: the first `count` items of
274318/// the outbox, all bound for `recipient`.
@@ -294,6 +338,12 @@ pub struct Model {
294338 profile: OwnProfile,
295339 contacts: BTreeMap<String, Contact>,
296340 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>>,
297347 outbox: VecDeque<OutboxItem>,
298348 cursor: i64,
299349 token: Option<String>,
@@ -336,6 +386,12 @@ pub struct Model {
336386 /// items it sent off the head of the outbox, so two overlapping sends would
337387 /// pop each other's messages.
338388 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,
339395 /// What the in-flight send will pop off the head of the outbox when it
340396 /// succeeds — see `InFlight`. `None` while nothing is in flight, and also
341397 /// while a send is in flight whose items have since been removed from under
@@ -400,6 +456,7 @@ impl Default for Model {
400456 profile: OwnProfile::default(),
401457 contacts: BTreeMap::new(),
402458 messages: BTreeMap::new(),
459+ seen_ids: BTreeMap::new(),
403460 outbox: VecDeque::new(),
404461 cursor: 0,
405462 token: None,
@@ -411,6 +468,7 @@ impl Default for Model {
411468 authenticating: false,
412469 auth_gen: 0,
413470 flushing: false,
471+ flush_gen: 0,
414472 in_flight: None,
415473 send_batch_budget: None,
416474 loads_pending: 0,
@@ -462,6 +520,24 @@ impl Model {
462520 self.auth_gen
463521 }
464522
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+
465541 /// Abandon the in-flight auth round-trip (either leg).
466542 fn abandon_auth(&mut self) {
467543 self.authenticating = false;
@@ -508,7 +584,7 @@ fn now_ms() -> i64 {
508584 SystemTime::now()
509585 .duration_since(UNIX_EPOCH)
510586 .map(|d| d.as_millis() as i64)
511− .unwrap_or(0)
587+ .unwrap_or(1)
512588}
513589
514590/// Bare lowercased hostname of a server URL (no scheme/port/trailing dot).
@@ -596,6 +672,16 @@ struct PollEvent {
596672 #[serde(rename = "encryptedBlob")]
597673 encrypted_blob: String,
598674}
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+}
599685#[derive(Serialize)]
600686struct SendBatch {
601687 messages: Vec<Envelope>,
@@ -639,6 +725,7 @@ impl App for Skrepka {
639725 KeyValue::get(K_CONTACTS).then_send(Event::LoadedContacts),
640726 KeyValue::get(K_CURSOR).then_send(Event::LoadedCursor),
641727 KeyValue::get(K_OUTBOX).then_send(Event::LoadedOutbox),
728+ KeyValue::get(K_SEEN_IDS).then_send(Event::LoadedSeenIds),
642729 ])
643730 .and(render())
644731 }
@@ -716,10 +803,51 @@ impl App for Skrepka {
716803 }
717804 self.startup_done(model).and(render())
718805 }
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+ }
719818 Event::LoadedMessages(peer, res) => {
720819 match parse_kv::<Vec<StoredMessage>>(res) {
721820 Ok(Some(list)) => {
722− model.messages.insert(peer, 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+ }
723851 }
724852 Ok(None) => {}
725853 Err(()) => {
@@ -829,11 +957,20 @@ impl App for Skrepka {
829957 // would then ack a sequence number B never issued, and B would
830958 // drop every message waiting for us.
831959 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();
832966 // The old relay's rate limit was the old relay's opinion. A new one is
833967 // entitled to take our sends immediately, and carrying the pause over
834968 // would park the outbox against a limit that no longer applies.
835969 model.flush_paused_until = 0;
836970 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;
837974 let writes = if self.refuse_write(model) {
838975 Command::done()
839976 } else {
@@ -881,6 +1018,8 @@ impl App for Skrepka {
8811018 }
8821019 }
8831020 Event::AddContact { input, nickname } => {
1021+ // Clear a stale error from a previous failed add.
1022+ model.error = None;
8841023 // A nickname is stored in the contacts kv blob, which is
8851024 // rewritten in full on every change. An unbounded nickname
8861025 // bloats the blob and every subsequent write.
@@ -917,6 +1056,9 @@ impl App for Skrepka {
9171056 if let Some(c) = model.contacts.get_mut(&peer) {
9181057 c.blocked = blocked;
9191058 }
1059+ // Clear a stale error (e.g. "cannot send to a blocked contact")
1060+ // when the user unblocks — the block that caused it is now lifted.
1061+ model.error = None;
9201062 if blocked {
9211063 // Gating *ingest* on the block only stops the acks we have not
9221064 // queued yet. Anything already sitting in the outbox for this
@@ -935,6 +1077,7 @@ impl App for Skrepka {
9351077 Event::DeleteContact { peer } => {
9361078 model.contacts.remove(&peer);
9371079 model.messages.remove(&peer);
1080+ model.seen_ids.remove(&peer);
9381081 model.outbox.retain(|item| item.recipient != peer);
9391082 // The send in flight may be carrying the very items we just
9401083 // dropped; if so its result must pop nothing.
@@ -951,6 +1094,8 @@ impl App for Skrepka {
9511094 // message the user just asked us to forget. Delete the key.
9521095 KeyValue::delete(k_messages(&peer)).then_send(Event::Saved),
9531096 self.persist_outbox(model),
1097+ // Q18: persist the seen_ids map without the removed peer.
1098+ self.persist_seen_ids(model),
9541099 ])
9551100 .and(render())
9561101 }
@@ -959,6 +1104,8 @@ impl App for Skrepka {
9591104 bio,
9601105 photo,
9611106 } => {
1107+ // Clear a stale error from a previous failed save.
1108+ model.error = None;
9621109 // The recipient's `parse_payload` enforces these caps and
9631110 // silently drops oversized profiles. Without these checks the
9641111 // sender's local profile would be saved, broadcast to every
@@ -976,6 +1123,13 @@ impl App for Skrepka {
9761123 model.error = Some("photo too large".into());
9771124 return render();
9781125 }
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+ }
9791133 model.profile = OwnProfile {
9801134 display_name,
9811135 bio,
@@ -984,8 +1138,8 @@ impl App for Skrepka {
9841138 model.page = Page::Conversations;
9851139 let ts = now_ms();
9861140 let payload = Payload::Profile {
987− display_name: model.profile.display_name.clone(),
988− bio: model.profile.bio.clone(),
1141+ display_name: Some(model.profile.display_name.clone()),
1142+ bio: Some(model.profile.bio.clone()),
9891143 photo: model.profile.photo.clone(),
9901144 };
9911145 // Broadcast the new profile to every contact we haven't blocked.
@@ -1064,6 +1218,11 @@ impl App for Skrepka {
10641218 let gen = model.next_auth_gen();
10651219 req.build()
10661220 .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+ )
10671226 .and(render())
10681227 }
10691228 Err(_) => {
@@ -1242,35 +1401,90 @@ impl App for Skrepka {
12421401 return render();
12431402 }
12441403 let bytes = resp.take_body().unwrap_or_default();
1245− let parsed: PollResp = serde_json::from_slice(&bytes).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;
12461423 let cmd = self.ingest_poll(model, parsed);
12471424 // The re-poll hangs off the *cursor write*, not off this event.
12481425 // Polling again is what acks the batch we just took — the server
12491426 // deletes everything up to `cursor` — so it must not happen until
12501427 // the batch is durable on our side. `SavedCursor` re-polls.
1251− cmd.and(
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 {
12521440 KeyValue::set(K_CURSOR, json_bytes(&model.cursor))
1253− .then_send(Event::SavedCursor),
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),
12541449 )
12551450 .and(render())
12561451 }
12571452
12581453 // ---------------- outbox ----------------
12591454 Event::StartFlush => self.flush_next(model),
1260− Event::FlushWatchdog => {
1261− // The send request never came back — the shell dropped the HTTP
1262− // effect (backgrounded app, bug), and `flushing` has been stuck
1263− // ever since. Release the guard and retry the flush. The late
1264− // `SendResult`, if it ever arrives, finds `in_flight` already
1265− // cleared and pops nothing.
1266− if !model.flushing {
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 {
12671462 return Command::done();
12681463 }
1269− model.flushing = false;
1270− model.in_flight = None;
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();
12711469 self.flush_next(model)
12721470 }
1273− Event::SendResult(Ok(resp)) => {
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+ }
12741488 // The in-flight send is over; `flush_next` refuses to run while
12751489 // this is set, so it must be cleared before *any* branch below.
12761490 model.flushing = false;
@@ -1360,16 +1574,24 @@ impl App for Skrepka {
13601574 // later, under load that has nothing to do with this one — starts
13611575 // at whatever height the last one climbed to.
13621576 model.send_retries = 0;
1363− for _ in 0..batch.map_or(0, |b| b.count) {
1364− model.outbox.pop_front();
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+ }
13651584 }
13661585 return self.persist_outbox(model).and(self.flush_next(model));
13671586 }
13681587 // A transient failure: a 5xx, a rate limit, a relay restarting.
13691588 self.retry_batch(model, batch)
13701589 }
1371− Event::SendResult(Err(_)) => {
1590+ Event::SendResult(gen, Err(_)) => {
13721591 // A transport error is the same transient failure as a 5xx.
1592+ if gen != model.flush_gen {
1593+ return Command::done();
1594+ }
13731595 model.flushing = false;
13741596 let batch = model.in_flight.take();
13751597 self.retry_batch(model, batch)
@@ -1544,6 +1766,16 @@ impl Skrepka {
15441766 KeyValue::set(k_messages(peer), json_bytes(&list)).then_send(Event::Saved)
15451767 }
15461768
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+
15471779 fn schedule_reconnect(&self, model: &mut Model) -> Command<Effect, Event> {
15481780 let delay = backoff_ms(model.poll_retries);
15491781 model.poll_retries = model.poll_retries.saturating_add(1);
@@ -1574,6 +1806,9 @@ impl Skrepka {
15741806 let Some(peer) = model.active_peer.clone() else {
15751807 return render();
15761808 };
1809+ // Clear a stale error from a previous failed send, so the error display
1810+ // does not persist after the user fixes the issue and sends again.
1811+ model.error = None;
15771812 // Blocking cuts the peer off in both directions (PROTOCOL.md §4). The
15781813 // ingest side drops their messages; the send side must refuse too, or
15791814 // a blocked peer still receives our texts, acks, and the liveness
@@ -1639,14 +1874,21 @@ impl Skrepka {
16391874 .insert(peer.clone(), Contact::new(peer.clone(), String::new(), ts));
16401875 }
16411876
1642− // On first contact, share our profile *before* the text so the
1643− // recipient's first impression is a name and avatar, not a truncated
1644− // @p that gets replaced a moment later. Both are queued behind any
1645− // in-flight items, so the ordering is preserved through the outbox.
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.)
16461888 if was_new {
16471889 let profile = Payload::Profile {
1648− display_name: model.profile.display_name.clone(),
1649− bio: model.profile.bio.clone(),
1890+ display_name: Some(model.profile.display_name.clone()),
1891+ bio: Some(model.profile.bio.clone()),
16501892 photo: model.profile.photo.clone(),
16511893 };
16521894 model.outbox.push_back(OutboxItem::profile(
@@ -1655,12 +1897,6 @@ impl Skrepka {
16551897 ));
16561898 }
16571899
1658− let text = Payload::Text { id, body };
1659− model.outbox.push_back(OutboxItem::new(
1660− peer.clone(),
1661− Arc::new(protocol::serialize_payload(&text, ts)),
1662− ));
1663−
16641900 Command::all([
16651901 self.persist_messages(model, &peer),
16661902 self.persist_contacts(model),
@@ -1788,6 +2024,7 @@ impl Skrepka {
17882024 };
17892025
17902026 model.flushing = true;
2027+ let gen = model.next_flush_gen();
17912028 model.in_flight = Some(InFlight {
17922029 recipient,
17932030 count: messages.len(),
@@ -1798,11 +2035,11 @@ impl Skrepka {
17982035 .header("authorization", format!("Bearer {token}"))
17992036 .body_json(&SendBatch { messages })
18002037 {
1801− Ok(req) => persist.and(req.build().then_send(Event::SendResult))
2038+ Ok(req) => persist.and(req.build().then_send(move |r| Event::SendResult(gen, r)))
18022039 .and(
18032040 Time::notify_after(Duration::from_millis(FLUSH_WATCHDOG_MS))
18042041 .0
1805− .then_send(|_| Event::FlushWatchdog),
2042+ .then_send(move |_| Event::FlushWatchdog(gen)),
18062043 ),
18072044 Err(_) => {
18082045 // Not a dead loop: the next StartFlush (from a poll page or a
@@ -1873,8 +2110,20 @@ impl Skrepka {
18732110 return Command::done();
18742111 }
18752112 let prev_cursor = model.cursor;
1876− if page.cursor > model.cursor {
1877− model.cursor = page.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;
18782127 }
18792128 let Some(id) = model.identity() else {
18802129 return Command::done();
@@ -1883,12 +2132,24 @@ impl Skrepka {
18832132 let now = now_ms();
18842133 let mut touched_convos: HashSet<String> = HashSet::new();
18852134 let mut contacts_dirty = false;
2135+ let mut outbox_dirty = false;
18862136 // sender hex -> message ids needing acks
18872137 let mut ack_targets: BTreeMap<String, Vec<String>> = BTreeMap::new();
1888− // Lazily-built per-sender sets of known message IDs for O(1) text dedup.
1889− let mut seen_ids: HashMap<String, HashSet<String>> = HashMap::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();
18902142
18912143 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;
18922153 for ev in page.events {
18932154 // Reject an oversized blob on its hex length, before it costs us a
18942155 // decode. No conforming sender produces one (the relay caps blobs at
@@ -1896,19 +2157,37 @@ impl Skrepka {
18962157 if ev.encrypted_blob.len() > MAX_BLOB_LEN * 2 {
18972158 continue;
18982159 }
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;
18992177 match budget.checked_sub(ev.encrypted_blob.len() / 2) {
19002178 Some(left) => budget = left,
19012179 None => {
1902− // A page this large is a misbehaving relay, not real mail.
1903− // Stop here; the rest of the page is discarded exactly as an
1904− // undecryptable blob would be.
1905− //
1906− // Restore the cursor: it was advanced to `page.cursor` above,
1907− // but we have not processed the remaining events. If we keep
1908− // the advanced cursor, the next poll acks the whole page to
1909− // the relay — including every event we skipped — and the
1910− // relay deletes them. Those messages are permanently lost.
1911− model.cursor = prev_cursor;
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+ }
19122191 model.error = Some("relay sent an oversized poll page".into());
19132192 break;
19142193 }
@@ -1922,6 +2201,10 @@ impl Skrepka {
19222201 let Some(parsed) = protocol::parse_payload(&dec.plaintext) else {
19232202 continue;
19242203 };
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;
19252208 let sender = dec.sender_hex;
19262209
19272210 // Mail from ourselves. A relay cannot forge this — the sender identity
@@ -1973,12 +2256,35 @@ impl Skrepka {
19732256 Contact::new(sender.clone(), String::new(), ts),
19742257 );
19752258 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;
19762271 }
19772272 let convo = model.messages.entry(sender.clone()).or_default();
1978− // Lazily build the per-sender ID set on first touch.
1979− let id_set = seen_ids.entry(sender.clone()).or_insert_with(|| {
1980− convo.iter().map(|m| m.id.clone()).collect()
1981− });
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+ }
19822288 // dedup by id — O(1) via HashSet
19832289 if !id_set.contains(&msg_id) {
19842290 // The conversation is *kept* sorted rather than re-sorted:
@@ -1996,6 +2302,29 @@ impl Skrepka {
19962302 },
19972303 );
19982304 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+ }
19992328 trim_history(convo);
20002329 touched_convos.insert(sender.clone());
20012330 // Only ack a message we actually stored — a duplicate id
@@ -2004,39 +2333,26 @@ impl Skrepka {
20042333 }
20052334 }
20062335 Payload::DeliveryAck { ack_ids } => {
2007− // Staleness check: drop a replayed ack whose ts predates the
2008− // last one we processed for this contact.
2009− let stale = model
2010− .contacts
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
20112344 .get(&sender)
2012− .is_some_and(|c| ts < c.last_ack_ts);
2013− if stale {
2014− continue;
2015− }
2016− // The watermark is clamped to *now*, not merely into the skew
2017− // window: `ts` is the peer's to choose, and `last_ack_ts` is
2018− // what every later ack is measured against. A single ack at
2019− // `ts = i64::MAX` would otherwise park the watermark a minute
2020− // into the future, and every honest ack that followed — the
2021− // ones that actually mark our messages delivered — would read
2022− // as stale and be dropped.
2023− if let Some(c) = model.contacts.get_mut(&sender) {
2024− c.last_ack_ts = ts.min(now);
2025− contacts_dirty = true;
2026− }
2027− let ack_set: HashSet<String> = ack_ids.iter().cloned().collect();
2028− if let Some(convo) = model.messages.get_mut(&sender) {
2029− let mut changed = false;
2030− for m in convo.iter_mut() {
2031− if m.outgoing && ack_set.contains(&m.id) && !m.delivered {
2032− m.delivered = true;
2033− changed = true;
2034− }
2035− }
2036− if changed {
2037− touched_convos.insert(sender);
2038− }
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;
20392350 }
2351+ pending_acks
2352+ .entry(sender.clone())
2353+ .or_default()
2354+ .ids
2355+ .extend(ack_ids);
20402356 }
20412357 Payload::Profile {
20422358 display_name,
@@ -2063,9 +2379,17 @@ impl Skrepka {
20632379 // A replayed profile at the same `ts` overwrites with identical
20642380 // content, which is a no-op in practice.
20652381 if ts >= c.last_profile_ts {
2066− c.display_name = display_name;
2067− c.bio = bio;
2068− c.photo = photo;
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+ }
20692393 c.last_profile_ts = ts.min(now);
20702394 contacts_dirty = true;
20712395 }
@@ -2073,6 +2397,42 @@ impl Skrepka {
20732397 }
20742398 }
20752399
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+
20762436 // Queue delivery acks, in batches the recipient will actually accept.
20772437 //
20782438 // `parse_payload` drops a `delivery.ack` carrying more than `MAX_ACK_IDS`
@@ -2095,6 +2455,7 @@ impl Skrepka {
20952455 Arc::new(protocol::serialize_payload(&payload, ack_ts)),
20962456 ));
20972457 }
2458+ outbox_dirty = true;
20982459 }
20992460
21002461 // Persist everything touched.
@@ -2102,10 +2463,16 @@ impl Skrepka {
21022463 for peer in &touched_convos {
21032464 cmds.push(self.persist_messages(model, peer));
21042465 }
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+ }
21052470 if contacts_dirty {
21062471 cmds.push(self.persist_contacts(model));
21072472 }
2108− cmds.push(self.persist_outbox(model));
2473+ if outbox_dirty {
2474+ cmds.push(self.persist_outbox(model));
2475+ }
21092476 cmds.push(Command::event(Event::StartFlush));
21102477 Command::all(cmds)
21112478 }
@@ -2178,7 +2545,7 @@ mod tests {
21782545 use crate::model::{MAX_OUTBOX_RETRIES, OUTBOX_TTL_MS};
21792546
21802547 fn with_identity() -> Model {
2181− let id = Identity::from_seed(&[3u8; 32]);
2548+ let id = Identity::from_seed(&[3u8; 32]).unwrap();
21822549 Model {
21832550 secret_key: Some(Zeroizing::new(id.secret_key.to_vec())),
21842551 my_pubkey: id.public_key_hex(),
@@ -2187,7 +2554,7 @@ mod tests {
21872554 }
21882555
21892556 fn peer_hex(seed: u8) -> String {
2190− Identity::from_seed(&[seed; 32]).public_key_hex()
2557+ Identity::from_seed(&[seed; 32]).unwrap().public_key_hex()
21912558 }
21922559
21932560 /// `Command::event` *queues* an event; it does not run it. The runtime drains
@@ -2210,7 +2577,7 @@ mod tests {
22102577
22112578 /// A poll page carrying one payload, encrypted from `sender_seed` to `me`.
22122579 fn page_from(sender_seed: u8, me: &Model, payload: &Payload, ts: i64) -> PollResp {
2213− let sender = Identity::from_seed(&[sender_seed; 32]);
2580+ let sender = Identity::from_seed(&[sender_seed; 32]).unwrap();
22142581 let recipient = Identity::from_secret_bytes(me.secret_key.as_ref().unwrap()).unwrap();
22152582 let json = protocol::serialize_payload(payload, ts);
22162583 let blob = crate::crypto::encrypt(
@@ -2299,7 +2666,8 @@ mod tests {
22992666 /// One full send round-trip: flush the outbox, then answer with `status`.
23002667 fn send_and_answer(app: &Skrepka, m: &mut Model, status: u16) {
23012668 let _ = app.update(Event::StartFlush, m);
2302− let _ = app.update(Event::SendResult(Ok(ok_response(status))), m);
2669+ let gen = m.flush_gen;
2670+ let _ = app.update(Event::SendResult(gen, Ok(ok_response(status))), m);
23032671 }
23042672
23052673 // -----------------------------------------------------------------------
@@ -2328,7 +2696,7 @@ mod tests {
23282696 // The last one exhausts the budget: the head goes, and what was stuck
23292697 // behind it goes out.
23302698 let _ = app.update(Event::StartFlush, &mut m);
2331− let mut cmd = app.update(Event::SendResult(Ok(ok_response(503))), &mut m);
2699+ let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(503))), &mut m);
23322700
23332701 assert_eq!(m.outbox.len(), 1, "the undeliverable item is dropped");
23342702 assert_eq!(m.outbox[0].recipient, behind, "and the queue moves on");
@@ -2447,7 +2815,7 @@ mod tests {
24472815
24482816 let _ = app.update(Event::StartFlush, &mut m);
24492817 let mut cmd = app.update(
2450− Event::SendResult(Ok(response_with_header(429, "retry-after", "2"))),
2818+ Event::SendResult(m.flush_gen, Ok(response_with_header(429, "retry-after", "2"))),
24512819 &mut m,
24522820 );
24532821
@@ -2478,7 +2846,7 @@ mod tests {
24782846 m.poll_retries = 0;
24792847
24802848 let _ = app.update(Event::StartFlush, &mut m);
2481− let mut cmd = app.update(Event::SendResult(Ok(ok_response(429))), &mut m);
2849+ let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(429))), &mut m);
24822850
24832851 assert_eq!(m.outbox[0].retries, 0, "still not the message's fault");
24842852 assert_eq!(
@@ -2491,7 +2859,7 @@ mod tests {
24912859 // A second 429 climbs, rather than restarting at 3 s.
24922860 m.flush_paused_until = 0;
24932861 let _ = app.update(Event::StartFlush, &mut m);
2494− let mut cmd = app.update(Event::SendResult(Ok(ok_response(429))), &mut m);
2862+ let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(429))), &mut m);
24952863 assert_eq!(scheduled_delays(&mut cmd), vec![backoff_ms(1)], "it climbs");
24962864 assert_eq!(m.send_retries, 2);
24972865 }
@@ -2509,7 +2877,7 @@ mod tests {
25092877
25102878 let _ = app.update(Event::StartFlush, &mut m);
25112879 let _ = app.update(
2512− Event::SendResult(Ok(response_with_header(429, "retry-after", "2"))),
2880+ Event::SendResult(m.flush_gen, Ok(response_with_header(429, "retry-after", "2"))),
25132881 &mut m,
25142882 );
25152883 assert!(
@@ -2565,7 +2933,7 @@ mod tests {
25652933
25662934 let _ = app.update(Event::StartFlush, &mut m);
25672935 let mut cmd = app.update(
2568− Event::SendResult(Ok(response_with_header(429, "retry-after", "999999999"))),
2936+ Event::SendResult(m.flush_gen, Ok(response_with_header(429, "retry-after", "999999999"))),
25692937 &mut m,
25702938 );
25712939
@@ -2600,7 +2968,7 @@ mod tests {
26002968 assert!(batches[0].iter().all(|e| e.to == peer));
26012969
26022970 // ...and its success clears the whole batch, not just the head.
2603− let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
2971+ let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
26042972 assert!(m.outbox.is_empty());
26052973 }
26062974
@@ -2625,7 +2993,7 @@ mod tests {
26252993 assert_eq!(batches[0].len(), 2, "only the run at the head");
26262994 assert!(batches[0].iter().all(|e| e.to == a));
26272995
2628− let mut cmd = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
2996+ let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
26292997 assert_eq!(m.outbox.len(), 2);
26302998 let batches = sent_batches(&mut cmd);
26312999 assert_eq!(batches[0].len(), 1, "then b's single message");
@@ -2648,7 +3016,7 @@ mod tests {
26483016 let batches = sent_batches(&mut cmd);
26493017 assert_eq!(batches[0].len(), MAX_SEND_BATCH);
26503018
2651− let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
3019+ let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
26523020 assert_eq!(m.outbox.len(), 20, "the rest stays queued for the next batch");
26533021 }
26543022
@@ -2710,7 +3078,7 @@ mod tests {
27103078 let sent_bytes: usize = first[0].iter().map(|e| e.encrypted_blob.len()).sum();
27113079
27123080 // ...and the relay says that was too big.
2713− let mut cmd = app.update(Event::SendResult(Ok(ok_response(413))), &mut m);
3081+ let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(413))), &mut m);
27143082
27153083 assert_eq!(
27163084 m.send_batch_budget,
@@ -2731,7 +3099,7 @@ mod tests {
27313099 assert!(shrunk > 0, "but not an empty one");
27323100
27333101 // The relay takes the smaller batch, and exactly it is popped.
2734− let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
3102+ let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
27353103 assert_eq!(m.outbox.len(), 5 - shrunk, "exactly the batch that landed is gone");
27363104 assert_eq!(
27373105 m.send_batch_budget,
@@ -2836,7 +3204,7 @@ mod tests {
28363204 let parsed = protocol::parse_payload(item.envelope_json.as_bytes()).unwrap();
28373205 match parsed.payload {
28383206 Payload::Profile { display_name, .. } => {
2839− assert_eq!(display_name, "Three", "and it is the latest one");
3207+ assert_eq!(display_name, Some("Three".to_string()), "and it is the latest one");
28403208 }
28413209 _ => panic!("expected a profile"),
28423210 }
@@ -2879,12 +3247,12 @@ mod tests {
28793247 );
28803248 assert_eq!(m.outbox.len(), 2, "the in-flight item is left alone");
28813249
2882− let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
3250+ let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
28833251 assert_eq!(m.outbox.len(), 1, "the sent one is popped");
28843252 let parsed = protocol::parse_payload(m.outbox[0].envelope_json.as_bytes()).unwrap();
28853253 match parsed.payload {
28863254 Payload::Profile { display_name, .. } => assert_eq!(
2887− display_name, "New",
3255+ display_name, Some("New".to_string()),
28883256 "and the newer profile survives to be sent"
28893257 ),
28903258 _ => panic!("expected a profile"),
@@ -2925,7 +3293,7 @@ mod tests {
29253293 assert_eq!(m.outbox.len(), 1);
29263294
29273295 // The send lands. It must pop nothing.
2928− let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
3296+ let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
29293297 assert_eq!(m.outbox.len(), 1, "the friend's message is untouched");
29303298 assert_eq!(m.outbox[0].recipient, friend);
29313299 }
@@ -3066,8 +3434,8 @@ mod tests {
30663434 .insert(peer.clone(), Contact::new(peer.clone(), String::new(), 0));
30673435
30683436 let profile = |name: &str| Payload::Profile {
3069− display_name: name.into(),
3070− bio: String::new(),
3437+ display_name: Some(name.to_string()),
3438+ bio: Some(String::new()),
30713439 photo: None,
30723440 };
30733441 let page = page_from(9, &m, &profile("Far"), i64::MAX);
@@ -3164,8 +3532,11 @@ mod tests {
31643532 let (m, peer) = ingest_one_text_from_an_unblocked_peer();
31653533 assert_eq!(m.messages[&peer].len(), 1);
31663534 assert_eq!(m.messages[&peer][0].body, "hi");
3167− assert_eq!(m.outbox.len(), 1, "one delivery.ack is queued");
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");
31683538 assert_eq!(m.outbox[0].recipient, peer);
3539+ assert_eq!(m.outbox[1].recipient, peer);
31693540 }
31703541
31713542 /// A peer sending `ts = i64::MAX` used to pin itself to the top of the
@@ -3346,6 +3717,28 @@ mod tests {
33463717 assert!(m.error.is_some());
33473718 }
33483719
3720+ /// A stale error from a previous failed add is cleared on the next attempt,
3721+ /// so the AddContact form does not show an old error after the user fixes
3722+ /// the input and tries again.
3723+ #[test]
3724+ fn add_contact_clears_a_stale_error_on_success() {
3725+ let app = Skrepka;
3726+ let mut m = with_identity();
3727+ m.error = Some("invalid public key".into());
3728+
3729+ let peer = peer_hex(9);
3730+ let _ = app.update(
3731+ Event::AddContact {
3732+ input: peer.clone(),
3733+ nickname: "Bob".into(),
3734+ },
3735+ &mut m,
3736+ );
3737+
3738+ assert!(m.error.is_none(), "a successful add clears the stale error");
3739+ assert!(m.contacts.contains_key(&peer));
3740+ }
3741+
33493742 #[test]
33503743 fn send_text_enqueues_message_profile_and_outbox() {
33513744 let app = Skrepka;
@@ -3437,6 +3830,30 @@ mod tests {
34373830 assert!(vm.contacts.iter().any(|c| c.pubkey == peer && c.blocked));
34383831 }
34393832
3833+ /// Unblocking clears a stale "cannot send to a blocked contact" error, so the
3834+ /// chat error display does not persist after the user lifts the block.
3835+ #[test]
3836+ fn unblocking_clears_the_stale_send_error() {
3837+ let app = Skrepka;
3838+ let mut m = with_identity();
3839+ let peer = peer_hex(9);
3840+ let mut contact = Contact::new(peer.clone(), "Bob".into(), 0);
3841+ contact.blocked = true;
3842+ m.contacts.insert(peer.clone(), contact);
3843+ m.error = Some("cannot send to a blocked contact".into());
3844+
3845+ let _ = app.update(
3846+ Event::SetBlocked {
3847+ peer: peer.clone(),
3848+ blocked: false,
3849+ },
3850+ &mut m,
3851+ );
3852+
3853+ assert!(!m.contacts[&peer].blocked);
3854+ assert!(m.error.is_none(), "unblocking clears the stale error");
3855+ }
3856+
34403857 /// Gating ingest stops the acks we have not queued yet — but the ack for the
34413858 /// message that made the user hit "block" is already in the outbox, and an ack
34423859 /// is exactly the "we are online and reading you" signal the block exists to
@@ -3556,7 +3973,7 @@ mod tests {
35563973 assert!(m.flushing);
35573974
35583975 let resp = crux_http::testing::ResponseBuilder::ok().body(Vec::new()).build();
3559− let _ = app.update(Event::SendResult(Ok(resp)), &mut m);
3976+ let _ = app.update(Event::SendResult(m.flush_gen, Ok(resp)), &mut m);
35603977
35613978 assert!(!m.flushing, "flushing must be cleared on the success path");
35623979 assert!(m.outbox.is_empty(), "the sent item is dropped from the outbox");
@@ -3578,7 +3995,7 @@ mod tests {
35783995
35793996 // Its success must both drop the item and let the *next* one go out.
35803997 let resp = crux_http::testing::ResponseBuilder::ok().body(Vec::new()).build();
3581− let _ = app.update(Event::SendResult(Ok(resp)), &mut m);
3998+ let _ = app.update(Event::SendResult(m.flush_gen, Ok(resp)), &mut m);
35823999 assert_eq!(m.outbox.len(), 1);
35834000 assert!(m.flushing, "the second item is now in flight");
35844001 }
@@ -3721,6 +4138,7 @@ mod tests {
37214138 Event::LoadedProfile(Ok(None)),
37224139 Event::LoadedContacts(Ok(None)),
37234140 Event::LoadedCursor(Ok(None)),
4141+ Event::LoadedSeenIds(Ok(None)),
37244142 ]
37254143 .into_iter()
37264144 .enumerate()
@@ -3734,7 +4152,7 @@ mod tests {
37344152 assert_eq!(connects(&mut cmd), 0, "and none of them connects");
37354153 }
37364154
3737− // The fifth — and only the fifth — emits Connect.
4155+ // The sixth — and only the sixth — emits Connect.
37384156 let mut cmd = app.update(Event::LoadedOutbox(Ok(None)), &mut m);
37394157 assert_eq!(m.loads_pending, 0);
37404158 assert_eq!(connects(&mut cmd), 1, "the last load connects, exactly once");
@@ -3824,7 +4242,7 @@ mod tests {
38244242 );
38254243
38264244 // The recovery is a relaunch, which re-runs the loads and clears the latch.
3827− let sk = Identity::from_seed(&[3u8; 32]).secret_key.to_vec();
4245+ let sk = Identity::from_seed(&[3u8; 32]).unwrap().secret_key.to_vec();
38284246 let _ = app.update(Event::IdentityLoaded(sk), &mut m);
38294247 assert!(
38304248 !m.kv_load_failed,
@@ -3931,8 +4349,8 @@ mod tests {
39314349 );
39324350
39334351 let profile = Payload::Profile {
3934− display_name: "Mallory".into(),
3935− bio: "spam".into(),
4352+ display_name: Some("Mallory".into()),
4353+ bio: Some("spam".into()),
39364354 photo: Some("aGk=".into()),
39374355 };
39384356 let page = page_from(9, &m, &profile, now_ms());
@@ -4013,8 +4431,8 @@ mod tests {
40134431 let peer = peer_hex(9);
40144432
40154433 let profile = Payload::Profile {
4016− display_name: "Mallory".into(),
4017− bio: String::new(),
4434+ display_name: Some("Mallory".into()),
4435+ bio: Some(String::new()),
40184436 photo: Some("aGk=".into()),
40194437 };
40204438 let page = page_from(9, &m, &profile, now_ms());
@@ -4226,7 +4644,7 @@ mod tests {
42264644
42274645 // One page holding more messages from one sender than fits in a single ack.
42284646 let count = protocol::MAX_ACK_IDS + 5;
4229− let sender = Identity::from_seed(&[9u8; 32]);
4647+ let sender = Identity::from_seed(&[9u8; 32]).unwrap();
42304648 let me = Identity::from_secret_bytes(m.secret_key.as_ref().unwrap()).unwrap();
42314649 let events = (0..count)
42324650 .map(|i| {
@@ -4435,6 +4853,25 @@ mod tests {
44354853 assert!(m.error.as_ref().is_some_and(|e| e.contains("too long")));
44364854 }
44374855
4856+ /// A stale error from a previous failed send must be cleared on the next
4857+ /// attempt, so the error display does not persist after the user fixes the
4858+ /// issue and sends again successfully.
4859+ #[test]
4860+ fn send_text_clears_a_stale_error_on_success() {
4861+ let app = Skrepka;
4862+ let mut m = with_identity();
4863+ let peer = peer_hex(9);
4864+ m.active_peer = Some(peer.clone());
4865+ // Simulate a prior failure that left an error behind.
4866+ m.error = Some("message too long".into());
4867+
4868+ m.compose = "hello".into();
4869+ let _ = app.update(Event::SendText, &mut m);
4870+
4871+ assert!(m.error.is_none(), "a successful send clears the stale error");
4872+ assert!(!m.outbox.is_empty(), "the message was queued");
4873+ }
4874+
44384875 // -----------------------------------------------------------------------
44394876 // SaveProfile field length checks
44404877 // -----------------------------------------------------------------------
@@ -4527,7 +4964,7 @@ mod tests {
45274964
45284965 // The send never comes back. The watchdog fires, clears the stuck guard,
45294966 // and retries the flush — the message goes out again.
4530− let mut cmd = app.update(Event::FlushWatchdog, &mut m);
4967+ let mut cmd = app.update(Event::FlushWatchdog(m.flush_gen), &mut m);
45314968
45324969 assert!(m.flushing, "a fresh send is in flight (the retry)");
45334970 assert!(m.in_flight.is_some(), "in_flight is set for the retry");
@@ -4544,22 +4981,23 @@ mod tests {
45444981 m.outbox.push_back(queued(&peer_hex(9), "hi"));
45454982
45464983 let _ = app.update(Event::StartFlush, &mut m);
4547− let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
4984+ let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
45484985 assert!(!m.flushing);
45494986 assert!(m.outbox.is_empty());
45504987
45514988 // The watchdog for the completed send fires late.
4552− let _ = app.update(Event::FlushWatchdog, &mut m);
4989+ let _ = app.update(Event::FlushWatchdog(m.flush_gen), &mut m);
45534990 assert!(!m.flushing, "a completed send is not re-armed");
45544991 assert!(m.outbox.is_empty());
45554992 }
45564993
4557− /// A late `SendResult` after the watchdog has already retried sees the
4558− /// retry's `in_flight` and pops its items. This is correct: the retry sent
4559− /// the same items, and a 200 means they were accepted — whether by the
4560− /// original request or the retry is irrelevant. The items are delivered.
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.
45614999 #[test]
4562− fn a_late_send_result_after_the_watchdog_pops_the_retrys_items() {
5000+ fn a_late_send_result_after_the_watchdog_is_dropped_by_generation() {
45635001 let app = Skrepka;
45645002 let mut m = with_identity();
45655003 m.token = Some("t".into());
@@ -4567,15 +5005,23 @@ mod tests {
45675005 m.outbox.push_back(queued(&peer, "stuck"));
45685006
45695007 let _ = app.update(Event::StartFlush, &mut m);
4570− // The watchdog fires and retries the flush.
4571− let _ = app.update(Event::FlushWatchdog, &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);
45725011 assert_eq!(m.outbox.len(), 1, "the item is still queued (re-sent)");
4573−
4574− // Now the original send's result arrives late. The retry's in_flight is
4575− // set, so a 200 pops the item — which is correct, the message was
4576− // accepted by one of the two sends.
4577− let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
4578− assert!(m.outbox.is_empty(), "the late 200 pops the retry's items — the message was accepted");
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");
45795025 assert!(!m.flushing, "and the guard is released");
45805026 }
45815027}