@@ -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
@@ -73,6 +94,26 @@ const MAX_POLL_TOTAL_BYTES: usize = 64 * 1024 * 1024;
7394/// relay, so leave the margin.
7495const POLL_WATCHDOG_MS: u64 = 90_000;
7596
97+/// How long a send may stay in flight before we assume the shell lost it.
98+///
99+/// `flushing` gates the whole send loop, and it is only cleared by a
100+/// `SendResult`. If the shell ever drops the HTTP effect (a cancelled task, a
101+/// backgrounded app, a bug), `flushing` stays `true` and the outbox is
102+/// permanently stuck — new messages queue up but never send. The watchdog
103+/// releases the guard so the next `StartFlush` can retry.
104+///
105+/// Kept above the shell's 70 s HTTP timeout for the same reason as
106+/// `POLL_WATCHDOG_MS`: a misfire on a merely slow send is survivable (the late
107+/// `SendResult` finds `in_flight` already cleared and pops nothing), but each
108+/// one costs a redundant request.
109+const FLUSH_WATCHDOG_MS: u64 = 90_000;
110+
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+
76117/// Messages per `/messages` request.
77118///
78119/// The relay rejects a longer batch outright (`413 batch_too_large`,
@@ -106,10 +147,18 @@ const K_PROFILE: &str = "profile";
106147const K_CONTACTS: &str = "contacts";
107148const K_CURSOR: &str = "cursor";
108149const K_OUTBOX: &str = "outbox";
150+const K_SEEN_IDS: &str = "seen_ids";
109151fn k_messages(peer: &str) -> String {
110152 format!("messages:{peer}")
111153}
112154
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+
113162// ---------------------------------------------------------------------------
114163// Effect
115164// ---------------------------------------------------------------------------
@@ -177,6 +226,9 @@ pub enum Event {
177226 LoadedOutbox(#[facet(opaque)] KvData),
178227 #[serde(skip)]
179228 #[facet(skip)]
229+ LoadedSeenIds(#[facet(opaque)] KvData),
230+ #[serde(skip)]
231+ #[facet(skip)]
180232 LoadedMessages(String, #[facet(opaque)] KvData),
181233 #[serde(skip)]
182234 #[facet(skip)]
@@ -210,6 +262,17 @@ pub enum Event {
210262 #[serde(skip)]
211263 #[facet(skip)]
212264 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.
273+ #[serde(skip)]
274+ #[facet(skip)]
275+ AuthWatchdog(u64),
213276 #[serde(skip)]
214277 #[facet(skip)]
215278 ChallengeResult(u64, #[facet(opaque)] HttpResult),
@@ -221,7 +284,7 @@ pub enum Event {
221284 PollResult(u64, #[facet(opaque)] HttpResult),
222285 #[serde(skip)]
223286 #[facet(skip)]
224− SendResult(#[facet(opaque)] HttpResult),
287+ SendResult(u64, #[facet(opaque)] HttpResult),
225288}
226289
227290// ---------------------------------------------------------------------------
@@ -247,9 +310,9 @@ pub enum Page {
247310 EditProfile,
248311}
249312
250−/// The five kv keys loaded at startup (`settings`, `profile`, `contacts`,
251−/// `cursor`, `outbox`). `Connect` waits for all of them.
252−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;
253316
254317/// The batch a live `/messages` request is carrying: the first `count` items of
255318/// the outbox, all bound for `recipient`.
@@ -275,11 +338,23 @@ pub struct Model {
275338 profile: OwnProfile,
276339 contacts: BTreeMap<String, Contact>,
277340 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>>,
278347 outbox: VecDeque<OutboxItem>,
279348 cursor: i64,
280349 token: Option<String>,
281350 conn: ConnStatus,
282351 poll_retries: u32,
352+ /// The send loop's own backoff counter, spent when a `429` arrives with no
353+ /// `Retry-After` to obey. Kept apart from `poll_retries` because the poll loop
354+ /// resets that one on every page it takes: a client polling happily while the
355+ /// relay rate-limits its sends would restart the send backoff at 3 s forever,
356+ /// which is not a backoff.
357+ send_retries: u32,
283358 /// A poll request is in flight. Guards against stacking concurrent long-poll
284359 /// loops (every extra loop re-polls itself forever and never terminates).
285360 polling: bool,
@@ -311,6 +386,12 @@ pub struct Model {
311386 /// items it sent off the head of the outbox, so two overlapping sends would
312387 /// pop each other's messages.
313388 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,
314395 /// What the in-flight send will pop off the head of the outbox when it
315396 /// succeeds — see `InFlight`. `None` while nothing is in flight, and also
316397 /// while a send is in flight whose items have since been removed from under
@@ -335,6 +416,31 @@ pub struct Model {
335416 /// can't land and be persisted against a half-loaded model — which would
336417 /// then be clobbered by the loads still in flight.
337418 loads_pending: u8,
419+ /// A startup kv *read* failed — as opposed to finding the key absent.
420+ ///
421+ /// The distinction is the whole point of `parse_kv`'s `Result<Option<T>, ()>`:
422+ /// an absent key is a fresh install and defaulting is right, but a failed read
423+ /// leaves a hole in the model over data that is still on disk, and writing that
424+ /// hole back destroys it. The outbox is the sharpest case — `ingest_poll`
425+ /// persists it on *every* poll page, so a single failed read followed by one
426+ /// empty poll writes `[]` over a queue of real, unsent messages.
427+ ///
428+ /// So this latches the model as untrustworthy, and every write path checks it:
429+ /// no persist, no ingest, no flush. Nothing is lost, because nothing is
430+ /// written — the data stays on disk for the next launch, which re-runs the
431+ /// loads (`IdentityLoaded` fires on every boot) and clears the flag if they
432+ /// succeed. It is deliberately one flag and not a per-key set: a read that
433+ /// fails because the device is locked fails for every key, and a model that is
434+ /// wrong about its outbox is not one to be trusted about its contacts either.
435+ kv_load_failed: bool,
436+ /// `now_ms()` before which `flush_next` must not send.
437+ ///
438+ /// A `429` schedules a `StartFlush` for the far side of the relay's
439+ /// `Retry-After`, but that timer is not the only thing that fires one — a poll
440+ /// page and a fresh `SendText` both do. Without a pause the wait is decorative:
441+ /// the next poll page walks straight through it and hammers the relay that just
442+ /// asked us to stop.
443+ flush_paused_until: i64,
338444 page: Page,
339445 active_peer: Option<String>,
340446 compose: String,
@@ -350,19 +456,24 @@ impl Default for Model {
350456 profile: OwnProfile::default(),
351457 contacts: BTreeMap::new(),
352458 messages: BTreeMap::new(),
459+ seen_ids: BTreeMap::new(),
353460 outbox: VecDeque::new(),
354461 cursor: 0,
355462 token: None,
356463 conn: ConnStatus::Offline,
357464 poll_retries: 0,
465+ send_retries: 0,
358466 polling: false,
359467 poll_gen: 0,
360468 authenticating: false,
361469 auth_gen: 0,
362470 flushing: false,
471+ flush_gen: 0,
363472 in_flight: None,
364473 send_batch_budget: None,
365474 loads_pending: 0,
475+ kv_load_failed: false,
476+ flush_paused_until: 0,
366477 page: Page::default(),
367478 active_peer: None,
368479 compose: String::new(),
@@ -409,6 +520,24 @@ impl Model {
409520 self.auth_gen
410521 }
411522
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+
412541 /// Abandon the in-flight auth round-trip (either leg).
413542 fn abandon_auth(&mut self) {
414543 self.authenticating = false;
@@ -455,7 +584,7 @@ fn now_ms() -> i64 {
455584 SystemTime::now()
456585 .duration_since(UNIX_EPOCH)
457586 .map(|d| d.as_millis() as i64)
458− .unwrap_or(0)
587+ .unwrap_or(1)
459588}
460589
461590/// Bare lowercased hostname of a server URL (no scheme/port/trailing dot).
@@ -543,6 +672,16 @@ struct PollEvent {
543672 #[serde(rename = "encryptedBlob")]
544673 encrypted_blob: String,
545674}
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+}
546685#[derive(Serialize)]
547686struct SendBatch {
548687 messages: Vec<Envelope>,
@@ -576,12 +715,17 @@ impl App for Skrepka {
576715 // Fan out the startup loads. `Connect` waits until the last of
577716 // them lands (see `startup_load_done`).
578717 model.loads_pending = STARTUP_LOADS;
718+ // A fresh round of loads: whatever a previous one failed to read,
719+ // this one gets to answer for. Clearing it here is what lets a
720+ // relaunch recover from a locked-device read failure.
721+ model.kv_load_failed = false;
579722 Command::all([
580723 KeyValue::get(K_SETTINGS).then_send(Event::LoadedSettings),
581724 KeyValue::get(K_PROFILE).then_send(Event::LoadedProfile),
582725 KeyValue::get(K_CONTACTS).then_send(Event::LoadedContacts),
583726 KeyValue::get(K_CURSOR).then_send(Event::LoadedCursor),
584727 KeyValue::get(K_OUTBOX).then_send(Event::LoadedOutbox),
728+ KeyValue::get(K_SEEN_IDS).then_send(Event::LoadedSeenIds),
585729 ])
586730 .and(render())
587731 }
@@ -600,6 +744,7 @@ impl App for Skrepka {
600744 }
601745 Ok(None) => {}
602746 Err(()) => {
747+ model.kv_load_failed = true;
603748 model.error = Some("could not read settings from storage".into());
604749 }
605750 }
@@ -609,7 +754,10 @@ impl App for Skrepka {
609754 match parse_kv::<OwnProfile>(res) {
610755 Ok(Some(p)) => model.profile = p,
611756 Ok(None) => {}
612− Err(()) => model.error = Some("could not read profile from storage".into()),
757+ Err(()) => {
758+ model.kv_load_failed = true;
759+ model.error = Some("could not read profile from storage".into());
760+ }
613761 }
614762 self.startup_done(model).and(render())
615763 }
@@ -626,7 +774,10 @@ impl App for Skrepka {
626774 }));
627775 }
628776 Ok(None) => {}
629− Err(()) => model.error = Some("could not read contacts from storage".into()),
777+ Err(()) => {
778+ model.kv_load_failed = true;
779+ model.error = Some("could not read contacts from storage".into());
780+ }
630781 }
631782 Command::all(cmds).and(render())
632783 }
@@ -634,7 +785,10 @@ impl App for Skrepka {
634785 match parse_kv::<i64>(res) {
635786 Ok(Some(c)) => model.cursor = c,
636787 Ok(None) => {}
637− Err(()) => model.error = Some("could not read cursor from storage".into()),
788+ Err(()) => {
789+ model.kv_load_failed = true;
790+ model.error = Some("could not read cursor from storage".into());
791+ }
638792 }
639793 self.startup_done(model).and(render())
640794 }
@@ -642,20 +796,70 @@ impl App for Skrepka {
642796 match parse_kv::<Vec<OutboxItem>>(res) {
643797 Ok(Some(list)) => model.outbox = list.into(),
644798 Ok(None) => {}
645− Err(()) => model.error = Some("could not read outbox from storage".into()),
799+ Err(()) => {
800+ model.kv_load_failed = true;
801+ model.error = Some("could not read outbox from storage".into());
802+ }
803+ }
804+ self.startup_done(model).and(render())
805+ }
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+ }
646815 }
647816 self.startup_done(model).and(render())
648817 }
649818 Event::LoadedMessages(peer, res) => {
650819 match parse_kv::<Vec<StoredMessage>>(res) {
651820 Ok(Some(list)) => {
652− 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+ }
653851 }
654852 Ok(None) => {}
655853 Err(()) => {
656− // Don't overwrite a potentially-loaded conversation with
657− // an empty default — the read failed, the data is still
658− // on disk.
854+ // Leaving the conversation absent is not enough on its own.
855+ // An absent conversation is an *empty* one to everything
856+ // downstream, so the next message from this peer would be
857+ // appended to nothing and `persist_messages` would write
858+ // that one message over the whole history still on disk.
859+ // Latch the model as untrustworthy instead — same reasoning
860+ // as the outbox, same fix.
861+ model.kv_load_failed = true;
862+ model.error = Some("could not read messages from storage".into());
659863 }
660864 }
661865 render()
@@ -753,12 +957,30 @@ impl App for Skrepka {
753957 // would then ack a sequence number B never issued, and B would
754958 // drop every message waiting for us.
755959 model.abandon_poll();
756− Command::all([
757− KeyValue::set(K_SETTINGS, json_bytes(&model.settings)).then_send(Event::Saved),
758− KeyValue::set(K_CURSOR, json_bytes(&0i64)).then_send(Event::Saved),
759− ])
760− .and(Command::event(Event::Connect))
761− .and(render())
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();
966+ // The old relay's rate limit was the old relay's opinion. A new one is
967+ // entitled to take our sends immediately, and carrying the pause over
968+ // would park the outbox against a limit that no longer applies.
969+ model.flush_paused_until = 0;
970+ 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;
974+ let writes = if self.refuse_write(model) {
975+ Command::done()
976+ } else {
977+ Command::all([
978+ KeyValue::set(K_SETTINGS, json_bytes(&model.settings))
979+ .then_send(Event::Saved),
980+ KeyValue::set(K_CURSOR, json_bytes(&0i64)).then_send(Event::Saved),
981+ ])
982+ };
983+ writes.and(Command::event(Event::Connect)).and(render())
762984 }
763985 Event::Connect => {
764986 // Connect restarts the auth flow, so whatever challenge/verify was
@@ -769,13 +991,42 @@ impl App for Skrepka {
769991 // and about to resolve. Clearing the guard alone would let it land
770992 // and install its token on top of the attempt replacing it.
771993 model.abandon_auth();
994+ // The shell fires `Connect` on every return to the foreground. A model
995+ // that failed a startup read must not take that invitation: connecting
996+ // leads to polling, and polling into a hole acks mail to the relay that
997+ // we cannot store. Staying offline is what keeps the disk intact until
998+ // a relaunch re-runs the loads.
999+ if model.kv_load_failed {
1000+ return render();
1001+ }
7721002 if model.secret_key.is_some() && model.conn != ConnStatus::Online {
7731003 Command::event(Event::Authenticate)
1004+ } else if model.secret_key.is_some()
1005+ && model.conn == ConnStatus::Online
1006+ && !model.polling
1007+ {
1008+ // The poll loop can die without `conn` going offline: after
1009+ // `PollResult` clears `polling`, the re-poll is chained off
1010+ // `SavedCursor`. If the shell drops that effect (backgrounded
1011+ // app, bug), `polling` stays `false` with no pending re-poll
1012+ // and the watchdog sees `!polling` and does nothing. The
1013+ // status shows "online" but no polling ever happens. Restart
1014+ // the loop here — `Connect` is fired on every foreground return.
1015+ Command::event(Event::Poll)
7741016 } else {
7751017 render()
7761018 }
7771019 }
7781020 Event::AddContact { input, nickname } => {
1021+ // Clear a stale error from a previous failed add.
1022+ model.error = None;
1023+ // A nickname is stored in the contacts kv blob, which is
1024+ // rewritten in full on every change. An unbounded nickname
1025+ // bloats the blob and every subsequent write.
1026+ if nickname.len() > protocol::MAX_DISPLAY_NAME_LEN {
1027+ model.error = Some("nickname too long".into());
1028+ return render();
1029+ }
7791030 match phonemic::try_parse_pubkey(&input) {
7801031 Some(hex) if hex != model.my_pubkey => {
7811032 model
@@ -805,6 +1056,9 @@ impl App for Skrepka {
8051056 if let Some(c) = model.contacts.get_mut(&peer) {
8061057 c.blocked = blocked;
8071058 }
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;
8081062 if blocked {
8091063 // Gating *ingest* on the block only stops the acks we have not
8101064 // queued yet. Anything already sitting in the outbox for this
@@ -823,6 +1077,7 @@ impl App for Skrepka {
8231077 Event::DeleteContact { peer } => {
8241078 model.contacts.remove(&peer);
8251079 model.messages.remove(&peer);
1080+ model.seen_ids.remove(&peer);
8261081 model.outbox.retain(|item| item.recipient != peer);
8271082 // The send in flight may be carrying the very items we just
8281083 // dropped; if so its result must pop nothing.
@@ -839,6 +1094,8 @@ impl App for Skrepka {
8391094 // message the user just asked us to forget. Delete the key.
8401095 KeyValue::delete(k_messages(&peer)).then_send(Event::Saved),
8411096 self.persist_outbox(model),
1097+ // Q18: persist the seen_ids map without the removed peer.
1098+ self.persist_seen_ids(model),
8421099 ])
8431100 .and(render())
8441101 }
@@ -847,6 +1104,32 @@ impl App for Skrepka {
8471104 bio,
8481105 photo,
8491106 } => {
1107+ // Clear a stale error from a previous failed save.
1108+ model.error = None;
1109+ // The recipient's `parse_payload` enforces these caps and
1110+ // silently drops oversized profiles. Without these checks the
1111+ // sender's local profile would be saved, broadcast to every
1112+ // contact, and silently dropped by every recipient — a silent
1113+ // divergence between what the sender sees and what contacts see.
1114+ if display_name.chars().count() > protocol::MAX_DISPLAY_NAME_LEN {
1115+ model.error = Some("display name too long".into());
1116+ return render();
1117+ }
1118+ if bio.chars().count() > protocol::MAX_BIO_LEN {
1119+ model.error = Some("bio too long".into());
1120+ return render();
1121+ }
1122+ if photo.as_ref().is_some_and(|p| p.len() > protocol::MAX_PHOTO_LEN) {
1123+ model.error = Some("photo too large".into());
1124+ return render();
1125+ }
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+ }
8501133 model.profile = OwnProfile {
8511134 display_name,
8521135 bio,
@@ -855,8 +1138,8 @@ impl App for Skrepka {
8551138 model.page = Page::Conversations;
8561139 let ts = now_ms();
8571140 let payload = Payload::Profile {
858− display_name: model.profile.display_name.clone(),
859− bio: model.profile.bio.clone(),
1141+ display_name: Some(model.profile.display_name.clone()),
1142+ bio: Some(model.profile.bio.clone()),
8601143 photo: model.profile.photo.clone(),
8611144 };
8621145 // Broadcast the new profile to every contact we haven't blocked.
@@ -901,7 +1184,7 @@ impl App for Skrepka {
9011184 }
9021185 }
9031186 Command::all([
904− KeyValue::set(K_PROFILE, json_bytes(&model.profile)).then_send(Event::Saved),
1187+ self.persist_profile(model),
9051188 self.persist_outbox(model),
9061189 ])
9071190 .and(Command::event(Event::StartFlush))
@@ -935,6 +1218,11 @@ impl App for Skrepka {
9351218 let gen = model.next_auth_gen();
9361219 req.build()
9371220 .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+ )
9381226 .and(render())
9391227 }
9401228 Err(_) => {
@@ -1099,23 +1387,104 @@ impl App for Skrepka {
10991387 return self.backoff_poll(model);
11001388 }
11011389 model.poll_retries = 0;
1390+ // Ingesting a page into a model with a hole in it is the worst thing
1391+ // this client can do. The cursor write below is what lets the relay
1392+ // delete the mail we just took, so a page ingested into a model that
1393+ // cannot persist it is a page destroyed on both sides at once. Stop
1394+ // the poll loop instead — it restarts on the next launch, which
1395+ // re-runs the loads.
1396+ if model.kv_load_failed {
1397+ model.error = Some(
1398+ "storage read failed — restart the app; messages are not being saved"
1399+ .into(),
1400+ );
1401+ return render();
1402+ }
11021403 let bytes = resp.take_body().unwrap_or_default();
1103− 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;
11041423 let cmd = self.ingest_poll(model, parsed);
11051424 // The re-poll hangs off the *cursor write*, not off this event.
11061425 // Polling again is what acks the batch we just took — the server
11071426 // deletes everything up to `cursor` — so it must not happen until
11081427 // the batch is durable on our side. `SavedCursor` re-polls.
1109− 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 {
11101440 KeyValue::set(K_CURSOR, json_bytes(&model.cursor))
1111− .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),
11121449 )
11131450 .and(render())
11141451 }
11151452
11161453 // ---------------- outbox ----------------
11171454 Event::StartFlush => self.flush_next(model),
1118− Event::SendResult(Ok(resp)) => {
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 {
1462+ return Command::done();
1463+ }
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();
1469+ self.flush_next(model)
1470+ }
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+ }
11191488 // The in-flight send is over; `flush_next` refuses to run while
11201489 // this is set, so it must be cleared before *any* branch below.
11211490 model.flushing = false;
@@ -1143,13 +1512,32 @@ impl App for Skrepka {
11431512 .map(|v| v.last().as_str())
11441513 .and_then(|v| v.trim().parse::<u64>().ok())
11451514 .map_or_else(
1146− || Duration::from_millis(backoff_ms(model.poll_retries)),
1515+ || {
1516+ // Nothing to honour, so guess — on the *send* loop's
1517+ // counter, not the poll loop's. `poll_retries` is reset
1518+ // by every page the poll loop takes, so a client polling
1519+ // happily while the relay throttles its sends would
1520+ // restart the send backoff at 3 s on every page, forever.
1521+ // That is not a backoff; it is a fixed 3 s retry dressed
1522+ // as one.
1523+ let ms = backoff_ms(model.send_retries);
1524+ model.send_retries = model.send_retries.saturating_add(1);
1525+ Duration::from_millis(ms)
1526+ },
11471527 |secs| {
11481528 Duration::from_millis(
11491529 secs.saturating_mul(1000).min(MAX_RETRY_AFTER_MS),
11501530 )
11511531 },
11521532 );
1533+ // Park the send loop for the duration, on the *model*. The timer
1534+ // below is not the only thing that fires `StartFlush` — a poll page
1535+ // and a fresh `SendText` both do — and without a pause they walk
1536+ // straight through the wait and hammer the relay that just asked us
1537+ // to stop. `flush_next` refuses to send until this passes, whoever
1538+ // wakes it.
1539+ model.flush_paused_until =
1540+ now_ms().saturating_add(i64::try_from(delay.as_millis()).unwrap_or(i64::MAX));
11531541 return Time::notify_after(delay)
11541542 .0
11551543 .then_send(|_| Event::StartFlush)
@@ -1180,16 +1568,30 @@ impl App for Skrepka {
11801568 // The batch was accepted, or rejected outright (`self_send` /
11811569 // `invalid_message` — both permanent, and the relay takes a
11821570 // batch all-or-nothing). Either way it is finished with.
1183− for _ in 0..batch.map_or(0, |b| b.count) {
1184− model.outbox.pop_front();
1571+ //
1572+ // The relay is taking our sends again, so the rate-limit backoff
1573+ // has done its job: forget it, or the *next* 429 — possibly hours
1574+ // later, under load that has nothing to do with this one — starts
1575+ // at whatever height the last one climbed to.
1576+ 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+ }
11851584 }
11861585 return self.persist_outbox(model).and(self.flush_next(model));
11871586 }
11881587 // A transient failure: a 5xx, a rate limit, a relay restarting.
11891588 self.retry_batch(model, batch)
11901589 }
1191− Event::SendResult(Err(_)) => {
1590+ Event::SendResult(gen, Err(_)) => {
11921591 // A transport error is the same transient failure as a 5xx.
1592+ if gen != model.flush_gen {
1593+ return Command::done();
1594+ }
11931595 model.flushing = false;
11941596 let batch = model.in_flight.take();
11951597 self.retry_batch(model, batch)
@@ -1213,18 +1615,13 @@ impl App for Skrepka {
12131615
12141616 let active_peer = model.active_peer.clone().unwrap_or_default();
12151617
1216− // `view()` runs on every render, and `ComposeChanged` renders on every
1217− // keystroke. A contact's `photo` is a base64 avatar — tens of KiB each —
1218− // so cloning all of them per keypress meant megabytes of allocation and
1219− // bincode-encoding to type one message.
1220− //
1221− // Only the chat page has a composer (`ComposeChanged` is sent from
1222− // `ChatView` and nowhere else), so only the chat page is on that hot path
1223− // — and while it is up, the only avatar on screen is the peer we are
1224− // talking to. Every *other* page renders from a cold event, and the
1225− // conversations list draws an avatar per row, so it gets the real photos.
1226− // Withholding them there would not save a single keystroke's work; it
1227− // would just replace every contact's picture with their initials.
1618+ // `view()` is called after every event the core processes. The
1619+ // per-keystroke optimization below assumes `ComposeChanged` fires on
1620+ // every keystroke, but the Swift shell only sends it before `SendText`
1621+ // — so in practice this path runs once per message, not per keypress.
1622+ // The optimization is still correct (it withholds avatars on the chat
1623+ // page regardless of how often `view` runs), just the original comment
1624+ // overstated the frequency.
12281625 let chat_open = model.page == Page::Chat;
12291626
12301627 let mut contacts: Vec<ContactVM> = model
@@ -1302,28 +1699,83 @@ impl Skrepka {
13021699 /// still in flight would then overwrite what the poll just ingested, and the
13031700 /// stale `cursor` (0) would make us re-fetch mail we had already acked.
13041701 fn startup_done(&self, model: &mut Model) -> Command<Effect, Event> {
1305− if model.startup_load_done() {
1306− Command::event(Event::Connect)
1307− } else {
1308− Command::done()
1702+ if !model.startup_load_done() {
1703+ return Command::done();
13091704 }
1705+ if model.kv_load_failed {
1706+ // One of the loads could not be read (as opposed to being absent), so
1707+ // the model has a hole in it over data that is still on disk. Do not
1708+ // connect: a poll would ingest mail into that hole and ack it, and the
1709+ // relay would then delete the only copy of messages we never persisted.
1710+ //
1711+ // Everything downstream is already write-guarded, so this is belt and
1712+ // braces — but it is also the only place that can *say* so, and the
1713+ // recovery is a relaunch, which re-runs the loads.
1714+ model.error = Some(
1715+ "storage read failed — restart the app; changes will not be saved until you do"
1716+ .into(),
1717+ );
1718+ return render();
1719+ }
1720+ Command::event(Event::Connect)
13101721 }
13111722
1312− fn persist_contacts(&self, model: &Model) -> Command<Effect, Event> {
1723+ /// The three model-state writes, and the one rule they share: a model that
1724+ /// failed to read a key must never write one.
1725+ ///
1726+ /// `persist_outbox` is the reason this matters. `ingest_poll` calls it on every
1727+ /// poll page — including an empty one — so a failed outbox read followed by a
1728+ /// single idle poll is enough to write `[]` over a queue of real unsent
1729+ /// messages. The user's mail is destroyed by a client doing nothing at all.
1730+ fn refuse_write(&self, model: &mut Model) -> bool {
1731+ if model.kv_load_failed {
1732+ model.error = Some("storage read failed — changes will not be saved".into());
1733+ return true;
1734+ }
1735+ false
1736+ }
1737+
1738+ fn persist_contacts(&self, model: &mut Model) -> Command<Effect, Event> {
1739+ if self.refuse_write(model) {
1740+ return Command::done();
1741+ }
13131742 let list: Vec<Contact> = model.contacts.values().cloned().collect();
13141743 KeyValue::set(K_CONTACTS, json_bytes(&list)).then_send(Event::Saved)
13151744 }
13161745
1317− fn persist_outbox(&self, model: &Model) -> Command<Effect, Event> {
1746+ fn persist_profile(&self, model: &mut Model) -> Command<Effect, Event> {
1747+ if self.refuse_write(model) {
1748+ return Command::done();
1749+ }
1750+ KeyValue::set(K_PROFILE, json_bytes(&model.profile)).then_send(Event::Saved)
1751+ }
1752+
1753+ fn persist_outbox(&self, model: &mut Model) -> Command<Effect, Event> {
1754+ if self.refuse_write(model) {
1755+ return Command::done();
1756+ }
13181757 let list: Vec<OutboxItem> = model.outbox.iter().cloned().collect();
13191758 KeyValue::set(K_OUTBOX, json_bytes(&list)).then_send(Event::Saved)
13201759 }
13211760
1322− fn persist_messages(&self, model: &Model, peer: &str) -> Command<Effect, Event> {
1761+ fn persist_messages(&self, model: &mut Model, peer: &str) -> Command<Effect, Event> {
1762+ if self.refuse_write(model) {
1763+ return Command::done();
1764+ }
13231765 let list = model.messages.get(peer).cloned().unwrap_or_default();
13241766 KeyValue::set(k_messages(peer), json_bytes(&list)).then_send(Event::Saved)
13251767 }
13261768
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+
13271779 fn schedule_reconnect(&self, model: &mut Model) -> Command<Effect, Event> {
13281780 let delay = backoff_ms(model.poll_retries);
13291781 model.poll_retries = model.poll_retries.saturating_add(1);
@@ -1354,16 +1806,49 @@ impl Skrepka {
13541806 let Some(peer) = model.active_peer.clone() else {
13551807 return render();
13561808 };
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;
1812+ // Blocking cuts the peer off in both directions (PROTOCOL.md §4). The
1813+ // ingest side drops their messages; the send side must refuse too, or
1814+ // a blocked peer still receives our texts, acks, and the liveness
1815+ // signal that blocking is supposed to cut off.
1816+ if model.contacts.get(&peer).is_some_and(|c| c.blocked) {
1817+ model.error = Some("cannot send to a blocked contact".into());
1818+ return render();
1819+ }
13571820 let body = model.compose.trim().to_string();
13581821 if body.is_empty() {
13591822 return render();
13601823 }
1824+ // The recipient's `parse_payload` enforces `MAX_BODY_LEN` and silently
1825+ // drops oversized text. Without this check the message would be stored
1826+ // locally, sent to the relay, and marked "sent" — but never arrive, never
1827+ // get acked, and the sender would see it stuck as "sent" forever with no
1828+ // error.
1829+ if body.len() > protocol::MAX_BODY_LEN {
1830+ model.error = Some("message too long".into());
1831+ return render();
1832+ }
1833+
1834+ // Check the contact cap *before* mutating anything. The old order
1835+ // cleared `compose` and inserted the message into `model.messages`
1836+ // first, then hit the cap and returned early — so the user's text was
1837+ // gone (compose cleared), the message sat in an invisible orphan
1838+ // conversation (no contact entry to click on), and nothing was
1839+ // persisted (no `persist_messages` before the early return). On next
1840+ // launch the message was silently lost.
1841+ let was_new = !model.contacts.contains_key(&peer);
1842+ if was_new && model.contacts.len() >= MAX_CONTACTS {
1843+ model.error = Some("contact list is full".into());
1844+ return render();
1845+ }
1846+
13611847 model.compose.clear();
13621848 let ts = now_ms();
13631849 let id = uuid::Uuid::new_v4().to_string();
13641850
13651851 let convo = model.messages.entry(peer.clone()).or_default();
1366− let first_to_peer = !convo.iter().any(|m| m.outgoing);
13671852 // Insert in order rather than appending: an incoming message may hold a
13681853 // `ts` up to `MAX_FUTURE_SKEW_MS` ahead of ours, so appending would leave
13691854 // the conversation unsorted — and `trim_history` ages out the *front*.
@@ -1379,24 +1864,31 @@ impl Skrepka {
13791864 );
13801865 trim_history(convo);
13811866
1382− // Auto-create a bare contact if messaging a brand-new key.
1383− model
1384− .contacts
1385− .entry(peer.clone())
1386− .or_insert_with(|| Contact::new(peer.clone(), String::new(), ts));
1867+ // Auto-create a bare contact if messaging a brand-new key, respecting
1868+ // the same `MAX_CONTACTS` cap `ingest_poll` enforces. This is
1869+ // user-driven (not an attack vector), but the cap exists to bound the
1870+ // `contacts` kv blob, which is rewritten in full on every change.
1871+ if was_new {
1872+ model
1873+ .contacts
1874+ .insert(peer.clone(), Contact::new(peer.clone(), String::new(), ts));
1875+ }
13871876
13881877 let text = Payload::Text { id, body };
13891878 model.outbox.push_back(OutboxItem::new(
13901879 peer.clone(),
13911880 Arc::new(protocol::serialize_payload(&text, ts)),
13921881 ));
1393− // On first contact, also share our profile. Marked as a profile like any
1394− // other, so a `SaveProfile` before this ever goes out supersedes it rather
1395− // than queueing a second one behind it.
1396− if first_to_peer {
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.)
1888+ if was_new {
13971889 let profile = Payload::Profile {
1398− display_name: model.profile.display_name.clone(),
1399− bio: model.profile.bio.clone(),
1890+ display_name: Some(model.profile.display_name.clone()),
1891+ bio: Some(model.profile.bio.clone()),
14001892 photo: model.profile.photo.clone(),
14011893 };
14021894 model.outbox.push_back(OutboxItem::profile(
@@ -1441,6 +1933,20 @@ impl Skrepka {
14411933 if model.flushing {
14421934 return Command::done();
14431935 }
1936+ // A failed startup read means the outbox in memory is not the outbox on
1937+ // disk. Sending from it is not the danger — `persist_outbox` refusing to
1938+ // write is — but a send that cannot record having happened is one that will
1939+ // happen again on the next launch, from the queue still on disk. Sit still.
1940+ if model.kv_load_failed {
1941+ return Command::done();
1942+ }
1943+ // A 429 parked the loop until the far side of the relay's `Retry-After`.
1944+ // The timer that set the pause also scheduled the `StartFlush` that resumes
1945+ // us, so whoever woke us early — a poll page, a fresh `SendText` — the
1946+ // answer is the same: the relay asked us to stop until then.
1947+ if now_ms() < model.flush_paused_until {
1948+ return Command::done();
1949+ }
14441950 let (Some(id), Some(token)) = (model.identity(), model.token.clone()) else {
14451951 return Command::done();
14461952 };
@@ -1518,6 +2024,7 @@ impl Skrepka {
15182024 };
15192025
15202026 model.flushing = true;
2027+ let gen = model.next_flush_gen();
15212028 model.in_flight = Some(InFlight {
15222029 recipient,
15232030 count: messages.len(),
@@ -1528,7 +2035,12 @@ impl Skrepka {
15282035 .header("authorization", format!("Bearer {token}"))
15292036 .body_json(&SendBatch { messages })
15302037 {
1531− 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)))
2039+ .and(
2040+ Time::notify_after(Duration::from_millis(FLUSH_WATCHDOG_MS))
2041+ .0
2042+ .then_send(move |_| Event::FlushWatchdog(gen)),
2043+ ),
15322044 Err(_) => {
15332045 // Not a dead loop: the next StartFlush (from a poll page or a
15342046 // fresh send) retries the head of the outbox.
@@ -1590,8 +2102,28 @@ impl Skrepka {
15902102
15912103 /// Decrypt and fold a poll page into the model; queue delivery acks.
15922104 fn ingest_poll(&self, model: &mut Model, page: PollResp) -> Command<Effect, Event> {
1593− if page.cursor > model.cursor {
1594− model.cursor = page.cursor;
2105+ // The caller already refuses to poll into a broken model, and must — the
2106+ // cursor write is its own, and advancing it is what acks the page. This
2107+ // stands guard for whoever calls this next: ingesting is only safe if what
2108+ // is ingested can be written down.
2109+ if model.kv_load_failed {
2110+ return Command::done();
2111+ }
2112+ 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;
15952127 }
15962128 let Some(id) = model.identity() else {
15972129 return Command::done();
@@ -1600,12 +2132,24 @@ impl Skrepka {
16002132 let now = now_ms();
16012133 let mut touched_convos: HashSet<String> = HashSet::new();
16022134 let mut contacts_dirty = false;
2135+ let mut outbox_dirty = false;
16032136 // sender hex -> message ids needing acks
16042137 let mut ack_targets: BTreeMap<String, Vec<String>> = BTreeMap::new();
1605− // Lazily-built per-sender sets of known message IDs for O(1) text dedup.
1606− 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();
16072142
16082143 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;
16092153 for ev in page.events {
16102154 // Reject an oversized blob on its hex length, before it costs us a
16112155 // decode. No conforming sender produces one (the relay caps blobs at
@@ -1613,12 +2157,37 @@ impl Skrepka {
16132157 if ev.encrypted_blob.len() > MAX_BLOB_LEN * 2 {
16142158 continue;
16152159 }
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;
16162177 match budget.checked_sub(ev.encrypted_blob.len() / 2) {
16172178 Some(left) => budget = left,
16182179 None => {
1619− // A page this large is a misbehaving relay, not real mail.
1620− // Stop here; the rest of the page is discarded exactly as an
1621− // undecryptable blob would be.
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+ }
16222191 model.error = Some("relay sent an oversized poll page".into());
16232192 break;
16242193 }
@@ -1632,6 +2201,10 @@ impl Skrepka {
16322201 let Some(parsed) = protocol::parse_payload(&dec.plaintext) else {
16332202 continue;
16342203 };
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;
16352208 let sender = dec.sender_hex;
16362209
16372210 // Mail from ourselves. A relay cannot forge this — the sender identity
@@ -1658,8 +2231,14 @@ impl Skrepka {
16582231 // `ts` is chosen by the sender and drives conversation ordering,
16592232 // in-chat ordering, and the displayed time. Bound it to the near
16602233 // future so a peer cannot pin itself to the top of the list forever
1661− // with `ts = i64::MAX`.
1662− let ts = parsed.ts.min(now.saturating_add(MAX_FUTURE_SKEW_MS));
2234+ // with `ts = i64::MAX`, and bound it to the non-negative past so a
2235+ // malicious peer cannot make their own message invisible by dating
2236+ // it at `i64::MIN` — `insert_sorted` would put it at position 0 and
2237+ // `trim_history` would age it out first.
2238+ let ts = parsed
2239+ .ts
2240+ .max(0)
2241+ .min(now.saturating_add(MAX_FUTURE_SKEW_MS));
16632242 let known = model.contacts.contains_key(&sender);
16642243 match parsed.payload {
16652244 Payload::Text { id: msg_id, body } => {
@@ -1677,12 +2256,35 @@ impl Skrepka {
16772256 Contact::new(sender.clone(), String::new(), ts),
16782257 );
16792258 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;
16802271 }
16812272 let convo = model.messages.entry(sender.clone()).or_default();
1682− // Lazily build the per-sender ID set on first touch.
1683− let id_set = seen_ids.entry(sender.clone()).or_insert_with(|| {
1684− convo.iter().map(|m| m.id.clone()).collect()
1685− });
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+ }
16862288 // dedup by id — O(1) via HashSet
16872289 if !id_set.contains(&msg_id) {
16882290 // The conversation is *kept* sorted rather than re-sorted:
@@ -1700,6 +2302,29 @@ impl Skrepka {
17002302 },
17012303 );
17022304 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+ }
17032328 trim_history(convo);
17042329 touched_convos.insert(sender.clone());
17052330 // Only ack a message we actually stored — a duplicate id
@@ -1708,39 +2333,26 @@ impl Skrepka {
17082333 }
17092334 }
17102335 Payload::DeliveryAck { ack_ids } => {
1711− // Staleness check: drop a replayed ack whose ts predates the
1712− // last one we processed for this contact.
1713− let stale = model
1714− .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
17152344 .get(&sender)
1716− .is_some_and(|c| ts < c.last_ack_ts);
1717− if stale {
1718− continue;
1719− }
1720− // The watermark is clamped to *now*, not merely into the skew
1721− // window: `ts` is the peer's to choose, and `last_ack_ts` is
1722− // what every later ack is measured against. A single ack at
1723− // `ts = i64::MAX` would otherwise park the watermark a minute
1724− // into the future, and every honest ack that followed — the
1725− // ones that actually mark our messages delivered — would read
1726− // as stale and be dropped.
1727− if let Some(c) = model.contacts.get_mut(&sender) {
1728− c.last_ack_ts = ts.min(now);
1729− contacts_dirty = true;
1730− }
1731− let ack_set: HashSet<String> = ack_ids.iter().cloned().collect();
1732− if let Some(convo) = model.messages.get_mut(&sender) {
1733− let mut changed = false;
1734− for m in convo.iter_mut() {
1735− if m.outgoing && ack_set.contains(&m.id) && !m.delivered {
1736− m.delivered = true;
1737− changed = true;
1738− }
1739− }
1740− if changed {
1741− touched_convos.insert(sender);
1742− }
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;
17432350 }
2351+ pending_acks
2352+ .entry(sender.clone())
2353+ .or_default()
2354+ .ids
2355+ .extend(ack_ids);
17442356 }
17452357 Payload::Profile {
17462358 display_name,
@@ -1760,17 +2372,24 @@ impl Skrepka {
17602372 // profile, since every honest update it sent afterwards would
17612373 // read as a stale replay of itself.
17622374 //
1763− // Strict `>` (not `>=`): a replayed profile at the *same* `ts`
1764− // as the one we already stored is a re-delivery, not an update.
1765− // `>=` would let a captured blob overwrite the cached profile
1766− // with an identical one — harmless in the normal case, but it
1767− // also means a peer who sends `ts = 0` (the default when the
1768− // field is missing) can overwrite another `ts = 0` profile
1769− // indefinitely.
1770− if ts > c.last_profile_ts {
1771− c.display_name = display_name;
1772− c.bio = bio;
1773− c.photo = photo;
2375+ // `>=` (not `>`): a contact's `last_profile_ts` starts at 0,
2376+ // and a first profile whose `ts` is also 0 — the default when
2377+ // the field is missing — must still be accepted. `>` would
2378+ // silently drop it, leaving the contact with no profile at all.
2379+ // A replayed profile at the same `ts` overwrites with identical
2380+ // content, which is a no-op in practice.
2381+ 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+ }
17742393 c.last_profile_ts = ts.min(now);
17752394 contacts_dirty = true;
17762395 }
@@ -1778,6 +2397,42 @@ impl Skrepka {
17782397 }
17792398 }
17802399
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+
17812436 // Queue delivery acks, in batches the recipient will actually accept.
17822437 //
17832438 // `parse_payload` drops a `delivery.ack` carrying more than `MAX_ACK_IDS`
@@ -1800,6 +2455,7 @@ impl Skrepka {
18002455 Arc::new(protocol::serialize_payload(&payload, ack_ts)),
18012456 ));
18022457 }
2458+ outbox_dirty = true;
18032459 }
18042460
18052461 // Persist everything touched.
@@ -1807,10 +2463,16 @@ impl Skrepka {
18072463 for peer in &touched_convos {
18082464 cmds.push(self.persist_messages(model, peer));
18092465 }
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+ }
18102470 if contacts_dirty {
18112471 cmds.push(self.persist_contacts(model));
18122472 }
1813− cmds.push(self.persist_outbox(model));
2473+ if outbox_dirty {
2474+ cmds.push(self.persist_outbox(model));
2475+ }
18142476 cmds.push(Command::event(Event::StartFlush));
18152477 Command::all(cmds)
18162478 }
@@ -1883,7 +2545,7 @@ mod tests {
18832545 use crate::model::{MAX_OUTBOX_RETRIES, OUTBOX_TTL_MS};
18842546
18852547 fn with_identity() -> Model {
1886− let id = Identity::from_seed(&[3u8; 32]);
2548+ let id = Identity::from_seed(&[3u8; 32]).unwrap();
18872549 Model {
18882550 secret_key: Some(Zeroizing::new(id.secret_key.to_vec())),
18892551 my_pubkey: id.public_key_hex(),
@@ -1892,7 +2554,7 @@ mod tests {
18922554 }
18932555
18942556 fn peer_hex(seed: u8) -> String {
1895− Identity::from_seed(&[seed; 32]).public_key_hex()
2557+ Identity::from_seed(&[seed; 32]).unwrap().public_key_hex()
18962558 }
18972559
18982560 /// `Command::event` *queues* an event; it does not run it. The runtime drains
@@ -1915,7 +2577,7 @@ mod tests {
19152577
19162578 /// A poll page carrying one payload, encrypted from `sender_seed` to `me`.
19172579 fn page_from(sender_seed: u8, me: &Model, payload: &Payload, ts: i64) -> PollResp {
1918− let sender = Identity::from_seed(&[sender_seed; 32]);
2580+ let sender = Identity::from_seed(&[sender_seed; 32]).unwrap();
19192581 let recipient = Identity::from_secret_bytes(me.secret_key.as_ref().unwrap()).unwrap();
19202582 let json = protocol::serialize_payload(payload, ts);
19212583 let blob = crate::crypto::encrypt(
@@ -2004,7 +2666,8 @@ mod tests {
20042666 /// One full send round-trip: flush the outbox, then answer with `status`.
20052667 fn send_and_answer(app: &Skrepka, m: &mut Model, status: u16) {
20062668 let _ = app.update(Event::StartFlush, m);
2007− 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);
20082671 }
20092672
20102673 // -----------------------------------------------------------------------
@@ -2033,7 +2696,7 @@ mod tests {
20332696 // The last one exhausts the budget: the head goes, and what was stuck
20342697 // behind it goes out.
20352698 let _ = app.update(Event::StartFlush, &mut m);
2036− 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);
20372700
20382701 assert_eq!(m.outbox.len(), 1, "the undeliverable item is dropped");
20392702 assert_eq!(m.outbox[0].recipient, behind, "and the queue moves on");
@@ -2152,7 +2815,7 @@ mod tests {
21522815
21532816 let _ = app.update(Event::StartFlush, &mut m);
21542817 let mut cmd = app.update(
2155− Event::SendResult(Ok(response_with_header(429, "retry-after", "2"))),
2818+ Event::SendResult(m.flush_gen, Ok(response_with_header(429, "retry-after", "2"))),
21562819 &mut m,
21572820 );
21582821
@@ -2168,18 +2831,95 @@ mod tests {
21682831
21692832 /// Without the header there is nothing to honour, so fall back to our own
21702833 /// backoff rather than hammering the relay that just asked us to slow down.
2834+ ///
2835+ /// On the *send* loop's counter. `poll_retries` is reset by every page the poll
2836+ /// loop takes, so spending it here means a client that polls happily while the
2837+ /// relay throttles its sends restarts the send backoff at 3 s forever.
21712838 #[test]
2172− fn a_429_without_retry_after_falls_back_to_the_usual_backoff() {
2839+ fn a_429_without_header_uses_send_side_backoff() {
21732840 let app = Skrepka;
21742841 let mut m = with_identity();
21752842 m.token = Some("t".into());
21762843 m.outbox.push_back(queued(&peer_hex(9), "hi"));
2844+ // The poll loop is healthy and has just taken a page — exactly the state that
2845+ // used to reset the send backoff out from under itself.
2846+ m.poll_retries = 0;
21772847
21782848 let _ = app.update(Event::StartFlush, &mut m);
2179− 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);
21802850
21812851 assert_eq!(m.outbox[0].retries, 0, "still not the message's fault");
2182− assert_eq!(scheduled_delays(&mut cmd), vec![backoff_ms(m.poll_retries)]);
2852+ assert_eq!(
2853+ scheduled_delays(&mut cmd),
2854+ vec![backoff_ms(0)],
2855+ "the first send-side backoff"
2856+ );
2857+ assert_eq!(m.send_retries, 1, "and the send loop's own counter is spent");
2858+
2859+ // A second 429 climbs, rather than restarting at 3 s.
2860+ m.flush_paused_until = 0;
2861+ let _ = app.update(Event::StartFlush, &mut m);
2862+ let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(429))), &mut m);
2863+ assert_eq!(scheduled_delays(&mut cmd), vec![backoff_ms(1)], "it climbs");
2864+ assert_eq!(m.send_retries, 2);
2865+ }
2866+
2867+ /// The `Retry-After` timer is not the only thing that fires `StartFlush` — a poll
2868+ /// page and a fresh `SendText` both do. Without a pause latched on the model the
2869+ /// wait is decorative: the next poll page walks through it and hammers the relay
2870+ /// that just asked us to stop.
2871+ #[test]
2872+ fn a_429_pauses_flush_until_the_delay_elapses() {
2873+ let app = Skrepka;
2874+ let mut m = with_identity();
2875+ m.token = Some("t".into());
2876+ m.outbox.push_back(queued(&peer_hex(9), "hi"));
2877+
2878+ let _ = app.update(Event::StartFlush, &mut m);
2879+ let _ = app.update(
2880+ Event::SendResult(m.flush_gen, Ok(response_with_header(429, "retry-after", "2"))),
2881+ &mut m,
2882+ );
2883+ assert!(
2884+ m.flush_paused_until > now_ms(),
2885+ "the relay's wait is latched on the model, not just on a timer"
2886+ );
2887+
2888+ // Anything else that wakes the send loop during the wait — a poll page, a new
2889+ // message — must find it closed.
2890+ let mut cmd = app.update(Event::StartFlush, &mut m);
2891+ assert!(
2892+ sent_batches(&mut cmd).is_empty(),
2893+ "nothing goes out while the relay has asked us to wait"
2894+ );
2895+ assert!(!m.flushing, "and the loop is left free to resume later");
2896+ assert_eq!(m.outbox.len(), 1, "the message is still queued");
2897+
2898+ // Time passes (the clock cannot be moved in a test, so retire the pause the
2899+ // way the elapsing of it would) and the timer's own `StartFlush` lands.
2900+ m.flush_paused_until = 0;
2901+ let mut cmd = app.update(Event::StartFlush, &mut m);
2902+ assert_eq!(
2903+ sent_batches(&mut cmd).len(),
2904+ 1,
2905+ "and once the wait is over the batch goes out"
2906+ );
2907+ }
2908+
2909+ /// A send the relay accepts means the throttle is off. Keeping the counter would
2910+ /// make the next 429 — hours later, under unrelated load — resume at whatever
2911+ /// height the last one climbed to.
2912+ #[test]
2913+ fn a_successful_send_clears_the_send_side_backoff() {
2914+ let app = Skrepka;
2915+ let mut m = with_identity();
2916+ m.token = Some("t".into());
2917+ m.send_retries = 4;
2918+ m.outbox.push_back(queued(&peer_hex(9), "hi"));
2919+
2920+ send_and_answer(&app, &mut m, 200);
2921+
2922+ assert_eq!(m.send_retries, 0, "the relay is taking our sends again");
21832923 }
21842924
21852925 /// `Retry-After` is a number a relay chooses and we obey. Unbounded, one of
@@ -2193,7 +2933,7 @@ mod tests {
21932933
21942934 let _ = app.update(Event::StartFlush, &mut m);
21952935 let mut cmd = app.update(
2196− Event::SendResult(Ok(response_with_header(429, "retry-after", "999999999"))),
2936+ Event::SendResult(m.flush_gen, Ok(response_with_header(429, "retry-after", "999999999"))),
21972937 &mut m,
21982938 );
21992939
@@ -2228,7 +2968,7 @@ mod tests {
22282968 assert!(batches[0].iter().all(|e| e.to == peer));
22292969
22302970 // ...and its success clears the whole batch, not just the head.
2231− 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);
22322972 assert!(m.outbox.is_empty());
22332973 }
22342974
@@ -2253,7 +2993,7 @@ mod tests {
22532993 assert_eq!(batches[0].len(), 2, "only the run at the head");
22542994 assert!(batches[0].iter().all(|e| e.to == a));
22552995
2256− 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);
22572997 assert_eq!(m.outbox.len(), 2);
22582998 let batches = sent_batches(&mut cmd);
22592999 assert_eq!(batches[0].len(), 1, "then b's single message");
@@ -2276,7 +3016,7 @@ mod tests {
22763016 let batches = sent_batches(&mut cmd);
22773017 assert_eq!(batches[0].len(), MAX_SEND_BATCH);
22783018
2279− 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);
22803020 assert_eq!(m.outbox.len(), 20, "the rest stays queued for the next batch");
22813021 }
22823022
@@ -2338,7 +3078,7 @@ mod tests {
23383078 let sent_bytes: usize = first[0].iter().map(|e| e.encrypted_blob.len()).sum();
23393079
23403080 // ...and the relay says that was too big.
2341− 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);
23423082
23433083 assert_eq!(
23443084 m.send_batch_budget,
@@ -2359,7 +3099,7 @@ mod tests {
23593099 assert!(shrunk > 0, "but not an empty one");
23603100
23613101 // The relay takes the smaller batch, and exactly it is popped.
2362− 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);
23633103 assert_eq!(m.outbox.len(), 5 - shrunk, "exactly the batch that landed is gone");
23643104 assert_eq!(
23653105 m.send_batch_budget,
@@ -2464,7 +3204,7 @@ mod tests {
24643204 let parsed = protocol::parse_payload(item.envelope_json.as_bytes()).unwrap();
24653205 match parsed.payload {
24663206 Payload::Profile { display_name, .. } => {
2467− 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");
24683208 }
24693209 _ => panic!("expected a profile"),
24703210 }
@@ -2507,12 +3247,12 @@ mod tests {
25073247 );
25083248 assert_eq!(m.outbox.len(), 2, "the in-flight item is left alone");
25093249
2510− 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);
25113251 assert_eq!(m.outbox.len(), 1, "the sent one is popped");
25123252 let parsed = protocol::parse_payload(m.outbox[0].envelope_json.as_bytes()).unwrap();
25133253 match parsed.payload {
25143254 Payload::Profile { display_name, .. } => assert_eq!(
2515− display_name, "New",
3255+ display_name, Some("New".to_string()),
25163256 "and the newer profile survives to be sent"
25173257 ),
25183258 _ => panic!("expected a profile"),
@@ -2553,7 +3293,7 @@ mod tests {
25533293 assert_eq!(m.outbox.len(), 1);
25543294
25553295 // The send lands. It must pop nothing.
2556− 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);
25573297 assert_eq!(m.outbox.len(), 1, "the friend's message is untouched");
25583298 assert_eq!(m.outbox[0].recipient, friend);
25593299 }
@@ -2694,8 +3434,8 @@ mod tests {
26943434 .insert(peer.clone(), Contact::new(peer.clone(), String::new(), 0));
26953435
26963436 let profile = |name: &str| Payload::Profile {
2697− display_name: name.into(),
2698− bio: String::new(),
3437+ display_name: Some(name.to_string()),
3438+ bio: Some(String::new()),
26993439 photo: None,
27003440 };
27013441 let page = page_from(9, &m, &profile("Far"), i64::MAX);
@@ -2703,8 +3443,8 @@ mod tests {
27033443 let first_ts = m.contacts[&peer].last_profile_ts;
27043444 assert!(first_ts <= now_ms());
27053445
2706− // A later profile at a strictly greater ts must still apply (the
2707− // staleness check is strict `>`, not `>=`).
3446+ // A later profile at a greater ts must still apply — the clamping
3447+ // is what stops the far-future one from freezing the peer.
27083448 let page = page_from(9, &m, &profile("Now"), first_ts + 1);
27093449 let _ = app.ingest_poll(&mut m, page);
27103450 assert_eq!(
@@ -2792,8 +3532,11 @@ mod tests {
27923532 let (m, peer) = ingest_one_text_from_an_unblocked_peer();
27933533 assert_eq!(m.messages[&peer].len(), 1);
27943534 assert_eq!(m.messages[&peer][0].body, "hi");
2795− 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");
27963538 assert_eq!(m.outbox[0].recipient, peer);
3539+ assert_eq!(m.outbox[1].recipient, peer);
27973540 }
27983541
27993542 /// A peer sending `ts = i64::MAX` used to pin itself to the top of the
@@ -2974,6 +3717,28 @@ mod tests {
29743717 assert!(m.error.is_some());
29753718 }
29763719
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+
29773742 #[test]
29783743 fn send_text_enqueues_message_profile_and_outbox() {
29793744 let app = Skrepka;
@@ -3011,6 +3776,41 @@ mod tests {
30113776 assert!(m.outbox.is_empty());
30123777 }
30133778
3779+ /// The contact cap must be checked *before* mutating anything. The old
3780+ /// order cleared `compose` and inserted the message into `model.messages`
3781+ /// first, then hit the cap and returned early — losing the user's text and
3782+ /// leaving an invisible orphan conversation that was never persisted.
3783+ #[test]
3784+ fn send_text_to_a_new_peer_at_the_contact_cap_preserves_compose() {
3785+ let app = Skrepka;
3786+ let mut m = with_identity();
3787+ // Fill the contact list to the cap.
3788+ for i in 0..MAX_CONTACTS {
3789+ let k = format!("{i:064x}");
3790+ m.contacts.insert(k.clone(), Contact::new(k, String::new(), 0));
3791+ }
3792+ let new_peer = peer_hex(9);
3793+ m.active_peer = Some(new_peer.clone());
3794+ m.compose = "hello new friend".into();
3795+
3796+ let _ = app.update(Event::SendText, &mut m);
3797+
3798+ assert_eq!(
3799+ m.compose, "hello new friend",
3800+ "compose must not be cleared on rejection"
3801+ );
3802+ assert!(
3803+ m.messages.get(&new_peer).is_none_or(Vec::is_empty),
3804+ "no orphan conversation must be created"
3805+ );
3806+ assert!(m.outbox.is_empty(), "nothing must be queued");
3807+ assert!(!m.contacts.contains_key(&new_peer), "no contact created");
3808+ assert!(
3809+ m.error.as_ref().is_some_and(|e| e.contains("full")),
3810+ "the user is told why"
3811+ );
3812+ }
3813+
30143814 #[test]
30153815 fn set_blocked_toggles_and_view_reflects_it() {
30163816 let app = Skrepka;
@@ -3030,6 +3830,30 @@ mod tests {
30303830 assert!(vm.contacts.iter().any(|c| c.pubkey == peer && c.blocked));
30313831 }
30323832
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+
30333857 /// Gating ingest stops the acks we have not queued yet — but the ack for the
30343858 /// message that made the user hit "block" is already in the outbox, and an ack
30353859 /// is exactly the "we are online and reading you" signal the block exists to
@@ -3149,7 +3973,7 @@ mod tests {
31493973 assert!(m.flushing);
31503974
31513975 let resp = crux_http::testing::ResponseBuilder::ok().body(Vec::new()).build();
3152− let _ = app.update(Event::SendResult(Ok(resp)), &mut m);
3976+ let _ = app.update(Event::SendResult(m.flush_gen, Ok(resp)), &mut m);
31533977
31543978 assert!(!m.flushing, "flushing must be cleared on the success path");
31553979 assert!(m.outbox.is_empty(), "the sent item is dropped from the outbox");
@@ -3171,7 +3995,7 @@ mod tests {
31713995
31723996 // Its success must both drop the item and let the *next* one go out.
31733997 let resp = crux_http::testing::ResponseBuilder::ok().body(Vec::new()).build();
3174− let _ = app.update(Event::SendResult(Ok(resp)), &mut m);
3998+ let _ = app.update(Event::SendResult(m.flush_gen, Ok(resp)), &mut m);
31753999 assert_eq!(m.outbox.len(), 1);
31764000 assert!(m.flushing, "the second item is now in flight");
31774001 }
@@ -3314,6 +4138,7 @@ mod tests {
33144138 Event::LoadedProfile(Ok(None)),
33154139 Event::LoadedContacts(Ok(None)),
33164140 Event::LoadedCursor(Ok(None)),
4141+ Event::LoadedSeenIds(Ok(None)),
33174142 ]
33184143 .into_iter()
33194144 .enumerate()
@@ -3327,7 +4152,7 @@ mod tests {
33274152 assert_eq!(connects(&mut cmd), 0, "and none of them connects");
33284153 }
33294154
3330− // The fifth — and only the fifth — emits Connect.
4155+ // The sixth — and only the sixth — emits Connect.
33314156 let mut cmd = app.update(Event::LoadedOutbox(Ok(None)), &mut m);
33324157 assert_eq!(m.loads_pending, 0);
33334158 assert_eq!(connects(&mut cmd), 1, "the last load connects, exactly once");
@@ -3337,6 +4162,94 @@ mod tests {
33374162 assert_eq!(connects(&mut cmd), 0);
33384163 }
33394164
4165+ /// The sharpest edge of a failed read, and the one that costs the user real
4166+ /// messages: `ingest_poll` persists the outbox on *every* poll page, including
4167+ /// an empty one. So a single unreadable outbox — a background wake with the
4168+ /// device locked — followed by one idle poll used to write `[]` over a queue of
4169+ /// unsent mail. The client destroys the user's messages by doing nothing at all.
4170+ ///
4171+ /// A read failure is not an empty key. Nothing may be written until a launch
4172+ /// reads the key successfully.
4173+ #[test]
4174+ fn a_failed_read_never_writes_over_what_it_could_not_read() {
4175+ let app = Skrepka;
4176+ let mut m = with_identity();
4177+ m.loads_pending = STARTUP_LOADS;
4178+
4179+ let connects = |cmd: &mut Command<Effect, Event>| {
4180+ cmd.events()
4181+ .filter(|e| matches!(e, Event::Connect))
4182+ .count()
4183+ };
4184+ // Every key a command actually writes. The model's own view of what it saved
4185+ // cannot see these — the write lives entirely in the effect.
4186+ let kv_writes = |cmd: &mut Command<Effect, Event>| {
4187+ cmd.effects()
4188+ .filter_map(|e| match e {
4189+ Effect::KeyValue(req) => match &req.operation {
4190+ crux_kv::KeyValueOperation::Set { key, .. } => Some(key.clone()),
4191+ _ => None,
4192+ },
4193+ _ => None,
4194+ })
4195+ .collect::<Vec<_>>()
4196+ };
4197+
4198+ let _ = app.update(Event::LoadedSettings(Ok(None)), &mut m);
4199+ let _ = app.update(Event::LoadedProfile(Ok(None)), &mut m);
4200+ let _ = app.update(Event::LoadedContacts(Ok(None)), &mut m);
4201+ let _ = app.update(Event::LoadedCursor(Ok(None)), &mut m);
4202+ // The outbox is the one that could not be read — a background wake with the
4203+ // device locked. The model now believes it is empty; the disk says otherwise.
4204+ let mut cmd = app.update(
4205+ Event::LoadedOutbox(Err(crux_kv::error::KeyValueError::Io {
4206+ message: "device is locked".into(),
4207+ })),
4208+ &mut m,
4209+ );
4210+
4211+ assert!(m.kv_load_failed, "the model knows it has a hole in it");
4212+ assert!(m.error.is_some(), "and says so");
4213+ assert_eq!(
4214+ connects(&mut cmd),
4215+ 0,
4216+ "the last load does not connect: a poll would ack mail we cannot store"
4217+ );
4218+
4219+ // Every write path is closed, whatever tries to open it.
4220+ let mut cmd = app.update(
4221+ Event::AddContact {
4222+ input: peer_hex(9),
4223+ nickname: "n".into(),
4224+ },
4225+ &mut m,
4226+ );
4227+ assert!(
4228+ kv_writes(&mut cmd).is_empty(),
4229+ "contacts are not written over an unread model"
4230+ );
4231+
4232+ // And the regression itself. `ingest_poll` persists the outbox on every page,
4233+ // so one idle poll used to be enough to write `[]` over the queue on disk.
4234+ m.token = Some("t".into());
4235+ let resp = crux_http::testing::ResponseBuilder::ok()
4236+ .body(br#"{"events":[],"cursor":42}"#.to_vec())
4237+ .build();
4238+ let mut cmd = app.update(Event::PollResult(m.poll_gen, Ok(resp)), &mut m);
4239+ assert!(
4240+ kv_writes(&mut cmd).is_empty(),
4241+ "an empty poll writes nothing — not the outbox, not the cursor"
4242+ );
4243+
4244+ // 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();
4246+ let _ = app.update(Event::IdentityLoaded(sk), &mut m);
4247+ assert!(
4248+ !m.kv_load_failed,
4249+ "a fresh round of loads gets to answer for itself"
4250+ );
4251+ }
4252+
33404253 /// `Authenticate` is reachable from startup, a 401 on poll, a 401 on send and
33414254 /// a reconnect timer. Without a guard a network blip fires all of them, and
33424255 /// every token that comes back starts its own poll loop.
@@ -3436,8 +4349,8 @@ mod tests {
34364349 );
34374350
34384351 let profile = Payload::Profile {
3439− display_name: "Mallory".into(),
3440− bio: "spam".into(),
4352+ display_name: Some("Mallory".into()),
4353+ bio: Some("spam".into()),
34414354 photo: Some("aGk=".into()),
34424355 };
34434356 let page = page_from(9, &m, &profile, now_ms());
@@ -3518,8 +4431,8 @@ mod tests {
35184431 let peer = peer_hex(9);
35194432
35204433 let profile = Payload::Profile {
3521− display_name: "Mallory".into(),
3522− bio: String::new(),
4434+ display_name: Some("Mallory".into()),
4435+ bio: Some(String::new()),
35234436 photo: Some("aGk=".into()),
35244437 };
35254438 let page = page_from(9, &m, &profile, now_ms());
@@ -3731,7 +4644,7 @@ mod tests {
37314644
37324645 // One page holding more messages from one sender than fits in a single ack.
37334646 let count = protocol::MAX_ACK_IDS + 5;
3734− let sender = Identity::from_seed(&[9u8; 32]);
4647+ let sender = Identity::from_seed(&[9u8; 32]).unwrap();
37354648 let me = Identity::from_secret_bytes(m.secret_key.as_ref().unwrap()).unwrap();
37364649 let events = (0..count)
37374650 .map(|i| {
@@ -3834,4 +4747,281 @@ mod tests {
38344747 "last_ack_ts advances to the fresh ack's ts"
38354748 );
38364749 }
4750+
4751+ // -----------------------------------------------------------------------
4752+ // cursor restoration on budget exhaustion
4753+ // -----------------------------------------------------------------------
4754+
4755+ /// The cursor is advanced to `page.cursor` before the event loop runs. If the
4756+ /// poll-byte budget is exhausted mid-page, the loop breaks — but the cursor
4757+ /// must be restored to its pre-page value. Otherwise the next poll acks the
4758+ /// whole page to the relay, which deletes every event we skipped: permanent
4759+ /// message loss.
4760+ #[test]
4761+ fn a_budget_exhausted_mid_page_restores_the_cursor() {
4762+ let app = Skrepka;
4763+ let mut m = with_identity();
4764+ let peer = peer_hex(9);
4765+ m.contacts
4766+ .insert(peer.clone(), Contact::new(peer.clone(), String::new(), 0));
4767+ m.cursor = 100;
4768+
4769+ // Two blobs, each at the per-blob ceiling (MAX_BLOB_LEN hex chars =
4770+ // MAX_BLOB_LEN/2 decoded bytes = 21 MiB). Individually they pass the
4771+ // per-blob size gate; together they exceed the 64 MiB page budget.
4772+ // The first passes the budget check (21 < 64) and fails decrypt (it's
4773+ // not a real blob); the second fails the budget check (21 + 21 + 21 > 64).
4774+ // Three blobs to be sure: 3 × 21 = 63 MiB (just under 64), so use four.
4775+ let big_hex = "ab".repeat(MAX_BLOB_LEN); // MAX_BLOB_LEN hex chars = MAX_BLOB_LEN/2 bytes
4776+ let page = PollResp {
4777+ events: (0..4)
4778+ .map(|_| PollEvent {
4779+ encrypted_blob: big_hex.clone(),
4780+ })
4781+ .collect(),
4782+ cursor: 200,
4783+ };
4784+ let _ = app.ingest_poll(&mut m, page);
4785+
4786+ assert_eq!(
4787+ m.cursor, 100,
4788+ "the cursor must be restored — the page was not fully processed"
4789+ );
4790+ assert!(
4791+ m.error.as_ref().is_some_and(|e| e.contains("oversized")),
4792+ "the user is told"
4793+ );
4794+ assert!(
4795+ m.messages.get(&peer).is_none_or(Vec::is_empty),
4796+ "nothing was stored from the oversized page"
4797+ );
4798+ }
4799+
4800+ // -----------------------------------------------------------------------
4801+ // send_text block check
4802+ // -----------------------------------------------------------------------
4803+
4804+ // -----------------------------------------------------------------------
4805+ // send_text body length check
4806+ // -----------------------------------------------------------------------
4807+
4808+ /// Blocking cuts a peer off in both directions. `send_text` must refuse to
4809+ /// enqueue a message to a blocked contact — otherwise the block only stops
4810+ /// their incoming mail while we keep sending them ours.
4811+ #[test]
4812+ fn send_text_to_a_blocked_peer_is_refused() {
4813+ let app = Skrepka;
4814+ let mut m = with_identity();
4815+ let peer = peer_hex(9);
4816+ let mut contact = Contact::new(peer.clone(), "Enemy".into(), 0);
4817+ contact.blocked = true;
4818+ m.contacts.insert(peer.clone(), contact);
4819+ m.active_peer = Some(peer.clone());
4820+ m.compose = "hello?".into();
4821+
4822+ let _ = app.update(Event::SendText, &mut m);
4823+
4824+ assert!(
4825+ m.outbox.is_empty(),
4826+ "no message is queued for a blocked peer"
4827+ );
4828+ assert!(
4829+ m.error.as_ref().is_some_and(|e| e.contains("blocked")),
4830+ "the user is told why"
4831+ );
4832+ // The compose buffer is not cleared — the user might want to unblock
4833+ // and resend, and losing their draft would be surprising.
4834+ assert_eq!(m.compose, "hello?");
4835+ }
4836+
4837+ /// A text body over `MAX_BODY_LEN` would be stored locally and sent to the
4838+ /// relay, but the recipient's `parse_payload` silently drops oversized text.
4839+ /// The sender would see it stuck as "sent" forever with no error.
4840+ #[test]
4841+ fn send_text_refuses_an_oversized_body() {
4842+ let app = Skrepka;
4843+ let mut m = with_identity();
4844+ let peer = peer_hex(9);
4845+ m.active_peer = Some(peer.clone());
4846+ m.compose = "x".repeat(protocol::MAX_BODY_LEN + 1);
4847+
4848+ let _ = app.update(Event::SendText, &mut m);
4849+
4850+ assert!(m.outbox.is_empty(), "nothing is queued");
4851+ assert!(m.messages.get(&peer).is_none_or(Vec::is_empty), "nothing stored");
4852+ assert!(!m.compose.is_empty(), "compose is preserved");
4853+ assert!(m.error.as_ref().is_some_and(|e| e.contains("too long")));
4854+ }
4855+
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+
4875+ // -----------------------------------------------------------------------
4876+ // SaveProfile field length checks
4877+ // -----------------------------------------------------------------------
4878+
4879+ #[test]
4880+ fn save_profile_rejects_an_oversized_display_name() {
4881+ let app = Skrepka;
4882+ let mut m = with_identity();
4883+
4884+ let _ = app.update(Event::SaveProfile {
4885+ display_name: "x".repeat(protocol::MAX_DISPLAY_NAME_LEN + 1),
4886+ bio: String::new(),
4887+ photo: None,
4888+ }, &mut m);
4889+
4890+ assert!(m.outbox.is_empty(), "no broadcast queued");
4891+ assert!(m.profile.display_name.is_empty(), "local profile not updated");
4892+ assert!(m.error.as_ref().is_some_and(|e| e.contains("display name")));
4893+ }
4894+
4895+ #[test]
4896+ fn save_profile_rejects_an_oversized_bio() {
4897+ let app = Skrepka;
4898+ let mut m = with_identity();
4899+
4900+ let _ = app.update(Event::SaveProfile {
4901+ display_name: "Alice".into(),
4902+ bio: "x".repeat(protocol::MAX_BIO_LEN + 1),
4903+ photo: None,
4904+ }, &mut m);
4905+
4906+ assert!(m.outbox.is_empty(), "no broadcast queued");
4907+ assert!(m.profile.bio.is_empty(), "local profile not updated");
4908+ assert!(m.error.as_ref().is_some_and(|e| e.contains("bio")));
4909+ }
4910+
4911+ #[test]
4912+ fn save_profile_rejects_an_oversized_photo() {
4913+ let app = Skrepka;
4914+ let mut m = with_identity();
4915+
4916+ let _ = app.update(Event::SaveProfile {
4917+ display_name: "Alice".into(),
4918+ bio: String::new(),
4919+ photo: Some("x".repeat(protocol::MAX_PHOTO_LEN + 1)),
4920+ }, &mut m);
4921+
4922+ assert!(m.outbox.is_empty(), "no broadcast queued");
4923+ assert!(m.profile.photo.is_none(), "local profile not updated");
4924+ assert!(m.error.as_ref().is_some_and(|e| e.contains("photo")));
4925+ }
4926+
4927+ // -----------------------------------------------------------------------
4928+ // AddContact nickname length check
4929+ // -----------------------------------------------------------------------
4930+
4931+ #[test]
4932+ fn add_contact_rejects_an_oversized_nickname() {
4933+ let app = Skrepka;
4934+ let mut m = with_identity();
4935+ let peer = peer_hex(9);
4936+
4937+ let _ = app.update(Event::AddContact {
4938+ input: peer.clone(),
4939+ nickname: "x".repeat(protocol::MAX_DISPLAY_NAME_LEN + 1),
4940+ }, &mut m);
4941+
4942+ assert!(!m.contacts.contains_key(&peer), "no contact created");
4943+ assert!(m.error.as_ref().is_some_and(|e| e.contains("nickname")));
4944+ }
4945+
4946+ // -----------------------------------------------------------------------
4947+ // FlushWatchdog unsticks a wedged send loop
4948+ // -----------------------------------------------------------------------
4949+
4950+ /// `flushing` is only cleared by `SendResult`. If the shell drops the HTTP
4951+ /// effect, the outbox is permanently stuck. The watchdog releases the guard
4952+ /// so the flush retries.
4953+ #[test]
4954+ fn the_flush_watchdog_unwedges_a_send_that_never_resolved() {
4955+ let app = Skrepka;
4956+ let mut m = with_identity();
4957+ m.token = Some("t".into());
4958+ let peer = peer_hex(9);
4959+ m.outbox.push_back(queued(&peer, "stuck"));
4960+
4961+ let _ = app.update(Event::StartFlush, &mut m);
4962+ assert!(m.flushing, "the send is in flight");
4963+ assert!(m.in_flight.is_some());
4964+
4965+ // The send never comes back. The watchdog fires, clears the stuck guard,
4966+ // and retries the flush — the message goes out again.
4967+ let mut cmd = app.update(Event::FlushWatchdog(m.flush_gen), &mut m);
4968+
4969+ assert!(m.flushing, "a fresh send is in flight (the retry)");
4970+ assert!(m.in_flight.is_some(), "in_flight is set for the retry");
4971+ let batches = sent_batches(&mut cmd);
4972+ assert_eq!(batches.len(), 1, "the message is re-sent");
4973+ }
4974+
4975+ /// A watchdog that fires after the send already completed is a no-op.
4976+ #[test]
4977+ fn a_stale_flush_watchdog_leaves_a_completed_send_alone() {
4978+ let app = Skrepka;
4979+ let mut m = with_identity();
4980+ m.token = Some("t".into());
4981+ m.outbox.push_back(queued(&peer_hex(9), "hi"));
4982+
4983+ let _ = app.update(Event::StartFlush, &mut m);
4984+ let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
4985+ assert!(!m.flushing);
4986+ assert!(m.outbox.is_empty());
4987+
4988+ // The watchdog for the completed send fires late.
4989+ let _ = app.update(Event::FlushWatchdog(m.flush_gen), &mut m);
4990+ assert!(!m.flushing, "a completed send is not re-armed");
4991+ assert!(m.outbox.is_empty());
4992+ }
4993+
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.
4999+ #[test]
5000+ fn a_late_send_result_after_the_watchdog_is_dropped_by_generation() {
5001+ let app = Skrepka;
5002+ let mut m = with_identity();
5003+ m.token = Some("t".into());
5004+ let peer = peer_hex(9);
5005+ m.outbox.push_back(queued(&peer, "stuck"));
5006+
5007+ 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);
5011+ 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");
5025+ assert!(!m.flushing, "and the guard is released");
5026+ }
38375027}