@@ -9,9 +9,6 @@ use std::collections::{BTreeMap, HashMap, HashSet, VecDeque};
99use std::sync::Arc;
1010use std::time::{Duration, SystemTime, UNIX_EPOCH};
1111
12−use base64::prelude::*;
13−use base64::Engine;
14−
1512use crux_core::{
1613 macros::effect,
1714 render::{render, RenderOperation},
@@ -61,24 +58,6 @@ const MAX_CONTACTS: usize = 500;
6158/// the budget is dropped exactly like an undecryptable blob would be.
6259const MAX_POLL_TOTAL_BYTES: usize = 64 * 1024 * 1024;
6360
64−/// Cap on the number of events we will attempt to decrypt out of a single poll
65−/// page, independent of the byte budget. Minimum-size blobs (173 bytes decoded)
66−/// would otherwise permit ~388K events within the 64 MiB byte budget, each
67−/// costing a full scalar mult + SHA-512 + AEAD pass — enough to freeze the app
68−/// for tens of seconds per poll. An honest relay caps at 50 events
69−/// (`maxPollEvents`); this is a hostile-relay bound. Set above `MAX_ACK_IDS`
70−/// (1024) so a legitimate large page from one sender still fully processes.
71−const MAX_POLL_EVENTS: usize = 2000;
72−
73−/// Upper bound on a poll response body we will buffer and parse, in bytes.
74−///
75−/// Before the per-blob byte/count budgets in `ingest_poll` ever run, the client
76−/// reads the entire response body into a `Vec<u8>` and deserializes it. A
77−/// multi-gigabyte response from a hostile relay would OOM the app before
78−/// `ingest_poll`'s defenses are consulted. This cap (generously above what 50
79−/// max-size blobs + JSON envelope could cost) bounds the allocation.
80−const MAX_POLL_RESPONSE_BYTES: usize = 64 * 1024 * 1024;
81−
8261/// How long a poll may stay in flight before we assume the shell lost it.
8362///
8463/// `polling` gates the whole long-poll loop, and it is only cleared by a
@@ -94,26 +73,6 @@ const MAX_POLL_RESPONSE_BYTES: usize = 64 * 1024 * 1024;
9473/// relay, so leave the margin.
9574const POLL_WATCHDOG_MS: u64 = 90_000;
9675
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−
11776/// Messages per `/messages` request.
11877///
11978/// The relay rejects a longer batch outright (`413 batch_too_large`,
@@ -147,18 +106,10 @@ const K_PROFILE: &str = "profile";
147106const K_CONTACTS: &str = "contacts";
148107const K_CURSOR: &str = "cursor";
149108const K_OUTBOX: &str = "outbox";
150−const K_SEEN_IDS: &str = "seen_ids";
151109fn k_messages(peer: &str) -> String {
152110 format!("messages:{peer}")
153111}
154112
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−
162113// ---------------------------------------------------------------------------
163114// Effect
164115// ---------------------------------------------------------------------------
@@ -226,9 +177,6 @@ pub enum Event {
226177 LoadedOutbox(#[facet(opaque)] KvData),
227178 #[serde(skip)]
228179 #[facet(skip)]
229− LoadedSeenIds(#[facet(opaque)] KvData),
230− #[serde(skip)]
231− #[facet(skip)]
232180 LoadedMessages(String, #[facet(opaque)] KvData),
233181 #[serde(skip)]
234182 #[facet(skip)]
@@ -262,17 +210,6 @@ pub enum Event {
262210 #[serde(skip)]
263211 #[facet(skip)]
264212 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),
276213 #[serde(skip)]
277214 #[facet(skip)]
278215 ChallengeResult(u64, #[facet(opaque)] HttpResult),
@@ -284,7 +221,7 @@ pub enum Event {
284221 PollResult(u64, #[facet(opaque)] HttpResult),
285222 #[serde(skip)]
286223 #[facet(skip)]
287− SendResult(u64, #[facet(opaque)] HttpResult),
224+ SendResult(#[facet(opaque)] HttpResult),
288225}
289226
290227// ---------------------------------------------------------------------------
@@ -310,9 +247,9 @@ pub enum Page {
310247 EditProfile,
311248}
312249
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;
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;
316253
317254/// The batch a live `/messages` request is carrying: the first `count` items of
318255/// the outbox, all bound for `recipient`.
@@ -338,12 +275,6 @@ pub struct Model {
338275 profile: OwnProfile,
339276 contacts: BTreeMap<String, Contact>,
340277 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>>,
347278 outbox: VecDeque<OutboxItem>,
348279 cursor: i64,
349280 token: Option<String>,
@@ -386,12 +317,6 @@ pub struct Model {
386317 /// items it sent off the head of the outbox, so two overlapping sends would
387318 /// pop each other's messages.
388319 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,
395320 /// What the in-flight send will pop off the head of the outbox when it
396321 /// succeeds — see `InFlight`. `None` while nothing is in flight, and also
397322 /// while a send is in flight whose items have since been removed from under
@@ -456,7 +381,6 @@ impl Default for Model {
456381 profile: OwnProfile::default(),
457382 contacts: BTreeMap::new(),
458383 messages: BTreeMap::new(),
459− seen_ids: BTreeMap::new(),
460384 outbox: VecDeque::new(),
461385 cursor: 0,
462386 token: None,
@@ -468,7 +392,6 @@ impl Default for Model {
468392 authenticating: false,
469393 auth_gen: 0,
470394 flushing: false,
471− flush_gen: 0,
472395 in_flight: None,
473396 send_batch_budget: None,
474397 loads_pending: 0,
@@ -520,24 +443,6 @@ impl Model {
520443 self.auth_gen
521444 }
522445
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−
541446 /// Abandon the in-flight auth round-trip (either leg).
542447 fn abandon_auth(&mut self) {
543448 self.authenticating = false;
@@ -584,7 +489,7 @@ fn now_ms() -> i64 {
584489 SystemTime::now()
585490 .duration_since(UNIX_EPOCH)
586491 .map(|d| d.as_millis() as i64)
587− .unwrap_or(1)
492+ .unwrap_or(0)
588493}
589494
590495/// Bare lowercased hostname of a server URL (no scheme/port/trailing dot).
@@ -672,16 +577,6 @@ struct PollEvent {
672577 #[serde(rename = "encryptedBlob")]
673578 encrypted_blob: String,
674579}
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−}
685580#[derive(Serialize)]
686581struct SendBatch {
687582 messages: Vec<Envelope>,
@@ -725,7 +620,6 @@ impl App for Skrepka {
725620 KeyValue::get(K_CONTACTS).then_send(Event::LoadedContacts),
726621 KeyValue::get(K_CURSOR).then_send(Event::LoadedCursor),
727622 KeyValue::get(K_OUTBOX).then_send(Event::LoadedOutbox),
728− KeyValue::get(K_SEEN_IDS).then_send(Event::LoadedSeenIds),
729623 ])
730624 .and(render())
731625 }
@@ -803,51 +697,10 @@ impl App for Skrepka {
803697 }
804698 self.startup_done(model).and(render())
805699 }
806− Event::LoadedSeenIds(res) => {
807− match parse_kv::<BTreeMap<String, HashSet<String>>>(res) {
808− Ok(Some(map)) => model.seen_ids = map,
809− Ok(None) => {}
810− Err(()) => {
811− // Non-fatal: dedup falls back to the conversation-derived
812− // set, which is weaker but functional.
813− model.seen_ids = BTreeMap::new();
814− }
815− }
816− self.startup_done(model).and(render())
817− }
818700 Event::LoadedMessages(peer, res) => {
819701 match parse_kv::<Vec<StoredMessage>>(res) {
820702 Ok(Some(list)) => {
821− // D5: Add on-disk message IDs to seen_ids so future
822− // replays don't pass dedup. Without this, a seen_ids
823− // kv load failure leaves the set empty, and on-disk
824− // messages aren't protected against replay.
825− if let Some(id_set) = model.seen_ids.get_mut(&peer) {
826− id_set.extend(list.iter().map(|m| m.id.clone()));
827− }
828− // Merge: a poll that beat this lazy load may have already
829− // inserted messages for this peer. Overwriting with the
830− // on-disk list (which the GET read before the poll wrote)
831− // would silently destroy the freshly-ingested messages.
832− // Union by id, keeping both the on-disk history and any
833− // in-memory additions.
834− match model.messages.get_mut(&peer) {
835− Some(existing) => {
836− let existing_ids: HashSet<&String> =
837− existing.iter().map(|m| &m.id).collect();
838− let new_msgs: Vec<StoredMessage> = list
839− .into_iter()
840− .filter(|m| !existing_ids.contains(&m.id))
841− .collect();
842− existing.extend(new_msgs);
843− // Re-sort by ts to maintain the invariant.
844− existing.sort_by_key(|m| m.ts);
845− trim_history(existing);
846− }
847− None => {
848− model.messages.insert(peer, list);
849− }
850− }
703+ model.messages.insert(peer, list);
851704 }
852705 Ok(None) => {}
853706 Err(()) => {
@@ -957,20 +810,11 @@ impl App for Skrepka {
957810 // would then ack a sequence number B never issued, and B would
958811 // drop every message waiting for us.
959812 model.abandon_poll();
960− // ...and retire the send that is still in flight *to relay A*,
961− // for the same reason: its `SendResult` carries A's acceptance,
962− // and popping items here would eat whatever the new relay's
963− // send is about to carry. The generation check in `SendResult`
964− // is what makes the late result a no-op.
965− model.abandon_flush();
966813 // The old relay's rate limit was the old relay's opinion. A new one is
967814 // entitled to take our sends immediately, and carrying the pause over
968815 // would park the outbox against a limit that no longer applies.
969816 model.flush_paused_until = 0;
970817 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;
974818 let writes = if self.refuse_write(model) {
975819 Command::done()
976820 } else {
@@ -1018,15 +862,6 @@ impl App for Skrepka {
1018862 }
1019863 }
1020864 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− }
1030865 match phonemic::try_parse_pubkey(&input) {
1031866 Some(hex) if hex != model.my_pubkey => {
1032867 model
@@ -1056,9 +891,6 @@ impl App for Skrepka {
1056891 if let Some(c) = model.contacts.get_mut(&peer) {
1057892 c.blocked = blocked;
1058893 }
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;
1062894 if blocked {
1063895 // Gating *ingest* on the block only stops the acks we have not
1064896 // queued yet. Anything already sitting in the outbox for this
@@ -1077,7 +909,6 @@ impl App for Skrepka {
1077909 Event::DeleteContact { peer } => {
1078910 model.contacts.remove(&peer);
1079911 model.messages.remove(&peer);
1080− model.seen_ids.remove(&peer);
1081912 model.outbox.retain(|item| item.recipient != peer);
1082913 // The send in flight may be carrying the very items we just
1083914 // dropped; if so its result must pop nothing.
@@ -1094,8 +925,6 @@ impl App for Skrepka {
1094925 // message the user just asked us to forget. Delete the key.
1095926 KeyValue::delete(k_messages(&peer)).then_send(Event::Saved),
1096927 self.persist_outbox(model),
1097− // Q18: persist the seen_ids map without the removed peer.
1098− self.persist_seen_ids(model),
1099928 ])
1100929 .and(render())
1101930 }
@@ -1104,32 +933,6 @@ impl App for Skrepka {
1104933 bio,
1105934 photo,
1106935 } => {
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− }
1133936 model.profile = OwnProfile {
1134937 display_name,
1135938 bio,
@@ -1138,8 +941,8 @@ impl App for Skrepka {
1138941 model.page = Page::Conversations;
1139942 let ts = now_ms();
1140943 let payload = Payload::Profile {
1141− display_name: Some(model.profile.display_name.clone()),
1142− bio: Some(model.profile.bio.clone()),
944+ display_name: model.profile.display_name.clone(),
945+ bio: model.profile.bio.clone(),
1143946 photo: model.profile.photo.clone(),
1144947 };
1145948 // Broadcast the new profile to every contact we haven't blocked.
@@ -1218,11 +1021,6 @@ impl App for Skrepka {
12181021 let gen = model.next_auth_gen();
12191022 req.build()
12201023 .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− )
12261024 .and(render())
12271025 }
12281026 Err(_) => {
@@ -1401,90 +1199,22 @@ impl App for Skrepka {
14011199 return render();
14021200 }
14031201 let bytes = resp.take_body().unwrap_or_default();
1404− // Bound the response before parsing: a hostile relay can return
1405− // a multi-gigabyte body that OOMs the app before `ingest_poll`'s
1406− // per-blob budgets are even consulted.
1407− if bytes.len() > MAX_POLL_RESPONSE_BYTES {
1408− model.error = Some("relay sent an oversized poll response".into());
1409− return self.backoff_poll(model);
1410− }
1411− // Q25: A malformed or unparseable poll response must not be
1412− // treated as an empty page — `unwrap_or_default()` would adopt
1413− // cursor 0, forcing a re-fetch from the beginning and wasting
1414− // bandwidth on already-seen (deduped) messages. Back off instead.
1415− let parsed: PollResp = match serde_json::from_slice(&bytes) {
1416− Ok(p) => p,
1417− Err(_) => {
1418− model.error = Some("relay sent a malformed poll response".into());
1419− return self.backoff_poll(model);
1420− }
1421− };
1422− let cursor_before = model.cursor;
1202+ let parsed: PollResp = serde_json::from_slice(&bytes).unwrap_or_default();
14231203 let cmd = self.ingest_poll(model, parsed);
14241204 // The re-poll hangs off the *cursor write*, not off this event.
14251205 // Polling again is what acks the batch we just took — the server
14261206 // deletes everything up to `cursor` — so it must not happen until
14271207 // the batch is durable on our side. `SavedCursor` re-polls.
1428− //
1429− // Arm a watchdog alongside the cursor write: if the shell ever
1430− // drops the KV effect (suspension, bug), `SavedCursor` never fires,
1431− // `polling` stays `false`, and the loop is dead until the next
1432− // foreground `Connect`. The watchdog fires `Poll` if the loop is
1433− // still idle — same pattern as the poll/send HTTP watchdogs.
1434− //
1435− // Skip the cursor write when the cursor did not advance (an empty
1436− // or fully-deduped page): there is nothing new to persist, so
1437− // re-poll directly. The watchdog is still armed as a fallback.
1438− let cursor_advanced = model.cursor != cursor_before;
1439− cmd.and(if cursor_advanced {
1208+ cmd.and(
14401209 KeyValue::set(K_CURSOR, json_bytes(&model.cursor))
1441− .then_send(Event::SavedCursor)
1442− } else {
1443− Command::event(Event::Poll)
1444− })
1445− .and(
1446− Time::notify_after(Duration::from_millis(POLL_WATCHDOG_MS))
1447− .0
1448− .then_send(|_| Event::Poll),
1210+ .then_send(Event::SavedCursor),
14491211 )
14501212 .and(render())
14511213 }
14521214
14531215 // ---------------- outbox ----------------
14541216 Event::StartFlush => self.flush_next(model),
1455− Event::FlushWatchdog(gen) => {
1456− // Only the flush this watchdog was armed for, and only while it
1457− // is still the current one and still in flight. A watchdog whose
1458− // send already completed (or was superseded) is a stale timer and
1459− // must not touch a healthy successor — same pattern as
1460− // `PollWatchdog`.
1461− if gen != model.flush_gen || !model.flushing {
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− }
1217+ Event::SendResult(Ok(resp)) => {
14881218 // The in-flight send is over; `flush_next` refuses to run while
14891219 // this is set, so it must be cleared before *any* branch below.
14901220 model.flushing = false;
@@ -1574,24 +1304,16 @@ impl App for Skrepka {
15741304 // later, under load that has nothing to do with this one — starts
15751305 // at whatever height the last one climbed to.
15761306 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− }
1307+ for _ in 0..batch.map_or(0, |b| b.count) {
1308+ model.outbox.pop_front();
15841309 }
15851310 return self.persist_outbox(model).and(self.flush_next(model));
15861311 }
15871312 // A transient failure: a 5xx, a rate limit, a relay restarting.
15881313 self.retry_batch(model, batch)
15891314 }
1590− Event::SendResult(gen, Err(_)) => {
1315+ Event::SendResult(Err(_)) => {
15911316 // A transport error is the same transient failure as a 5xx.
1592− if gen != model.flush_gen {
1593− return Command::done();
1594− }
15951317 model.flushing = false;
15961318 let batch = model.in_flight.take();
15971319 self.retry_batch(model, batch)
@@ -1766,16 +1488,6 @@ impl Skrepka {
17661488 KeyValue::set(k_messages(peer), json_bytes(&list)).then_send(Event::Saved)
17671489 }
17681490
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−
17791491 fn schedule_reconnect(&self, model: &mut Model) -> Command<Effect, Event> {
17801492 let delay = backoff_ms(model.poll_retries);
17811493 model.poll_retries = model.poll_retries.saturating_add(1);
@@ -1806,9 +1518,6 @@ impl Skrepka {
18061518 let Some(peer) = model.active_peer.clone() else {
18071519 return render();
18081520 };
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;
18121521 // Blocking cuts the peer off in both directions (PROTOCOL.md §4). The
18131522 // ingest side drops their messages; the send side must refuse too, or
18141523 // a blocked peer still receives our texts, acks, and the liveness
@@ -1821,15 +1530,6 @@ impl Skrepka {
18211530 if body.is_empty() {
18221531 return render();
18231532 }
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− }
18331533
18341534 // Check the contact cap *before* mutating anything. The old order
18351535 // cleared `compose` and inserted the message into `model.messages`
@@ -1874,21 +1574,14 @@ impl Skrepka {
18741574 .insert(peer.clone(), Contact::new(peer.clone(), String::new(), ts));
18751575 }
18761576
1877− let text = Payload::Text { id, body };
1878− model.outbox.push_back(OutboxItem::new(
1879− peer.clone(),
1880− Arc::new(protocol::serialize_payload(&text, ts)),
1881− ));
1882−
1883− // On first contact, share our profile *after* the text. The text
1884− // auto-creates the contact on the recipient's side; the profile that
1885− // follows in the same batch then applies to the now-existing contact.
1886− // (Previously profile was queued before the text, but the recipient's
1887− // ingest drops profiles from non-contacts, so the profile was lost.)
1577+ // On first contact, share our profile *before* the text so the
1578+ // recipient's first impression is a name and avatar, not a truncated
1579+ // @p that gets replaced a moment later. Both are queued behind any
1580+ // in-flight items, so the ordering is preserved through the outbox.
18881581 if was_new {
18891582 let profile = Payload::Profile {
1890− display_name: Some(model.profile.display_name.clone()),
1891− bio: Some(model.profile.bio.clone()),
1583+ display_name: model.profile.display_name.clone(),
1584+ bio: model.profile.bio.clone(),
18921585 photo: model.profile.photo.clone(),
18931586 };
18941587 model.outbox.push_back(OutboxItem::profile(
@@ -1897,6 +1590,12 @@ impl Skrepka {
18971590 ));
18981591 }
18991592
1593+ let text = Payload::Text { id, body };
1594+ model.outbox.push_back(OutboxItem::new(
1595+ peer.clone(),
1596+ Arc::new(protocol::serialize_payload(&text, ts)),
1597+ ));
1598+
19001599 Command::all([
19011600 self.persist_messages(model, &peer),
19021601 self.persist_contacts(model),
@@ -2024,7 +1723,6 @@ impl Skrepka {
20241723 };
20251724
20261725 model.flushing = true;
2027− let gen = model.next_flush_gen();
20281726 model.in_flight = Some(InFlight {
20291727 recipient,
20301728 count: messages.len(),
@@ -2035,12 +1733,7 @@ impl Skrepka {
20351733 .header("authorization", format!("Bearer {token}"))
20361734 .body_json(&SendBatch { messages })
20371735 {
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− ),
1736+ Ok(req) => persist.and(req.build().then_send(Event::SendResult)),
20441737 Err(_) => {
20451738 // Not a dead loop: the next StartFlush (from a poll page or a
20461739 // fresh send) retries the head of the outbox.
@@ -2110,20 +1803,8 @@ impl Skrepka {
21101803 return Command::done();
21111804 }
21121805 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;
1806+ if page.cursor > model.cursor {
1807+ model.cursor = page.cursor;
21271808 }
21281809 let Some(id) = model.identity() else {
21291810 return Command::done();
@@ -2132,24 +1813,12 @@ impl Skrepka {
21321813 let now = now_ms();
21331814 let mut touched_convos: HashSet<String> = HashSet::new();
21341815 let mut contacts_dirty = false;
2135− let mut outbox_dirty = false;
21361816 // sender hex -> message ids needing acks
21371817 let mut ack_targets: BTreeMap<String, Vec<String>> = BTreeMap::new();
2138− // Q17: Per-sender deferred ack accumulation. Collected during the event
2139− // loop, applied after it — so relay page ordering cannot cause an older
2140− // ack to be dropped as stale by a newer one processed first.
2141− let mut pending_acks: HashMap<String, PendingAck> = HashMap::new();
1818+ // Lazily-built per-sender sets of known message IDs for O(1) text dedup.
1819+ let mut seen_ids: HashMap<String, HashSet<String>> = HashMap::new();
21421820
21431821 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;
21531822 for ev in page.events {
21541823 // Reject an oversized blob on its hex length, before it costs us a
21551824 // decode. No conforming sender produces one (the relay caps blobs at
@@ -2157,37 +1826,19 @@ impl Skrepka {
21571826 if ev.encrypted_blob.len() > MAX_BLOB_LEN * 2 {
21581827 continue;
21591828 }
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;
21771829 match budget.checked_sub(ev.encrypted_blob.len() / 2) {
21781830 Some(left) => budget = left,
21791831 None => {
2180− // N9: Budget exhausted. If we stored at least one event,
2181− // keep the advanced cursor to ack the page — re-fetching
2182− // an oversized page forever is a livelock. The unprocessed
2183− // events are beyond our capacity and would be re-served
2184− // on every poll, wedging the client. Acking them breaks
2185− // the loop. If zero events were stored (completely hostile
2186− // page), restore the cursor to avoid acking mail we
2187− // never saw.
2188− if events_stored == 0 {
2189− model.cursor = prev_cursor;
2190− }
1832+ // A page this large is a misbehaving relay, not real mail.
1833+ // Stop here; the rest of the page is discarded exactly as an
1834+ // undecryptable blob would be.
1835+ //
1836+ // Restore the cursor: it was advanced to `page.cursor` above,
1837+ // but we have not processed the remaining events. If we keep
1838+ // the advanced cursor, the next poll acks the whole page to
1839+ // the relay — including every event we skipped — and the
1840+ // relay deletes them. Those messages are permanently lost.
1841+ model.cursor = prev_cursor;
21911842 model.error = Some("relay sent an oversized poll page".into());
21921843 break;
21931844 }
@@ -2201,10 +1852,6 @@ impl Skrepka {
22011852 let Some(parsed) = protocol::parse_payload(&dec.plaintext) else {
22021853 continue;
22031854 };
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;
22081855 let sender = dec.sender_hex;
22091856
22101857 // Mail from ourselves. A relay cannot forge this — the sender identity
@@ -2256,35 +1903,12 @@ impl Skrepka {
22561903 Contact::new(sender.clone(), String::new(), ts),
22571904 );
22581905 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;
22711906 }
22721907 let convo = model.messages.entry(sender.clone()).or_default();
2273− // Q18: Use the persistent per-peer dedup set from
2274− // `model.seen_ids` (loaded from kv at startup) instead of
2275− // deriving it from the trimmed conversation. Without this,
2276− // a replayed blob whose `id` was aged out by `trim_history`
2277− // (past the 1000-message cap) passes the dedup check and
2278− // re-enters the conversation.
2279− let id_set = model
2280− .seen_ids
2281− .entry(sender.clone())
2282− .or_default();
2283− // Also seed from the conversation in case the kv load was
2284− // missing or partial — covers the upgrade path.
2285− if id_set.is_empty() {
2286− id_set.extend(convo.iter().map(|m| m.id.clone()));
2287− }
1908+ // Lazily build the per-sender ID set on first touch.
1909+ let id_set = seen_ids.entry(sender.clone()).or_insert_with(|| {
1910+ convo.iter().map(|m| m.id.clone()).collect()
1911+ });
22881912 // dedup by id — O(1) via HashSet
22891913 if !id_set.contains(&msg_id) {
22901914 // The conversation is *kept* sorted rather than re-sorted:
@@ -2302,29 +1926,6 @@ impl Skrepka {
23021926 },
23031927 );
23041928 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− }
23281929 trim_history(convo);
23291930 touched_convos.insert(sender.clone());
23301931 // Only ack a message we actually stored — a duplicate id
@@ -2333,26 +1934,39 @@ impl Skrepka {
23331934 }
23341935 }
23351936 Payload::DeliveryAck { ack_ids } => {
2336− // Q17: Defer the staleness check until after the event loop.
2337− // Within a single page, a relay can reorder events so that a
2338− // newer ack (higher ts) is processed before an older one. If
2339− // we applied the watermark immediately, the older ack would be
2340− // dropped as stale — even if it acks different message IDs.
2341− // Instead, accumulate all ack_ids per sender and the max ts,
2342− // then apply them in a second pass below the loop.
2343− let prev_ts = pending_acks
1937+ // Staleness check: drop a replayed ack whose ts predates the
1938+ // last one we processed for this contact.
1939+ let stale = model
1940+ .contacts
23441941 .get(&sender)
2345− .map(|p| p.ts)
2346− .unwrap_or(0);
2347− if ts > prev_ts {
2348− let entry = pending_acks.entry(sender.clone()).or_default();
2349− entry.ts = ts;
1942+ .is_some_and(|c| ts < c.last_ack_ts);
1943+ if stale {
1944+ continue;
1945+ }
1946+ // The watermark is clamped to *now*, not merely into the skew
1947+ // window: `ts` is the peer's to choose, and `last_ack_ts` is
1948+ // what every later ack is measured against. A single ack at
1949+ // `ts = i64::MAX` would otherwise park the watermark a minute
1950+ // into the future, and every honest ack that followed — the
1951+ // ones that actually mark our messages delivered — would read
1952+ // as stale and be dropped.
1953+ if let Some(c) = model.contacts.get_mut(&sender) {
1954+ c.last_ack_ts = ts.min(now);
1955+ contacts_dirty = true;
1956+ }
1957+ let ack_set: HashSet<String> = ack_ids.iter().cloned().collect();
1958+ if let Some(convo) = model.messages.get_mut(&sender) {
1959+ let mut changed = false;
1960+ for m in convo.iter_mut() {
1961+ if m.outgoing && ack_set.contains(&m.id) && !m.delivered {
1962+ m.delivered = true;
1963+ changed = true;
1964+ }
1965+ }
1966+ if changed {
1967+ touched_convos.insert(sender);
1968+ }
23501969 }
2351− pending_acks
2352− .entry(sender.clone())
2353− .or_default()
2354− .ids
2355− .extend(ack_ids);
23561970 }
23571971 Payload::Profile {
23581972 display_name,
@@ -2379,17 +1993,9 @@ impl Skrepka {
23791993 // A replayed profile at the same `ts` overwrites with identical
23801994 // content, which is a no-op in practice.
23811995 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− }
1996+ c.display_name = display_name;
1997+ c.bio = bio;
1998+ c.photo = photo;
23931999 c.last_profile_ts = ts.min(now);
23942000 contacts_dirty = true;
23952001 }
@@ -2397,42 +2003,6 @@ impl Skrepka {
23972003 }
23982004 }
23992005
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−
24362006 // Queue delivery acks, in batches the recipient will actually accept.
24372007 //
24382008 // `parse_payload` drops a `delivery.ack` carrying more than `MAX_ACK_IDS`
@@ -2455,7 +2025,6 @@ impl Skrepka {
24552025 Arc::new(protocol::serialize_payload(&payload, ack_ts)),
24562026 ));
24572027 }
2458− outbox_dirty = true;
24592028 }
24602029
24612030 // Persist everything touched.
@@ -2463,16 +2032,10 @@ impl Skrepka {
24632032 for peer in &touched_convos {
24642033 cmds.push(self.persist_messages(model, peer));
24652034 }
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− }
24702035 if contacts_dirty {
24712036 cmds.push(self.persist_contacts(model));
24722037 }
2473− if outbox_dirty {
2474− cmds.push(self.persist_outbox(model));
2475− }
2038+ cmds.push(self.persist_outbox(model));
24762039 cmds.push(Command::event(Event::StartFlush));
24772040 Command::all(cmds)
24782041 }
@@ -2545,7 +2108,7 @@ mod tests {
25452108 use crate::model::{MAX_OUTBOX_RETRIES, OUTBOX_TTL_MS};
25462109
25472110 fn with_identity() -> Model {
2548− let id = Identity::from_seed(&[3u8; 32]).unwrap();
2111+ let id = Identity::from_seed(&[3u8; 32]);
25492112 Model {
25502113 secret_key: Some(Zeroizing::new(id.secret_key.to_vec())),
25512114 my_pubkey: id.public_key_hex(),
@@ -2554,7 +2117,7 @@ mod tests {
25542117 }
25552118
25562119 fn peer_hex(seed: u8) -> String {
2557− Identity::from_seed(&[seed; 32]).unwrap().public_key_hex()
2120+ Identity::from_seed(&[seed; 32]).public_key_hex()
25582121 }
25592122
25602123 /// `Command::event` *queues* an event; it does not run it. The runtime drains
@@ -2577,7 +2140,7 @@ mod tests {
25772140
25782141 /// A poll page carrying one payload, encrypted from `sender_seed` to `me`.
25792142 fn page_from(sender_seed: u8, me: &Model, payload: &Payload, ts: i64) -> PollResp {
2580− let sender = Identity::from_seed(&[sender_seed; 32]).unwrap();
2143+ let sender = Identity::from_seed(&[sender_seed; 32]);
25812144 let recipient = Identity::from_secret_bytes(me.secret_key.as_ref().unwrap()).unwrap();
25822145 let json = protocol::serialize_payload(payload, ts);
25832146 let blob = crate::crypto::encrypt(
@@ -2666,8 +2229,7 @@ mod tests {
26662229 /// One full send round-trip: flush the outbox, then answer with `status`.
26672230 fn send_and_answer(app: &Skrepka, m: &mut Model, status: u16) {
26682231 let _ = app.update(Event::StartFlush, m);
2669− let gen = m.flush_gen;
2670− let _ = app.update(Event::SendResult(gen, Ok(ok_response(status))), m);
2232+ let _ = app.update(Event::SendResult(Ok(ok_response(status))), m);
26712233 }
26722234
26732235 // -----------------------------------------------------------------------
@@ -2696,7 +2258,7 @@ mod tests {
26962258 // The last one exhausts the budget: the head goes, and what was stuck
26972259 // behind it goes out.
26982260 let _ = app.update(Event::StartFlush, &mut m);
2699− let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(503))), &mut m);
2261+ let mut cmd = app.update(Event::SendResult(Ok(ok_response(503))), &mut m);
27002262
27012263 assert_eq!(m.outbox.len(), 1, "the undeliverable item is dropped");
27022264 assert_eq!(m.outbox[0].recipient, behind, "and the queue moves on");
@@ -2815,7 +2377,7 @@ mod tests {
28152377
28162378 let _ = app.update(Event::StartFlush, &mut m);
28172379 let mut cmd = app.update(
2818− Event::SendResult(m.flush_gen, Ok(response_with_header(429, "retry-after", "2"))),
2380+ Event::SendResult(Ok(response_with_header(429, "retry-after", "2"))),
28192381 &mut m,
28202382 );
28212383
@@ -2846,7 +2408,7 @@ mod tests {
28462408 m.poll_retries = 0;
28472409
28482410 let _ = app.update(Event::StartFlush, &mut m);
2849− let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(429))), &mut m);
2411+ let mut cmd = app.update(Event::SendResult(Ok(ok_response(429))), &mut m);
28502412
28512413 assert_eq!(m.outbox[0].retries, 0, "still not the message's fault");
28522414 assert_eq!(
@@ -2859,7 +2421,7 @@ mod tests {
28592421 // A second 429 climbs, rather than restarting at 3 s.
28602422 m.flush_paused_until = 0;
28612423 let _ = app.update(Event::StartFlush, &mut m);
2862− let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(429))), &mut m);
2424+ let mut cmd = app.update(Event::SendResult(Ok(ok_response(429))), &mut m);
28632425 assert_eq!(scheduled_delays(&mut cmd), vec![backoff_ms(1)], "it climbs");
28642426 assert_eq!(m.send_retries, 2);
28652427 }
@@ -2877,7 +2439,7 @@ mod tests {
28772439
28782440 let _ = app.update(Event::StartFlush, &mut m);
28792441 let _ = app.update(
2880− Event::SendResult(m.flush_gen, Ok(response_with_header(429, "retry-after", "2"))),
2442+ Event::SendResult(Ok(response_with_header(429, "retry-after", "2"))),
28812443 &mut m,
28822444 );
28832445 assert!(
@@ -2933,7 +2495,7 @@ mod tests {
29332495
29342496 let _ = app.update(Event::StartFlush, &mut m);
29352497 let mut cmd = app.update(
2936− Event::SendResult(m.flush_gen, Ok(response_with_header(429, "retry-after", "999999999"))),
2498+ Event::SendResult(Ok(response_with_header(429, "retry-after", "999999999"))),
29372499 &mut m,
29382500 );
29392501
@@ -2968,7 +2530,7 @@ mod tests {
29682530 assert!(batches[0].iter().all(|e| e.to == peer));
29692531
29702532 // ...and its success clears the whole batch, not just the head.
2971− let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
2533+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
29722534 assert!(m.outbox.is_empty());
29732535 }
29742536
@@ -2993,7 +2555,7 @@ mod tests {
29932555 assert_eq!(batches[0].len(), 2, "only the run at the head");
29942556 assert!(batches[0].iter().all(|e| e.to == a));
29952557
2996− let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
2558+ let mut cmd = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
29972559 assert_eq!(m.outbox.len(), 2);
29982560 let batches = sent_batches(&mut cmd);
29992561 assert_eq!(batches[0].len(), 1, "then b's single message");
@@ -3016,7 +2578,7 @@ mod tests {
30162578 let batches = sent_batches(&mut cmd);
30172579 assert_eq!(batches[0].len(), MAX_SEND_BATCH);
30182580
3019− let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
2581+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
30202582 assert_eq!(m.outbox.len(), 20, "the rest stays queued for the next batch");
30212583 }
30222584
@@ -3078,7 +2640,7 @@ mod tests {
30782640 let sent_bytes: usize = first[0].iter().map(|e| e.encrypted_blob.len()).sum();
30792641
30802642 // ...and the relay says that was too big.
3081− let mut cmd = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(413))), &mut m);
2643+ let mut cmd = app.update(Event::SendResult(Ok(ok_response(413))), &mut m);
30822644
30832645 assert_eq!(
30842646 m.send_batch_budget,
@@ -3099,7 +2661,7 @@ mod tests {
30992661 assert!(shrunk > 0, "but not an empty one");
31002662
31012663 // The relay takes the smaller batch, and exactly it is popped.
3102− let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
2664+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
31032665 assert_eq!(m.outbox.len(), 5 - shrunk, "exactly the batch that landed is gone");
31042666 assert_eq!(
31052667 m.send_batch_budget,
@@ -3204,7 +2766,7 @@ mod tests {
32042766 let parsed = protocol::parse_payload(item.envelope_json.as_bytes()).unwrap();
32052767 match parsed.payload {
32062768 Payload::Profile { display_name, .. } => {
3207− assert_eq!(display_name, Some("Three".to_string()), "and it is the latest one");
2769+ assert_eq!(display_name, "Three", "and it is the latest one");
32082770 }
32092771 _ => panic!("expected a profile"),
32102772 }
@@ -3247,12 +2809,12 @@ mod tests {
32472809 );
32482810 assert_eq!(m.outbox.len(), 2, "the in-flight item is left alone");
32492811
3250− let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
2812+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
32512813 assert_eq!(m.outbox.len(), 1, "the sent one is popped");
32522814 let parsed = protocol::parse_payload(m.outbox[0].envelope_json.as_bytes()).unwrap();
32532815 match parsed.payload {
32542816 Payload::Profile { display_name, .. } => assert_eq!(
3255− display_name, Some("New".to_string()),
2817+ display_name, "New",
32562818 "and the newer profile survives to be sent"
32572819 ),
32582820 _ => panic!("expected a profile"),
@@ -3293,7 +2855,7 @@ mod tests {
32932855 assert_eq!(m.outbox.len(), 1);
32942856
32952857 // The send lands. It must pop nothing.
3296− let _ = app.update(Event::SendResult(m.flush_gen, Ok(ok_response(200))), &mut m);
2858+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
32972859 assert_eq!(m.outbox.len(), 1, "the friend's message is untouched");
32982860 assert_eq!(m.outbox[0].recipient, friend);
32992861 }
@@ -3434,8 +2996,8 @@ mod tests {
34342996 .insert(peer.clone(), Contact::new(peer.clone(), String::new(), 0));
34352997
34362998 let profile = |name: &str| Payload::Profile {
3437− display_name: Some(name.to_string()),
3438− bio: Some(String::new()),
2999+ display_name: name.into(),
3000+ bio: String::new(),
34393001 photo: None,
34403002 };
34413003 let page = page_from(9, &m, &profile("Far"), i64::MAX);
@@ -3532,11 +3094,8 @@ mod tests {
35323094 let (m, peer) = ingest_one_text_from_an_unblocked_peer();
35333095 assert_eq!(m.messages[&peer].len(), 1);
35343096 assert_eq!(m.messages[&peer][0].body, "hi");
3535− // D6-app: A profile broadcast is queued alongside the delivery.ack
3536− // so the new contact sees our display name and avatar.
3537− assert_eq!(m.outbox.len(), 2, "one delivery.ack + one profile queued");
3097+ assert_eq!(m.outbox.len(), 1, "one delivery.ack is queued");
35383098 assert_eq!(m.outbox[0].recipient, peer);
3539− assert_eq!(m.outbox[1].recipient, peer);
35403099 }
35413100
35423101 /// A peer sending `ts = i64::MAX` used to pin itself to the top of the
@@ -3717,28 +3276,6 @@ mod tests {
37173276 assert!(m.error.is_some());
37183277 }
37193278
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−
37423279 #[test]
37433280 fn send_text_enqueues_message_profile_and_outbox() {
37443281 let app = Skrepka;
@@ -3830,30 +3367,6 @@ mod tests {
38303367 assert!(vm.contacts.iter().any(|c| c.pubkey == peer && c.blocked));
38313368 }
38323369
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−
38573370 /// Gating ingest stops the acks we have not queued yet — but the ack for the
38583371 /// message that made the user hit "block" is already in the outbox, and an ack
38593372 /// is exactly the "we are online and reading you" signal the block exists to
@@ -3973,7 +3486,7 @@ mod tests {
39733486 assert!(m.flushing);
39743487
39753488 let resp = crux_http::testing::ResponseBuilder::ok().body(Vec::new()).build();
3976− let _ = app.update(Event::SendResult(m.flush_gen, Ok(resp)), &mut m);
3489+ let _ = app.update(Event::SendResult(Ok(resp)), &mut m);
39773490
39783491 assert!(!m.flushing, "flushing must be cleared on the success path");
39793492 assert!(m.outbox.is_empty(), "the sent item is dropped from the outbox");
@@ -3995,7 +3508,7 @@ mod tests {
39953508
39963509 // Its success must both drop the item and let the *next* one go out.
39973510 let resp = crux_http::testing::ResponseBuilder::ok().body(Vec::new()).build();
3998− let _ = app.update(Event::SendResult(m.flush_gen, Ok(resp)), &mut m);
3511+ let _ = app.update(Event::SendResult(Ok(resp)), &mut m);
39993512 assert_eq!(m.outbox.len(), 1);
40003513 assert!(m.flushing, "the second item is now in flight");
40013514 }
@@ -4138,7 +3651,6 @@ mod tests {
41383651 Event::LoadedProfile(Ok(None)),
41393652 Event::LoadedContacts(Ok(None)),
41403653 Event::LoadedCursor(Ok(None)),
4141− Event::LoadedSeenIds(Ok(None)),
41423654 ]
41433655 .into_iter()
41443656 .enumerate()
@@ -4152,7 +3664,7 @@ mod tests {
41523664 assert_eq!(connects(&mut cmd), 0, "and none of them connects");
41533665 }
41543666
4155− // The sixth — and only the sixth — emits Connect.
3667+ // The fifth — and only the fifth — emits Connect.
41563668 let mut cmd = app.update(Event::LoadedOutbox(Ok(None)), &mut m);
41573669 assert_eq!(m.loads_pending, 0);
41583670 assert_eq!(connects(&mut cmd), 1, "the last load connects, exactly once");
@@ -4242,7 +3754,7 @@ mod tests {
42423754 );
42433755
42443756 // 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();
3757+ let sk = Identity::from_seed(&[3u8; 32]).secret_key.to_vec();
42463758 let _ = app.update(Event::IdentityLoaded(sk), &mut m);
42473759 assert!(
42483760 !m.kv_load_failed,
@@ -4349,8 +3861,8 @@ mod tests {
43493861 );
43503862
43513863 let profile = Payload::Profile {
4352− display_name: Some("Mallory".into()),
4353− bio: Some("spam".into()),
3864+ display_name: "Mallory".into(),
3865+ bio: "spam".into(),
43543866 photo: Some("aGk=".into()),
43553867 };
43563868 let page = page_from(9, &m, &profile, now_ms());
@@ -4431,8 +3943,8 @@ mod tests {
44313943 let peer = peer_hex(9);
44323944
44333945 let profile = Payload::Profile {
4434− display_name: Some("Mallory".into()),
4435− bio: Some(String::new()),
3946+ display_name: "Mallory".into(),
3947+ bio: String::new(),
44363948 photo: Some("aGk=".into()),
44373949 };
44383950 let page = page_from(9, &m, &profile, now_ms());
@@ -4644,7 +4156,7 @@ mod tests {
46444156
46454157 // One page holding more messages from one sender than fits in a single ack.
46464158 let count = protocol::MAX_ACK_IDS + 5;
4647− let sender = Identity::from_seed(&[9u8; 32]).unwrap();
4159+ let sender = Identity::from_seed(&[9u8; 32]);
46484160 let me = Identity::from_secret_bytes(m.secret_key.as_ref().unwrap()).unwrap();
46494161 let events = (0..count)
46504162 .map(|i| {
@@ -4801,10 +4313,6 @@ mod tests {
48014313 // send_text block check
48024314 // -----------------------------------------------------------------------
48034315
4804− // -----------------------------------------------------------------------
4805− // send_text body length check
4806− // -----------------------------------------------------------------------
4807−
48084316 /// Blocking cuts a peer off in both directions. `send_text` must refuse to
48094317 /// enqueue a message to a blocked contact — otherwise the block only stops
48104318 /// their incoming mail while we keep sending them ours.
@@ -4833,195 +4341,4 @@ mod tests {
48334341 // and resend, and losing their draft would be surprising.
48344342 assert_eq!(m.compose, "hello?");
48354343 }
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− }
50274344}