@@ -18,11 +18,13 @@ use crux_kv::KeyValueOperation;
1818use crux_time::TimeRequest;
1919use facet::Facet;
2020use serde::{Deserialize, Serialize};
21+use url::Url;
22+use zeroize::Zeroizing;
2123
22−use crate::crypto::Identity;
24+use crate::crypto::{Identity, MAX_BLOB_LEN};
2325use crate::model::{
2426 hex_to_ob, trunc_ob, Contact, ContactVM, MessageVM, OutboxItem, OwnProfile, ProfileVM,
25− Settings, StoredMessage, ViewModel, MAX_MESSAGES_PER_PEER,
27+ Settings, StoredMessage, ViewModel, DEFAULT_SERVER_URL, MAX_MESSAGES_PER_PEER,
2628};
2729use crate::phonemic;
2830use crate::protocol::{self, Envelope, Payload};
@@ -38,6 +40,38 @@ type KvData = Result<Option<Vec<u8>>, crux_kv::error::KeyValueError>;
3840/// peer park itself at the top of the conversation list.
3941const MAX_FUTURE_SKEW_MS: i64 = 60_000;
4042
43+/// Cap on stored contacts.
44+///
45+/// Any stranger can create one just by sending us a message, so without a cap a
46+/// spammer holding a list of pubkeys can grow `contacts` (and the `contacts` kv
47+/// blob, rewritten in full on every change) without bound. Past the cap we drop
48+/// mail from unknown keys; contacts the user added by hand are never at risk,
49+/// because `AddContact` doesn't go through this path.
50+const MAX_CONTACTS: usize = 500;
51+
52+/// Cap on the total blob bytes we will decrypt out of a single poll page.
53+///
54+/// A hostile relay can answer `/poll` with as many events as it likes. Each one
55+/// costs a hex decode, a scalar mult and an AEAD pass, all before we can tell it
56+/// is junk. `MAX_BLOB_LEN` bounds one blob; this bounds the page. Anything past
57+/// the budget is dropped exactly like an undecryptable blob would be.
58+const MAX_POLL_TOTAL_BYTES: usize = 64 * 1024 * 1024;
59+
60+/// How long a poll may stay in flight before we assume the shell lost it.
61+///
62+/// `polling` gates the whole long-poll loop, and it is only cleared by a
63+/// `PollResult`. If the shell ever drops an HTTP effect (a cancelled task, a
64+/// backgrounded app, a bug), `polling` stays `true` and the client goes quiet
65+/// forever with no visible error. The watchdog re-arms the loop instead.
66+///
67+/// This **must** stay above the shell's own HTTP timeout (70 s, see
68+/// `Effects.swift`), or the watchdog stops being a watchdog: it would fire on
69+/// every poll that is merely slow rather than lost, abandoning a request the
70+/// shell is still going to resolve. The generation check in `PollResult` makes
71+/// that survivable, but each misfire still costs a redundant request to the
72+/// relay, so leave the margin.
73+const POLL_WATCHDOG_MS: u64 = 90_000;
74+
4175// kv keys
4276const K_SETTINGS: &str = "settings";
4377const K_PROFILE: &str = "profile";
@@ -115,20 +149,44 @@ pub enum Event {
115149 #[serde(skip)]
116150 #[facet(skip)]
117151 Saved(#[facet(opaque)] KvData),
152+ /// The cursor write specifically: the next poll is chained off it, so that a
153+ /// message is never acked to the server before it is durable locally.
154+ #[serde(skip)]
155+ #[facet(skip)]
156+ SavedCursor(#[facet(opaque)] KvData),
118157
119158 // ---- internal: auth / poll / send ----
159+ // These drive the core's own loops and are never sent by the shell, so they
160+ // are kept off the FFI surface entirely.
161+ //
162+ // The auth and poll results carry the *generation* of the round-trip that
163+ // produced them. An in-flight effect is never cancelled — it can only be
164+ // superseded — so without a generation a result belonging to an attempt we
165+ // already abandoned lands in the handler for the attempt that replaced it.
166+ // See `Model::poll_gen`.
167+ #[serde(skip)]
168+ #[facet(skip)]
120169 Authenticate,
170+ #[serde(skip)]
171+ #[facet(skip)]
121172 Poll,
173+ /// Fires `POLL_WATCHDOG_MS` after generation `n`'s poll is issued; a no-op
174+ /// unless that exact poll is somehow still in flight.
175+ #[serde(skip)]
176+ #[facet(skip)]
177+ PollWatchdog(u64),
178+ #[serde(skip)]
179+ #[facet(skip)]
122180 StartFlush,
123181 #[serde(skip)]
124182 #[facet(skip)]
125− ChallengeResult(#[facet(opaque)] HttpResult),
183+ ChallengeResult(u64, #[facet(opaque)] HttpResult),
126184 #[serde(skip)]
127185 #[facet(skip)]
128− VerifyResult(#[facet(opaque)] HttpResult),
186+ VerifyResult(u64, #[facet(opaque)] HttpResult),
129187 #[serde(skip)]
130188 #[facet(skip)]
131− PollResult(#[facet(opaque)] HttpResult),
189+ PollResult(u64, #[facet(opaque)] HttpResult),
132190 #[serde(skip)]
133191 #[facet(skip)]
134192 SendResult(#[facet(opaque)] HttpResult),
@@ -157,8 +215,14 @@ pub enum Page {
157215 EditProfile,
158216}
159217
218+/// The five kv keys loaded at startup (`settings`, `profile`, `contacts`,
219+/// `cursor`, `outbox`). `Connect` waits for all of them.
220+const STARTUP_LOADS: u8 = 5;
221+
160222pub struct Model {
161− secret_key: Option<Vec<u8>>,
223+ /// Zeroized on drop: this is the Ed25519 seed, the one secret whose leak
224+ /// costs the user their identity.
225+ secret_key: Option<Zeroizing<Vec<u8>>>,
162226 my_pubkey: String,
163227 settings: Settings,
164228 profile: OwnProfile,
@@ -172,7 +236,35 @@ pub struct Model {
172236 /// A poll request is in flight. Guards against stacking concurrent long-poll
173237 /// loops (every extra loop re-polls itself forever and never terminates).
174238 polling: bool,
239+ /// Which poll the in-flight `PollResult` is allowed to belong to.
240+ ///
241+ /// Bumped every time a poll is issued, and every time one is *abandoned* (the
242+ /// watchdog gives up on it; a relay switch invalidates it). A `PollResult`
243+ /// whose generation no longer matches is dropped on the floor.
244+ ///
245+ /// A wall-clock "has it been long enough?" test cannot do this job. An HTTP
246+ /// effect stays pending across an iOS suspension while `now_ms()` runs on, so
247+ /// on resume *every* in-flight poll looks stuck — the watchdog abandons a
248+ /// request the shell is still about to resolve, and that resolution then
249+ /// clears the guard belonging to its replacement. Two poll loops, each
250+ /// re-polling itself forever. The generation is what makes abandoning safe.
251+ poll_gen: u64,
252+ /// An auth round-trip is in flight. `Connect` is fired from several places
253+ /// (startup, a 401, a reconnect timer, a server change); without this guard
254+ /// they stack, and each one that lands installs a token and starts its own
255+ /// poll loop.
256+ authenticating: bool,
257+ /// The same idea for the auth round-trip, which spans two legs (challenge →
258+ /// verify) and is restarted by `Connect` — which the shell fires on *every*
259+ /// return to the foreground (`SkrepkaApp.swift`). The challenge still in
260+ /// flight from before the app was backgrounded must not install a token or
261+ /// overwrite `conn` on top of the attempt that replaced it.
262+ auth_gen: u64,
175263 flushing: bool,
264+ /// Startup kv loads still outstanding. `Connect` only fires at 0, so a poll
265+ /// can't land and be persisted against a half-loaded model — which would
266+ /// then be clobbered by the loads still in flight.
267+ loads_pending: u8,
176268 page: Page,
177269 active_peer: Option<String>,
178270 compose: String,
@@ -194,7 +286,11 @@ impl Default for Model {
194286 conn: ConnStatus::Offline,
195287 poll_retries: 0,
196288 polling: false,
289+ poll_gen: 0,
290+ authenticating: false,
291+ auth_gen: 0,
197292 flushing: false,
293+ loads_pending: 0,
198294 page: Page::default(),
199295 active_peer: None,
200296 compose: String::new(),
@@ -209,12 +305,59 @@ impl Model {
209305 .as_ref()
210306 .and_then(|sk| Identity::from_secret_bytes(sk).ok())
211307 }
308+
309+ /// Tick off one startup load. Returns `true` exactly once — when the last of
310+ /// the five lands — so `Connect` is emitted a single time.
311+ fn startup_load_done(&mut self) -> bool {
312+ if self.loads_pending == 0 {
313+ return false;
314+ }
315+ self.loads_pending -= 1;
316+ self.loads_pending == 0
317+ }
318+
319+ /// Open a new poll generation and return it. Any `PollResult` still in flight
320+ /// under the old generation is now dead to us.
321+ fn next_poll_gen(&mut self) -> u64 {
322+ self.poll_gen = self.poll_gen.wrapping_add(1);
323+ self.poll_gen
324+ }
325+
326+ /// Abandon the in-flight poll, if any: release the loop guard and retire the
327+ /// generation so a late `PollResult` cannot clear the *replacement* poll's
328+ /// guard or fold a stale page (and a stale cursor) into the model.
329+ fn abandon_poll(&mut self) {
330+ self.polling = false;
331+ self.poll_gen = self.poll_gen.wrapping_add(1);
332+ }
333+
334+ /// Open a new auth generation and return it.
335+ fn next_auth_gen(&mut self) -> u64 {
336+ self.auth_gen = self.auth_gen.wrapping_add(1);
337+ self.auth_gen
338+ }
339+
340+ /// Abandon the in-flight auth round-trip (either leg).
341+ fn abandon_auth(&mut self) {
342+ self.authenticating = false;
343+ self.auth_gen = self.auth_gen.wrapping_add(1);
344+ }
212345}
213346
214347// ---------------------------------------------------------------------------
215348// Helpers (pure)
216349// ---------------------------------------------------------------------------
217350
351+/// Wall-clock milliseconds.
352+///
353+/// This reads the system clock from inside `update()`, which is supposed to be a
354+/// pure function of `(Event, Model)`. It isn't, and that has real costs: the
355+/// state machine can't be replayed deterministically, and tests can't pin "now".
356+/// The clean fix is to take the time as a `crux_time` effect and feed it back in
357+/// as an event — but every timestamped path (`SendText`, `SaveProfile`,
358+/// `ingest_poll`) would have to become a two-step round-trip through the shell.
359+/// That refactor is worth doing; it is not worth folding into a bug-fix pass, so
360+/// the impurity is documented rather than hidden.
218361fn now_ms() -> i64 {
219362 SystemTime::now()
220363 .duration_since(UNIX_EPOCH)
@@ -223,36 +366,44 @@ fn now_ms() -> i64 {
223366}
224367
225368/// Bare lowercased hostname of a server URL (no scheme/port/trailing dot).
369+///
370+/// This feeds the auth signature, which is bound to the host (PROTOCOL.md §6),
371+/// so it must agree with what the relay itself computes. Splitting on `:` by hand
372+/// mangles an IPv6 literal (`http://[::1]:8080` → `[`), producing a signature the
373+/// server rejects; `Url` gets the bracket rules right.
226374fn server_host(url: &str) -> String {
227− let after_scheme = url.split("://").nth(1).unwrap_or(url);
228− let host_port = after_scheme.split('/').next().unwrap_or("");
229− let host = host_port.split('@').next_back().unwrap_or(host_port);
230− let host = host.split(':').next().unwrap_or(host);
231− host.trim_end_matches('.').to_lowercase()
375+ Url::parse(url)
376+ .ok()
377+ .and_then(|u| u.host_str().map(str::to_string))
378+ .map(|h| h.trim_end_matches('.').to_lowercase())
379+ .unwrap_or_default()
232380}
233381
234382/// Validate and canonicalize a user-entered relay URL.
235383///
236−/// The shell hands us raw text-field contents, and `Http::post` panics on a URL
237−/// with no scheme — so a schemeless entry must be rejected here, in the core,
238−/// before it can ever reach an effect. Returns the normalized URL (lowercased
239−/// scheme, no trailing slash) or `None` if it isn't an absolute http(s) URL.
384+/// The shell hands us raw text-field contents and `crux_http` unwraps the parse,
385+/// so anything the real parser rejects must be rejected *here* or it panics the
386+/// core. That means validating with the same parser rather than a lookalike: the
387+/// hand-rolled check this replaces accepted `http://[bad` and `http://ho^st`,
388+/// both of which `Url::parse` refuses.
389+///
390+/// A query or fragment is rejected too. Every request appends a path
391+/// (`{url}/poll`), so `http://x?y` would yield `http://x?y/poll` — the path
392+/// swallowed into the query string, pointing at nothing.
393+///
394+/// Returns the normalized URL (no trailing slash) or `None`.
240395fn normalize_server_url(input: &str) -> Option<String> {
241− let (scheme, rest) = input.trim().split_once("://")?;
242− if !scheme.eq_ignore_ascii_case("http") && !scheme.eq_ignore_ascii_case("https") {
396+ let url = Url::parse(input.trim()).ok()?;
397+ if !matches!(url.scheme(), "http" | "https") {
243398 return None;
244399 }
245− let rest = rest.trim_end_matches('/');
246− if rest.chars().any(char::is_whitespace) {
400+ if url.host_str().is_none_or(str::is_empty) {
247401 return None;
248402 }
249− // Require a non-empty authority: reject "http://", "http:///x", "http://:80".
250− let authority = rest.split('/').next().unwrap_or("");
251− let host = authority.split('@').next_back().unwrap_or(authority);
252− if host.split(':').next().unwrap_or("").is_empty() {
403+ if url.query().is_some() || url.fragment().is_some() {
253404 return None;
254405 }
255− Some(format!("{}://{}", scheme.to_lowercase(), rest))
406+ Some(url.as_str().trim_end_matches('/').to_string())
256407}
257408
258409fn json_bytes<T: Serialize>(value: &T) -> Vec<u8> {
@@ -326,10 +477,12 @@ impl App for Skrepka {
326477 model.error = Some("invalid identity key".into());
327478 return render();
328479 };
329− model.secret_key = Some(sk);
480+ model.secret_key = Some(Zeroizing::new(sk));
330481 model.my_pubkey = id.public_key_hex();
331482 model.page = Page::Conversations;
332− // Fan out the startup loads.
483+ // Fan out the startup loads. `Connect` waits until the last of
484+ // them lands (see `startup_load_done`).
485+ model.loads_pending = STARTUP_LOADS;
333486 Command::all([
334487 KeyValue::get(K_SETTINGS).then_send(Event::LoadedSettings),
335488 KeyValue::get(K_PROFILE).then_send(Event::LoadedProfile),
@@ -342,45 +495,47 @@ impl App for Skrepka {
342495
343496 Event::LoadedSettings(res) => {
344497 if let Some(s) = parse_kv::<Settings>(res) {
345− model.settings = s;
498+ // Re-validate on the way in. The URL was normalized when the
499+ // user typed it, but the blob on disk is not trustworthy: an
500+ // older build wrote URLs this parser rejects, and a corrupted
501+ // or hand-edited file would reach `Http::post`, which unwraps
502+ // the parse. That is a panic on every launch — a boot loop
503+ // with no way out from inside the app.
504+ model.settings.server_url = normalize_server_url(&s.server_url)
505+ .unwrap_or_else(|| DEFAULT_SERVER_URL.to_string());
346506 }
347− // Server URL known — connect and resume any queued sends.
348− Command::event(Event::Connect)
507+ self.startup_done(model).and(render())
349508 }
350509 Event::LoadedProfile(res) => {
351510 if let Some(p) = parse_kv::<OwnProfile>(res) {
352511 model.profile = p;
353512 }
354− render()
513+ self.startup_done(model).and(render())
355514 }
356515 Event::LoadedContacts(res) => {
516+ let mut cmds = vec![self.startup_done(model)];
357517 if let Some(list) = parse_kv::<Vec<Contact>>(res) {
358518 model.contacts = list.into_iter().map(|c| (c.pubkey.clone(), c)).collect();
359519 // Lazily load each conversation's messages.
360− let loads: Vec<_> = model
361− .contacts
362− .keys()
363− .map(|peer| {
364− let peer = peer.clone();
365− KeyValue::get(k_messages(&peer))
366− .then_send(move |r| Event::LoadedMessages(peer.clone(), r))
367− })
368− .collect();
369− return Command::all(loads).and(render());
520+ cmds.extend(model.contacts.keys().map(|peer| {
521+ let peer = peer.clone();
522+ KeyValue::get(k_messages(&peer))
523+ .then_send(move |r| Event::LoadedMessages(peer.clone(), r))
524+ }));
370525 }
371− render()
526+ Command::all(cmds).and(render())
372527 }
373528 Event::LoadedCursor(res) => {
374529 if let Some(c) = parse_kv::<i64>(res) {
375530 model.cursor = c;
376531 }
377− render()
532+ self.startup_done(model).and(render())
378533 }
379534 Event::LoadedOutbox(res) => {
380535 if let Some(list) = parse_kv::<Vec<OutboxItem>>(res) {
381536 model.outbox = list.into();
382537 }
383− render()
538+ self.startup_done(model).and(render())
384539 }
385540 Event::LoadedMessages(peer, res) => {
386541 if let Some(list) = parse_kv::<Vec<StoredMessage>>(res) {
@@ -397,15 +552,32 @@ impl App for Skrepka {
397552 }
398553 Command::done()
399554 }
555+ Event::SavedCursor(res) => {
556+ // The cursor write is what tells the relay it may delete the mail
557+ // we just took (it acks up to `cursor` on the next poll). Only
558+ // re-poll once it has actually landed: if we polled in parallel
559+ // and the write failed, the messages would be gone from the
560+ // server and absent from disk.
561+ if let Err(e) = res {
562+ model.error = Some(format!("could not save data: {e}"));
563+ return self.backoff_poll(model);
564+ }
565+ Command::event(Event::Poll)
566+ }
400567
401568 // ---------------- navigation ----------------
569+ // `error` is about the thing the user was just doing. Leaving it set
570+ // across a page change strands a stale message ("invalid public key")
571+ // on an unrelated screen, so every navigation clears it.
402572 Event::ShowConversations => {
403573 model.page = Page::Conversations;
404574 model.active_peer = None;
575+ model.error = None;
405576 render()
406577 }
407578 Event::ShowSettings => {
408579 model.page = Page::Settings;
580+ model.error = None;
409581 render()
410582 }
411583 Event::ShowAddContact => {
@@ -415,16 +587,19 @@ impl App for Skrepka {
415587 }
416588 Event::ShowEditProfile => {
417589 model.page = Page::EditProfile;
590+ model.error = None;
418591 render()
419592 }
420593 Event::OpenChat(peer) => {
421594 model.active_peer = Some(peer);
422595 model.page = Page::Chat;
596+ model.error = None;
423597 render()
424598 }
425599 Event::Back => {
426600 model.page = Page::Conversations;
427601 model.active_peer = None;
602+ model.error = None;
428603 render()
429604 }
430605
@@ -435,19 +610,42 @@ impl App for Skrepka {
435610 }
436611 Event::SetServerUrl(url) => {
437612 let Some(url) = normalize_server_url(&url) else {
438− model.error = Some("server URL must start with http:// or https://".into());
613+ model.error = Some("server URL must be an http(s) address".into());
439614 return render();
440615 };
441616 model.error = None;
442617 model.settings.server_url = url;
443618 model.token = None;
444619 model.conn = ConnStatus::Offline;
445− KeyValue::set(K_SETTINGS, json_bytes(&model.settings))
446− .then_send(Event::Saved)
447− .and(Command::event(Event::Connect))
448− .and(render())
620+ // The cursor is *per relay*: it is that server's own sequence
621+ // number, and the next poll acks everything up to it. Carrying
622+ // relay A's cursor (roughly `now_ms()`) over to relay B tells B
623+ // to delete every message queued for us before it hands any of
624+ // them over. Start B from the beginning.
625+ model.cursor = 0;
626+ // ...and retire the poll that is still in flight *to relay A*, or
627+ // resetting the cursor achieves nothing: A's answer is already on
628+ // its way, it carries A's cursor, and `ingest_poll` would write it
629+ // straight back over the zero we just set. The next poll — to B —
630+ // would then ack a sequence number B never issued, and B would
631+ // drop every message waiting for us.
632+ model.abandon_poll();
633+ Command::all([
634+ KeyValue::set(K_SETTINGS, json_bytes(&model.settings)).then_send(Event::Saved),
635+ KeyValue::set(K_CURSOR, json_bytes(&0i64)).then_send(Event::Saved),
636+ ])
637+ .and(Command::event(Event::Connect))
638+ .and(render())
449639 }
450640 Event::Connect => {
641+ // Connect restarts the auth flow, so whatever challenge/verify was
642+ // in flight is now stale and its guard must come off. Retiring the
643+ // generation as we do it is what makes that safe: the shell fires
644+ // Connect on every return to the foreground, so the round-trip we
645+ // are dropping here is usually one that is still very much alive
646+ // and about to resolve. Clearing the guard alone would let it land
647+ // and install its token on top of the attempt replacing it.
648+ model.abandon_auth();
451649 if model.secret_key.is_some() && model.conn != ConnStatus::Online {
452650 Command::event(Event::Authenticate)
453651 } else {
@@ -484,6 +682,18 @@ impl App for Skrepka {
484682 if let Some(c) = model.contacts.get_mut(&peer) {
485683 c.blocked = blocked;
486684 }
685+ if blocked {
686+ // Gating *ingest* on the block only stops the acks we have not
687+ // queued yet. Anything already sitting in the outbox for this
688+ // peer — an ack for the very message that made the user block
689+ // them, a profile from the last broadcast — still flushes, and
690+ // an ack is precisely the "we are online and reading you"
691+ // signal the block is supposed to cut off. Drop it all: once
692+ // blocked, nothing more goes to this peer.
693+ model.outbox.retain(|item| item.recipient != peer);
694+ return Command::all([self.persist_contacts(model), self.persist_outbox(model)])
695+ .and(render());
696+ }
487697 self.persist_contacts(model).and(render())
488698 }
489699 Event::SaveProfile {
@@ -503,14 +713,23 @@ impl App for Skrepka {
503713 bio: model.profile.bio.clone(),
504714 photo: model.profile.photo.clone(),
505715 };
506− // Broadcast the new profile to every contact.
716+ // Broadcast the new profile to every contact we haven't blocked.
717+ // Blocking is meant to cut the peer off in both directions; a
718+ // profile broadcast would keep feeding them our display name,
719+ // avatar, and a liveness signal.
507720 //
508721 // Each item carries its own copy of the serialized payload, so a
509722 // profile with a photo is duplicated (base64 and all) once per
510723 // contact, in memory and in the `outbox` kv blob. Acceptable at
511724 // today's contact counts; the fix is to intern the payload and
512725 // have `OutboxItem` reference it by id.
513− for peer in model.contacts.keys().cloned().collect::<Vec<_>>() {
726+ let recipients: Vec<String> = model
727+ .contacts
728+ .values()
729+ .filter(|c| !c.blocked)
730+ .map(|c| c.pubkey.clone())
731+ .collect();
732+ for peer in recipients {
514733 model.outbox.push_back(OutboxItem {
515734 recipient: peer,
516735 envelope_json: protocol::serialize_payload(&payload, ts),
@@ -527,6 +746,15 @@ impl App for Skrepka {
527746
528747 // ---------------- auth ----------------
529748 Event::Authenticate => {
749+ // Exactly one auth round-trip at a time. `Authenticate` is reached
750+ // from startup, a 401 on poll, a 401 on send, a reconnect timer
751+ // and a server change — all of which can fire at once after a
752+ // network blip. Each one that got through would hit
753+ // `/auth/challenge`, and each token that came back would install
754+ // itself and kick off another poll loop.
755+ if model.authenticating {
756+ return Command::done();
757+ }
530758 let Some(id) = model.identity() else {
531759 return render();
532760 };
@@ -537,7 +765,13 @@ impl App for Skrepka {
537765 pubkey: &id.public_key_hex(),
538766 };
539767 match Http::post(url).body_json(&body) {
540− Ok(req) => req.build().then_send(Event::ChallengeResult).and(render()),
768+ Ok(req) => {
769+ model.authenticating = true;
770+ let gen = model.next_auth_gen();
771+ req.build()
772+ .then_send(move |r| Event::ChallengeResult(gen, r))
773+ .and(render())
774+ }
541775 Err(_) => {
542776 model.conn = ConnStatus::Offline;
543777 model.error = Some("could not build auth request".into());
@@ -545,42 +779,67 @@ impl App for Skrepka {
545779 }
546780 }
547781 }
548− Event::ChallengeResult(Ok(mut resp)) => {
782+ Event::ChallengeResult(gen, res) => {
783+ // A challenge from an attempt we already gave up on (a relay
784+ // switch, a foreground `Connect`). Its `authenticating` is not
785+ // ours to clear and its challenge is not ours to sign.
786+ if gen != model.auth_gen {
787+ return Command::done();
788+ }
789+ // Cleared here and re-set only if the verify leg actually goes
790+ // out — every exit from this arm must leave the guard consistent.
791+ model.authenticating = false;
792+ let Ok(mut resp) = res else {
793+ model.conn = ConnStatus::Offline;
794+ return self.schedule_reconnect(model);
795+ };
549796 let bytes = resp.take_body().unwrap_or_default();
550797 let Some(id) = model.identity() else {
551798 return render();
552799 };
553− match serde_json::from_slice::<ChallengeResp>(&bytes) {
554− Ok(c) => {
555− let host = server_host(&model.settings.server_url);
556− let signature = id.sign_challenge(&host, &c.challenge);
557− let url = format!("{}/auth/verify", model.settings.server_url);
558− let body = VerifyReq {
559− pubkey: &id.public_key_hex(),
560− challenge: &c.challenge,
561− signature: &signature,
562− revoke_others: false,
563− };
564− match Http::post(url).body_json(&body) {
565− Ok(req) => req.build().then_send(Event::VerifyResult),
566− Err(_) => {
567− model.conn = ConnStatus::Offline;
568− model.error = Some("could not build auth request".into());
569− self.schedule_reconnect(model)
570− }
571− }
800+ let Ok(c) = serde_json::from_slice::<ChallengeResp>(&bytes) else {
801+ model.conn = ConnStatus::Offline;
802+ return self.schedule_reconnect(model);
803+ };
804+ let host = server_host(&model.settings.server_url);
805+ let Ok(signature) = id.sign_challenge(&host, &c.challenge) else {
806+ // An over-long challenge: a relay trying to get us to sign
807+ // something of its choosing. Back off rather than comply.
808+ model.conn = ConnStatus::Offline;
809+ model.error = Some("relay sent a malformed auth challenge".into());
810+ return self.schedule_reconnect(model);
811+ };
812+ let url = format!("{}/auth/verify", model.settings.server_url);
813+ let body = VerifyReq {
814+ pubkey: &id.public_key_hex(),
815+ challenge: &c.challenge,
816+ signature: &signature,
817+ revoke_others: false,
818+ };
819+ match Http::post(url).body_json(&body) {
820+ Ok(req) => {
821+ model.authenticating = true;
822+ // The verify leg belongs to the same attempt as the
823+ // challenge that produced it, so it keeps the generation
824+ // rather than opening a new one.
825+ req.build().then_send(move |r| Event::VerifyResult(gen, r))
572826 }
573827 Err(_) => {
574828 model.conn = ConnStatus::Offline;
829+ model.error = Some("could not build auth request".into());
575830 self.schedule_reconnect(model)
576831 }
577832 }
578833 }
579− Event::ChallengeResult(Err(_)) => {
580− model.conn = ConnStatus::Offline;
581− self.schedule_reconnect(model)
582− }
583− Event::VerifyResult(Ok(mut resp)) => {
834+ Event::VerifyResult(gen, res) => {
835+ if gen != model.auth_gen {
836+ return Command::done();
837+ }
838+ model.authenticating = false;
839+ let Ok(mut resp) = res else {
840+ model.conn = ConnStatus::Offline;
841+ return self.schedule_reconnect(model);
842+ };
584843 let bytes = resp.take_body().unwrap_or_default();
585844 match serde_json::from_slice::<VerifyResp>(&bytes) {
586845 Ok(v) if !v.token.is_empty() => {
@@ -598,10 +857,6 @@ impl App for Skrepka {
598857 }
599858 }
600859 }
601− Event::VerifyResult(Err(_)) => {
602− model.conn = ConnStatus::Offline;
603− self.schedule_reconnect(model)
604− }
605860
606861 // ---------------- poll ----------------
607862 Event::Poll => {
@@ -623,7 +878,16 @@ impl App for Skrepka {
623878 {
624879 Ok(req) => {
625880 model.polling = true;
626− req.build().then_send(Event::PollResult)
881+ let gen = model.next_poll_gen();
882+ // Arm the watchdog alongside the request: if this poll
883+ // never resolves, nothing else will ever clear `polling`.
884+ req.build()
885+ .then_send(move |r| Event::PollResult(gen, r))
886+ .and(
887+ Time::notify_after(Duration::from_millis(POLL_WATCHDOG_MS))
888+ .0
889+ .then_send(move |_| Event::PollWatchdog(gen)),
890+ )
627891 }
628892 Err(_) => {
629893 model.error = Some("could not build poll request".into());
@@ -631,8 +895,36 @@ impl App for Skrepka {
631895 }
632896 }
633897 }
634− Event::PollResult(Ok(mut resp)) => {
898+ Event::PollWatchdog(gen) => {
899+ // Only the poll this watchdog was armed for, and only while it is
900+ // still the current one and still in flight. A watchdog whose poll
901+ // already completed (or was superseded) is a stale timer and must
902+ // not touch a healthy successor.
903+ if gen != model.poll_gen || !model.polling {
904+ return Command::done();
905+ }
906+ // Retire it as we give up on it. The request itself is not
907+ // cancelled — nothing can cancel it — so if it *does* eventually
908+ // resolve, the generation check below is what stops it from
909+ // clearing the replacement poll's guard and forking the loop.
910+ model.abandon_poll();
911+ Command::event(Event::Poll)
912+ }
913+ Event::PollResult(gen, res) => {
914+ // A result for a poll we already abandoned. Dropping the page is
915+ // safe — and required. We never wrote a cursor for it, so the
916+ // relay still holds every message in it and the current poll will
917+ // be handed them again; whereas honouring it here would clear a
918+ // guard that now belongs to a *different*, live poll, and the
919+ // re-poll it chains would be a second loop re-polling itself
920+ // forever alongside the first.
921+ if gen != model.poll_gen {
922+ return Command::done();
923+ }
635924 model.polling = false;
925+ let Ok(mut resp) = res else {
926+ return self.backoff_poll(model);
927+ };
636928 let status = u16::from(resp.status());
637929 if status == 401 {
638930 model.token = None;
@@ -645,14 +937,15 @@ impl App for Skrepka {
645937 let bytes = resp.take_body().unwrap_or_default();
646938 let parsed: PollResp = serde_json::from_slice(&bytes).unwrap_or_default();
647939 let cmd = self.ingest_poll(model, parsed);
648− // Persist cursor and immediately re-poll (the server long-polled 25s).
649− cmd.and(KeyValue::set(K_CURSOR, json_bytes(&model.cursor)).then_send(Event::Saved))
650− .and(Command::event(Event::Poll))
651− .and(render())
652− }
653− Event::PollResult(Err(_)) => {
654− model.polling = false;
655− self.backoff_poll(model)
940+ // The re-poll hangs off the *cursor write*, not off this event.
941+ // Polling again is what acks the batch we just took — the server
942+ // deletes everything up to `cursor` — so it must not happen until
943+ // the batch is durable on our side. `SavedCursor` re-polls.
944+ cmd.and(
945+ KeyValue::set(K_CURSOR, json_bytes(&model.cursor))
946+ .then_send(Event::SavedCursor),
947+ )
948+ .and(render())
656949 }
657950
658951 // ---------------- outbox ----------------
@@ -695,17 +988,38 @@ impl App for Skrepka {
695988 ConnStatus::Online => "online",
696989 };
697990
991+ let active_peer = model.active_peer.clone().unwrap_or_default();
992+
993+ // `view()` runs on every render, and `ComposeChanged` renders on every
994+ // keystroke. A contact's `photo` is a base64 avatar — tens of KiB each —
995+ // so cloning all of them per keypress meant megabytes of allocation and
996+ // bincode-encoding to type one message.
997+ //
998+ // Only the chat page has a composer (`ComposeChanged` is sent from
999+ // `ChatView` and nowhere else), so only the chat page is on that hot path
1000+ // — and while it is up, the only avatar on screen is the peer we are
1001+ // talking to. Every *other* page renders from a cold event, and the
1002+ // conversations list draws an avatar per row, so it gets the real photos.
1003+ // Withholding them there would not save a single keystroke's work; it
1004+ // would just replace every contact's picture with their initials.
1005+ let chat_open = model.page == Page::Chat;
1006+
6981007 let mut contacts: Vec<ContactVM> = model
6991008 .contacts
7001009 .values()
7011010 .map(|c| {
7021011 let msgs = model.messages.get(&c.pubkey);
7031012 let last = msgs.and_then(|m| m.last());
1013+ let photo = if chat_open && c.pubkey != active_peer {
1014+ String::new()
1015+ } else {
1016+ c.photo.clone().unwrap_or_default()
1017+ };
7041018 ContactVM {
7051019 pubkey: c.pubkey.clone(),
7061020 name: c.label(),
7071021 ob: hex_to_ob(&c.pubkey),
708− photo: c.photo.clone().unwrap_or_default(),
1022+ photo,
7091023 blocked: c.blocked,
7101024 last_message: last.map(|m| m.body.clone()).unwrap_or_default(),
7111025 last_ts: last.map_or(0, |m| m.ts),
@@ -714,7 +1028,6 @@ impl App for Skrepka {
7141028 .collect();
7151029 contacts.sort_by(|a, b| b.last_ts.cmp(&a.last_ts).then(a.name.cmp(&b.name)));
7161030
717− let active_peer = model.active_peer.clone().unwrap_or_default();
7181031 let active_contact = model.contacts.get(&active_peer);
7191032 let messages: Vec<MessageVM> = model
7201033 .messages
@@ -760,6 +1073,20 @@ impl App for Skrepka {
7601073}
7611074
7621075impl Skrepka {
1076+ /// One of the five startup kv loads landed. Fires `Connect` on the last one.
1077+ ///
1078+ /// Connecting earlier would let a poll land — and be persisted — against a
1079+ /// model that is still missing its contacts, cursor, or outbox. The loads
1080+ /// still in flight would then overwrite what the poll just ingested, and the
1081+ /// stale `cursor` (0) would make us re-fetch mail we had already acked.
1082+ fn startup_done(&self, model: &mut Model) -> Command<Effect, Event> {
1083+ if model.startup_load_done() {
1084+ Command::event(Event::Connect)
1085+ } else {
1086+ Command::done()
1087+ }
1088+ }
1089+
7631090 fn persist_contacts(&self, model: &Model) -> Command<Effect, Event> {
7641091 let list: Vec<Contact> = model.contacts.values().cloned().collect();
7651092 KeyValue::set(K_CONTACTS, json_bytes(&list)).then_send(Event::Saved)
@@ -784,7 +1111,15 @@ impl Skrepka {
7841111 .and(render())
7851112 }
7861113
1114+ /// The poll loop stalled — retry after a backoff, and stop claiming to be online.
1115+ ///
1116+ /// Every path here (a transport error, a 5xx, a failed cursor write) means the loop
1117+ /// is not healthy, so leaving `conn` at `Online` both lies to the status dot and —
1118+ /// worse — makes `Connect` a no-op, because it stands down when it already believes
1119+ /// it is online. The shell fires `Connect` on every foreground precisely to recover
1120+ /// the long-poll that backgrounding killed; without this it recovers nothing.
7871121 fn backoff_poll(&self, model: &mut Model) -> Command<Effect, Event> {
1122+ model.conn = ConnStatus::Offline;
7881123 let delay = backoff_ms(model.poll_retries);
7891124 model.poll_retries = model.poll_retries.saturating_add(1);
7901125 Time::notify_after(Duration::from_millis(delay))
@@ -849,54 +1184,63 @@ impl Skrepka {
8491184 .and(render())
8501185 }
8511186
1187+ /// Encrypt and send the head of the outbox, discarding any unsendable items
1188+ /// ahead of it.
1189+ ///
1190+ /// The discards happen in one pass, followed by a single `persist_outbox`.
1191+ /// The previous shape recursed once per bad item and emitted a full outbox
1192+ /// rewrite each time, so a contact list holding `n` unusable keys cost
1193+ /// O(n) kv writes of an O(n) blob — and `n` stack frames — to reach the first
1194+ /// item that could actually go out.
8521195 fn flush_next(&self, model: &mut Model) -> Command<Effect, Event> {
8531196 if model.flushing {
8541197 return Command::done();
8551198 }
856− let Some(item) = model.outbox.front().cloned() else {
857− return Command::done();
858− };
8591199 let (Some(id), Some(token)) = (model.identity(), model.token.clone()) else {
8601200 return Command::done();
8611201 };
862− // An unusable recipient key can never be encrypted to, so the item has to
863− // be dropped or it blocks the head of the outbox forever — but say so,
864− // rather than letting the message vanish.
865− let drop_item = |model: &mut Model| {
866− model.error = Some(format!(
867− "cannot send to {}: invalid key",
868− trunc_ob(&item.recipient)
869− ));
870− model.outbox.pop_front();
1202+
1203+ let mut dropped = false;
1204+ let sendable = loop {
1205+ let Some(item) = model.outbox.front() else {
1206+ break None;
1207+ };
1208+ match encrypt_for(&id, item) {
1209+ Some(blob) => break Some((item.recipient.clone(), blob)),
1210+ None => {
1211+ // An unusable recipient key can never be encrypted to, so the
1212+ // item has to go or it blocks the head of the outbox forever —
1213+ // but say so, rather than letting the message vanish.
1214+ model.error = Some(format!(
1215+ "cannot send to {}: invalid key",
1216+ trunc_ob(&item.recipient)
1217+ ));
1218+ model.outbox.pop_front();
1219+ dropped = true;
1220+ }
1221+ }
8711222 };
8721223
873− let Ok(recipient) = hex::decode(&item.recipient) else {
874− drop_item(model);
875− return self.persist_outbox(model).and(self.flush_next(model));
1224+ let persist = if dropped {
1225+ self.persist_outbox(model)
1226+ } else {
1227+ Command::done()
8761228 };
877− let mut recip = [0u8; 32];
878− if recipient.len() != 32 {
879− drop_item(model);
880− return self.persist_outbox(model).and(self.flush_next(model));
881− }
882− recip.copy_from_slice(&recipient);
8831229
884− let mut rng = rand_core::OsRng;
885− let blob = match crate::crypto::encrypt(&mut rng, &id, &recip, item.envelope_json.as_bytes())
886− {
887− // Reachable for a 32-byte key that isn't a curve point: `AddContact`
888− // rejects those now, but an older contact list may still hold one.
889− Ok(b) => b,
890− Err(_) => {
891− drop_item(model);
892− return self.persist_outbox(model).and(self.flush_next(model));
893− }
1230+ let Some((recipient, blob)) = sendable else {
1231+ // Nothing left to send. Render only if we dropped something, so the
1232+ // user sees why their message went away.
1233+ return if dropped {
1234+ persist.and(render())
1235+ } else {
1236+ Command::done()
1237+ };
8941238 };
8951239
8961240 model.flushing = true;
8971241 let batch = SendBatch {
8981242 messages: vec![Envelope {
899− to: item.recipient,
1243+ to: recipient,
9001244 encrypted_blob: hex::encode(blob),
9011245 }],
9021246 };
@@ -905,13 +1249,13 @@ impl Skrepka {
9051249 .header("authorization", format!("Bearer {token}"))
9061250 .body_json(&batch)
9071251 {
908− Ok(req) => req.build().then_send(Event::SendResult),
1252+ Ok(req) => persist.and(req.build().then_send(Event::SendResult)),
9091253 Err(_) => {
9101254 // Not a dead loop: the next StartFlush (from a poll page or a
9111255 // fresh send) retries the head of the outbox.
9121256 model.flushing = false;
9131257 model.error = Some("could not build send request".into());
914− render()
1258+ persist.and(render())
9151259 }
9161260 }
9171261 }
@@ -931,7 +1275,24 @@ impl Skrepka {
9311275 // sender hex -> message ids needing acks
9321276 let mut ack_targets: BTreeMap<String, Vec<String>> = BTreeMap::new();
9331277
1278+ let mut budget = MAX_POLL_TOTAL_BYTES;
9341279 for ev in page.events {
1280+ // Reject an oversized blob on its hex length, before it costs us a
1281+ // decode. No conforming sender produces one (the relay caps blobs at
1282+ // the same size), so this can only be junk.
1283+ if ev.encrypted_blob.len() > MAX_BLOB_LEN * 2 {
1284+ continue;
1285+ }
1286+ match budget.checked_sub(ev.encrypted_blob.len() / 2) {
1287+ Some(left) => budget = left,
1288+ None => {
1289+ // A page this large is a misbehaving relay, not real mail.
1290+ // Stop here; the rest of the page is discarded exactly as an
1291+ // undecryptable blob would be.
1292+ model.error = Some("relay sent an oversized poll page".into());
1293+ break;
1294+ }
1295+ }
9351296 let Ok(blob) = hex::decode(&ev.encrypted_blob) else {
9361297 continue;
9371298 };
@@ -942,25 +1303,38 @@ impl Skrepka {
9421303 continue;
9431304 };
9441305 let sender = dec.sender_hex;
1306+
1307+ // Blocking cuts the peer off entirely (PROTOCOL.md §4) — not just
1308+ // their texts. A blocked peer must not be able to rewrite their
1309+ // display name and avatar in our contact list, mark our messages
1310+ // delivered, or (via an ack we send back) learn that we are online
1311+ // and reading. So the gate sits above the payload match, not inside
1312+ // the `Text` arm.
1313+ if model.contacts.get(&sender).is_some_and(|c| c.blocked) {
1314+ continue;
1315+ }
1316+
9451317 // `ts` is chosen by the sender and drives conversation ordering,
9461318 // in-chat ordering, and the displayed time. Bound it to the near
9471319 // future so a peer cannot pin itself to the top of the list forever
9481320 // with `ts = i64::MAX`.
9491321 let ts = parsed.ts.min(now.saturating_add(MAX_FUTURE_SKEW_MS));
950− let blocked = model.contacts.get(&sender).is_some_and(|c| c.blocked);
1322+ let known = model.contacts.contains_key(&sender);
9511323 match parsed.payload {
9521324 Payload::Text { id: msg_id, body } => {
953− // A blocked peer's messages are not displayed (PROTOCOL.md
954− // §4), so don't store them — and don't ack them either: an
955− // ack would tell them we are online and reading.
956− if blocked {
957− continue;
958− }
959− // auto-create contact
960− if !model.contacts.contains_key(&sender) {
961− model
962− .contacts
963− .insert(sender.clone(), Contact::new(sender.clone(), String::new(), ts));
1325+ // Auto-create a contact for a stranger who writes to us — but
1326+ // only up to `MAX_CONTACTS`. Past that, drop the message
1327+ // entirely: storing it would mean a `messages:<peer>` blob for
1328+ // a peer with no contact entry, i.e. a conversation nothing
1329+ // can open and nothing ever trims.
1330+ if !known {
1331+ if model.contacts.len() >= MAX_CONTACTS {
1332+ continue;
1333+ }
1334+ model.contacts.insert(
1335+ sender.clone(),
1336+ Contact::new(sender.clone(), String::new(), ts),
1337+ );
9641338 contacts_dirty = true;
9651339 }
9661340 let convo = model.messages.entry(sender.clone()).or_default();
@@ -1000,10 +1374,13 @@ impl Skrepka {
10001374 bio,
10011375 photo,
10021376 } => {
1003− let c = model
1004− .contacts
1005− .entry(sender.clone())
1006− .or_insert_with(|| Contact::new(sender.clone(), String::new(), ts));
1377+ // A bare profile from a stranger creates nothing. It is an
1378+ // unsolicited push with no message attached, so honouring it
1379+ // would let anyone holding our pubkey put an entry (with an
1380+ // avatar) in our contact list without ever saying a word.
1381+ let Some(c) = model.contacts.get_mut(&sender) else {
1382+ continue;
1383+ };
10071384 // Ignore stale profile replays (PROTOCOL.md §4). The clamped
10081385 // `ts` is what gets recorded, so a peer cannot park
10091386 // `last_profile_ts` at `i64::MAX` and freeze its own profile.
@@ -1018,14 +1395,26 @@ impl Skrepka {
10181395 }
10191396 }
10201397
1021− // Queue delivery acks (one payload per sender).
1398+ // Queue delivery acks, in batches the recipient will actually accept.
1399+ //
1400+ // `parse_payload` drops a `delivery.ack` carrying more than `MAX_ACK_IDS`
1401+ // ids, and a peer running this same code applies that cap to *us*. One
1402+ // payload per sender therefore silently stops working the moment a peer
1403+ // has more than `MAX_ACK_IDS` messages waiting for us — which is exactly
1404+ // what a long offline stretch produces — and every one of those messages
1405+ // stays un-delivered on their side forever, because a poll page is only
1406+ // ever acked once.
10221407 let ack_ts = now_ms();
10231408 for (sender, ids) in ack_targets {
1024− let payload = Payload::DeliveryAck { ack_ids: ids };
1025− model.outbox.push_back(OutboxItem {
1026− recipient: sender,
1027− envelope_json: protocol::serialize_payload(&payload, ack_ts),
1028− });
1409+ for batch in ids.chunks(protocol::MAX_ACK_IDS) {
1410+ let payload = Payload::DeliveryAck {
1411+ ack_ids: batch.to_vec(),
1412+ };
1413+ model.outbox.push_back(OutboxItem {
1414+ recipient: sender.clone(),
1415+ envelope_json: protocol::serialize_payload(&payload, ack_ts),
1416+ });
1417+ }
10291418 }
10301419
10311420 // Persist everything touched.
@@ -1042,6 +1431,16 @@ impl Skrepka {
10421431 }
10431432}
10441433
1434+/// Encrypt one outbox item to its recipient, or `None` if the recipient's key is
1435+/// unusable — a bad hex string, the wrong length, or 32 bytes that aren't a curve
1436+/// point. `AddContact` rejects all three now, but an older contact list (or a
1437+/// profile-broadcast fan-out over one) may still hold such a key.
1438+fn encrypt_for(sender: &Identity, item: &OutboxItem) -> Option<Vec<u8>> {
1439+ let recipient: [u8; 32] = hex::decode(&item.recipient).ok()?.try_into().ok()?;
1440+ let mut rng = rand_core::OsRng;
1441+ crate::crypto::encrypt(&mut rng, sender, &recipient, item.envelope_json.as_bytes()).ok()
1442+}
1443+
10451444fn backoff_ms(retries: u32) -> u64 {
10461445 (3000u64.saturating_mul(1u64 << retries.min(4))).min(30_000)
10471446}
@@ -1057,6 +1456,18 @@ fn trim_history(convo: &mut Vec<StoredMessage>) {
10571456}
10581457
10591458/// Decode a kv `get` result into a typed value, ignoring errors / absence.
1459+///
1460+/// `None` conflates three different things: the key is absent (a fresh install),
1461+/// the read failed, and the bytes are present but don't deserialize. The caller
1462+/// keeps its default in all three, and the next mutation writes that default back
1463+/// — so a blob we merely *failed to read* is destroyed rather than left alone.
1464+///
1465+/// The mitigation is upstream: every persisted type is `#[serde(default)]` (see
1466+/// `model.rs`), so adding a field can't make an old blob unreadable, which is by
1467+/// far the likeliest way this fires in practice. A genuinely corrupt blob is
1468+/// still lost. Fixing that properly means distinguishing the three cases here and
1469+/// refusing to overwrite on the third — a change to every caller and to the error
1470+/// surface, not a drive-by.
10601471fn parse_kv<T: for<'de> Deserialize<'de>>(res: KvData) -> Option<T> {
10611472 let bytes = res.ok().flatten()?;
10621473 serde_json::from_slice(&bytes).ok()
@@ -1070,7 +1481,7 @@ mod tests {
10701481 fn with_identity() -> Model {
10711482 let id = Identity::from_seed(&[3u8; 32]);
10721483 Model {
1073− secret_key: Some(id.secret_key.to_vec()),
1484+ secret_key: Some(Zeroizing::new(id.secret_key.to_vec())),
10741485 my_pubkey: id.public_key_hex(),
10751486 ..Default::default()
10761487 }
@@ -1080,6 +1491,24 @@ mod tests {
10801491 Identity::from_seed(&[seed; 32]).public_key_hex()
10811492 }
10821493
1494+ /// `Command::event` *queues* an event; it does not run it. The runtime drains
1495+ /// that queue back into `update`, so a test that stops at the first `update`
1496+ /// only ever sees half of what an event actually does — `Connect` would look
1497+ /// like it never authenticates, and the poll watchdog like it never re-polls.
1498+ /// Drain it the way the runtime does.
1499+ fn drive(app: &Skrepka, cmd: &mut Command<Effect, Event>, model: &mut Model) {
1500+ for event in cmd.events().collect::<Vec<_>>() {
1501+ let mut next = app.update(event, model);
1502+ drive(app, &mut next, model);
1503+ }
1504+ }
1505+
1506+ /// `update` + drain, i.e. what the shell actually observes.
1507+ fn dispatch(app: &Skrepka, event: Event, model: &mut Model) {
1508+ let mut cmd = app.update(event, model);
1509+ drive(app, &mut cmd, model);
1510+ }
1511+
10831512 /// A poll page carrying one payload, encrypted from `sender_seed` to `me`.
10841513 fn page_from(sender_seed: u8, me: &Model, payload: &Payload, ts: i64) -> PollResp {
10851514 let sender = Identity::from_seed(&[sender_seed; 32]);
@@ -1277,7 +1706,24 @@ mod tests {
12771706 fn server_host_strips_scheme_port_and_case() {
12781707 assert_eq!(server_host("https://Relay.Example.com:8443/x"), "relay.example.com");
12791708 assert_eq!(server_host("http://localhost:8080"), "localhost");
1280− assert_eq!(server_host("relay.example.com."), "relay.example.com");
1709+ assert_eq!(server_host("https://relay.example.com./x"), "relay.example.com");
1710+ // Its only caller passes `settings.server_url`, which is always an
1711+ // absolute URL (`normalize_server_url` guarantees it). Anything else is a
1712+ // bug, and an empty host produces a signature the relay rejects — which is
1713+ // the right failure: loud, not a silently-wrong host binding.
1714+ assert_eq!(server_host("relay.example.com"), "");
1715+ }
1716+
1717+ /// The auth signature is bound to the host (PROTOCOL.md §6), so `server_host`
1718+ /// has to agree with what the relay computes. Splitting on `:` by hand cut an
1719+ /// IPv6 literal at its first colon and produced `[`.
1720+ #[test]
1721+ fn server_host_handles_an_ipv6_literal() {
1722+ assert_eq!(server_host("http://[::1]:8080"), "[::1]");
1723+ assert_eq!(
1724+ server_host("https://[2001:DB8::1]/poll"),
1725+ "[2001:db8::1]"
1726+ );
12811727 }
12821728
12831729 #[test]
@@ -1371,16 +1817,63 @@ mod tests {
13711817 assert!(vm.contacts.iter().any(|c| c.pubkey == peer && c.blocked));
13721818 }
13731819
1820+ /// Gating ingest stops the acks we have not queued yet — but the ack for the
1821+ /// message that made the user hit "block" is already in the outbox, and an ack
1822+ /// is exactly the "we are online and reading you" signal the block exists to
1823+ /// cut off.
1824+ #[test]
1825+ fn blocking_a_peer_drops_what_is_already_queued_for_them() {
1826+ let app = Skrepka;
1827+ let mut m = with_identity();
1828+ let enemy = peer_hex(9);
1829+ let friend = peer_hex(11);
1830+ m.contacts
1831+ .insert(enemy.clone(), Contact::new(enemy.clone(), String::new(), 0));
1832+
1833+ // An ack to the peer we are about to block, plus unrelated mail.
1834+ for recipient in [&enemy, &friend] {
1835+ m.outbox.push_back(OutboxItem {
1836+ recipient: recipient.clone(),
1837+ envelope_json: protocol::serialize_payload(
1838+ &Payload::DeliveryAck {
1839+ ack_ids: vec!["m1".into()],
1840+ },
1841+ 1,
1842+ ),
1843+ });
1844+ }
1845+
1846+ let _ = app.update(
1847+ Event::SetBlocked {
1848+ peer: enemy.clone(),
1849+ blocked: true,
1850+ },
1851+ &mut m,
1852+ );
1853+
1854+ let recipients: Vec<&str> = m.outbox.iter().map(|o| o.recipient.as_str()).collect();
1855+ assert_eq!(
1856+ recipients,
1857+ vec![friend.as_str()],
1858+ "nothing more goes to a blocked peer — and everyone else is untouched"
1859+ );
1860+ }
1861+
13741862 #[test]
13751863 fn normalize_server_url_requires_an_http_scheme() {
1864+ // The real parser canonicalizes: scheme and host are lowercased.
13761865 assert_eq!(
13771866 normalize_server_url(" https://Relay.Example.com/ "),
1378− Some("https://Relay.Example.com".to_string())
1867+ Some("https://relay.example.com".to_string())
13791868 );
13801869 assert_eq!(
13811870 normalize_server_url("HTTP://localhost:8080///"),
13821871 Some("http://localhost:8080".to_string())
13831872 );
1873+ assert_eq!(
1874+ normalize_server_url("http://[::1]:8080"),
1875+ Some("http://[::1]:8080".to_string())
1876+ );
13841877 // The cases that used to panic Http::post or produce a dead URL.
13851878 assert_eq!(normalize_server_url("relay.example.com"), None);
13861879 assert_eq!(normalize_server_url("localhost:8080"), None);
@@ -1391,6 +1884,24 @@ mod tests {
13911884 assert_eq!(normalize_server_url(""), None);
13921885 }
13931886
1887+ /// The hand-rolled validator this replaces was not the parser it guarded:
1888+ /// it waved these through, and `crux_http` unwraps the parse.
1889+ #[test]
1890+ fn normalize_server_url_rejects_what_the_url_parser_rejects() {
1891+ assert_eq!(normalize_server_url("http://[bad"), None);
1892+ assert_eq!(normalize_server_url("http://ho^st"), None);
1893+ assert_eq!(normalize_server_url("http://a\u{7f}b"), None);
1894+ }
1895+
1896+ /// Every request appends a path, so a query or fragment would silently
1897+ /// swallow it: `http://x?y` + "/poll" is `http://x?y/poll` — a request to `/`
1898+ /// with a junk query, not to the poll endpoint.
1899+ #[test]
1900+ fn normalize_server_url_rejects_a_query_or_fragment() {
1901+ assert_eq!(normalize_server_url("http://x?y"), None);
1902+ assert_eq!(normalize_server_url("https://relay.example.com/#frag"), None);
1903+ }
1904+
13941905 #[test]
13951906 fn set_server_url_rejects_schemeless_and_keeps_the_old_url() {
13961907 let app = Skrepka;
@@ -1461,22 +1972,557 @@ mod tests {
14611972
14621973 let _ = app.update(Event::Poll, &mut m);
14631974 assert!(m.polling, "the first Poll issues a request");
1975+ let gen = m.poll_gen;
14641976
14651977 // A second Poll (e.g. from a re-auth) must not stack another loop.
14661978 let _ = app.update(Event::Poll, &mut m);
14671979 assert!(m.polling);
1980+ assert_eq!(m.poll_gen, gen, "and does not open a new generation");
14681981
14691982 // Both success and failure release the guard so the loop can continue.
14701983 let resp = crux_http::testing::ResponseBuilder::ok().body(Vec::new()).build();
1471− let _ = app.update(Event::PollResult(Ok(resp)), &mut m);
1984+ let _ = app.update(Event::PollResult(m.poll_gen, Ok(resp)), &mut m);
14721985 assert!(!m.polling);
14731986
14741987 let _ = app.update(Event::Poll, &mut m);
14751988 assert!(m.polling);
1476− let _ = app.update(Event::PollResult(Err(crux_http::HttpError::Timeout)), &mut m);
1989+ let _ = app.update(
1990+ Event::PollResult(m.poll_gen, Err(crux_http::HttpError::Timeout)),
1991+ &mut m,
1992+ );
14771993 assert!(!m.polling);
14781994 }
14791995
1996+ /// The cursor is the *relay's* sequence number, and the next poll acks
1997+ /// everything up to it. Carrying relay A's cursor over to relay B tells B to
1998+ /// delete every message waiting for us before handing any of them over.
1999+ #[test]
2000+ fn switching_relays_resets_the_cursor() {
2001+ let app = Skrepka;
2002+ let mut m = with_identity();
2003+ m.cursor = now_ms();
2004+ m.token = Some("t".into());
2005+ m.conn = ConnStatus::Online;
2006+
2007+ let _ = app.update(Event::SetServerUrl("https://other.example.com".into()), &mut m);
2008+
2009+ assert_eq!(m.cursor, 0, "the new relay is polled from the beginning");
2010+ assert!(m.token.is_none(), "the old relay's token is worthless");
2011+ assert_eq!(m.conn, ConnStatus::Offline);
2012+ }
2013+
2014+ /// Resetting the cursor is not enough on its own: the poll already in flight
2015+ /// is addressed to the *old* relay and its page carries the old relay's
2016+ /// sequence number. Letting it land writes that number straight back over the
2017+ /// reset, and the next poll hands it to the new relay — which reads it as an
2018+ /// ack and drops every message queued for us before serving any of them.
2019+ #[test]
2020+ fn switching_relays_retires_the_poll_still_in_flight_at_the_old_one() {
2021+ let app = Skrepka;
2022+ let mut m = with_identity();
2023+ m.token = Some("t".into());
2024+ m.conn = ConnStatus::Online;
2025+
2026+ let _ = app.update(Event::Poll, &mut m);
2027+ let old_relay_poll = m.poll_gen;
2028+ assert!(m.polling);
2029+
2030+ let _ = app.update(Event::SetServerUrl("https://other.example.com".into()), &mut m);
2031+ assert!(!m.polling, "the old relay's poll no longer holds the guard");
2032+
2033+ // It resolves after the switch, carrying the old relay's cursor.
2034+ let body = serde_json::to_vec(&serde_json::json!({
2035+ "events": [],
2036+ "cursor": 1_700_000_000_000i64,
2037+ }))
2038+ .unwrap();
2039+ let resp = crux_http::testing::ResponseBuilder::ok().body(body).build();
2040+ let _ = app.update(Event::PollResult(old_relay_poll, Ok(resp)), &mut m);
2041+
2042+ assert_eq!(
2043+ m.cursor, 0,
2044+ "the new relay must not be handed a cursor from a sequence it never issued"
2045+ );
2046+ }
2047+
2048+ /// A URL that the parser rejects reaches `Http::post`, which unwraps it. On
2049+ /// disk it is a panic on every launch — a boot loop with no way out.
2050+ #[test]
2051+ fn a_corrupt_persisted_server_url_falls_back_to_the_default() {
2052+ let app = Skrepka;
2053+ let mut m = with_identity();
2054+ m.loads_pending = STARTUP_LOADS;
2055+
2056+ let junk = serde_json::to_vec(&serde_json::json!({"server_url": "http://[bad"})).unwrap();
2057+ let _ = app.update(Event::LoadedSettings(Ok(Some(junk))), &mut m);
2058+
2059+ assert_eq!(m.settings.server_url, DEFAULT_SERVER_URL);
2060+ assert!(
2061+ normalize_server_url(&m.settings.server_url).is_some(),
2062+ "whatever we end up with must survive the parser"
2063+ );
2064+ }
2065+
2066+ /// A settings blob written before a field existed must still load — otherwise
2067+ /// `parse_kv` returns None, the default is used, and the next write destroys
2068+ /// what was there.
2069+ #[test]
2070+ fn an_old_settings_blob_still_loads() {
2071+ let app = Skrepka;
2072+ let mut m = with_identity();
2073+ m.loads_pending = STARTUP_LOADS;
2074+
2075+ let empty = serde_json::to_vec(&serde_json::json!({})).unwrap();
2076+ let _ = app.update(Event::LoadedSettings(Ok(Some(empty))), &mut m);
2077+ assert_eq!(m.settings.server_url, DEFAULT_SERVER_URL);
2078+
2079+ // ...and a Contact missing every optional field still becomes a contact.
2080+ let peer = peer_hex(9);
2081+ let old = serde_json::to_vec(&serde_json::json!([{"pubkey": peer}])).unwrap();
2082+ let _ = app.update(Event::LoadedContacts(Ok(Some(old))), &mut m);
2083+ assert!(m.contacts.contains_key(&peer));
2084+ }
2085+
2086+ /// Connecting before the startup loads land lets a poll be ingested against a
2087+ /// half-loaded model — and the loads still in flight then overwrite it.
2088+ #[test]
2089+ fn connect_waits_for_every_startup_load() {
2090+ let app = Skrepka;
2091+ let mut m = with_identity();
2092+ m.loads_pending = STARTUP_LOADS;
2093+
2094+ let connects = |cmd: &mut Command<Effect, Event>| {
2095+ cmd.events()
2096+ .filter(|e| matches!(e, Event::Connect))
2097+ .count()
2098+ };
2099+
2100+ for (i, ev) in [
2101+ Event::LoadedSettings(Ok(None)),
2102+ Event::LoadedProfile(Ok(None)),
2103+ Event::LoadedContacts(Ok(None)),
2104+ Event::LoadedCursor(Ok(None)),
2105+ ]
2106+ .into_iter()
2107+ .enumerate()
2108+ {
2109+ let mut cmd = app.update(ev, &mut m);
2110+ assert_eq!(
2111+ m.loads_pending,
2112+ STARTUP_LOADS - 1 - i as u8,
2113+ "each load ticks the counter down"
2114+ );
2115+ assert_eq!(connects(&mut cmd), 0, "and none of them connects");
2116+ }
2117+
2118+ // The fifth — and only the fifth — emits Connect.
2119+ let mut cmd = app.update(Event::LoadedOutbox(Ok(None)), &mut m);
2120+ assert_eq!(m.loads_pending, 0);
2121+ assert_eq!(connects(&mut cmd), 1, "the last load connects, exactly once");
2122+
2123+ // A late duplicate (a re-delivered load) must not connect again.
2124+ let mut cmd = app.update(Event::LoadedOutbox(Ok(None)), &mut m);
2125+ assert_eq!(connects(&mut cmd), 0);
2126+ }
2127+
2128+ /// `Authenticate` is reachable from startup, a 401 on poll, a 401 on send and
2129+ /// a reconnect timer. Without a guard a network blip fires all of them, and
2130+ /// every token that comes back starts its own poll loop.
2131+ #[test]
2132+ fn authentication_does_not_stack() {
2133+ let app = Skrepka;
2134+ let mut m = with_identity();
2135+
2136+ let _ = app.update(Event::Authenticate, &mut m);
2137+ assert!(m.authenticating, "the first one goes out");
2138+ assert_eq!(m.conn, ConnStatus::Connecting);
2139+
2140+ // A second Authenticate while the first is in flight is a no-op.
2141+ m.conn = ConnStatus::Offline;
2142+ let _ = app.update(Event::Authenticate, &mut m);
2143+ assert_eq!(m.conn, ConnStatus::Offline, "no second challenge was issued");
2144+
2145+ // Any terminal outcome releases the guard.
2146+ let _ = app.update(
2147+ Event::ChallengeResult(m.auth_gen, Err(crux_http::HttpError::Timeout)),
2148+ &mut m,
2149+ );
2150+ assert!(!m.authenticating);
2151+
2152+ let _ = app.update(Event::Authenticate, &mut m);
2153+ assert!(m.authenticating);
2154+ let _ = app.update(
2155+ Event::VerifyResult(m.auth_gen, Err(crux_http::HttpError::Timeout)),
2156+ &mut m,
2157+ );
2158+ assert!(!m.authenticating);
2159+ }
2160+
2161+ /// The shell fires `Connect` on *every* return to the foreground
2162+ /// (`SkrepkaApp.swift`), so a challenge issued before the app was backgrounded
2163+ /// is routinely still in flight when the next attempt starts. It must not come
2164+ /// back and install a token, or clobber `conn`, on top of the attempt that
2165+ /// replaced it.
2166+ #[test]
2167+ fn an_auth_round_trip_superseded_by_connect_cannot_land() {
2168+ let app = Skrepka;
2169+ let mut m = with_identity();
2170+
2171+ let _ = app.update(Event::Authenticate, &mut m);
2172+ let superseded = m.auth_gen;
2173+ assert!(m.authenticating);
2174+
2175+ // Foregrounding restarts the flow.
2176+ dispatch(&app, Event::Connect, &mut m);
2177+ let live = m.auth_gen;
2178+ assert_ne!(live, superseded, "the in-flight attempt was retired");
2179+ assert!(m.authenticating, "and a fresh one went out");
2180+
2181+ // The old challenge finally answers. It is not ours any more.
2182+ let body = serde_json::to_vec(&serde_json::json!({"challenge": "deadbeef"})).unwrap();
2183+ let resp = crux_http::testing::ResponseBuilder::ok().body(body).build();
2184+ let _ = app.update(Event::ChallengeResult(superseded, Ok(resp)), &mut m);
2185+ assert!(
2186+ m.authenticating,
2187+ "the live attempt still holds the guard — a stale leg must not release it"
2188+ );
2189+
2190+ // Nor may a stale *verify* install a token over the live attempt.
2191+ let body = serde_json::to_vec(&serde_json::json!({"token": "stale"})).unwrap();
2192+ let resp = crux_http::testing::ResponseBuilder::ok().body(body).build();
2193+ let _ = app.update(Event::VerifyResult(superseded, Ok(resp)), &mut m);
2194+ assert!(m.token.is_none(), "no token from a superseded attempt");
2195+ assert_ne!(m.conn, ConnStatus::Online);
2196+
2197+ // The live attempt still completes normally.
2198+ let body = serde_json::to_vec(&serde_json::json!({"token": "real"})).unwrap();
2199+ let resp = crux_http::testing::ResponseBuilder::ok().body(body).build();
2200+ let _ = app.update(Event::VerifyResult(live, Ok(resp)), &mut m);
2201+ assert_eq!(m.token.as_deref(), Some("real"));
2202+ assert_eq!(m.conn, ConnStatus::Online);
2203+ }
2204+
2205+ /// Blocking is meant to cut a peer off in *both* directions.
2206+ #[test]
2207+ fn a_blocked_peer_cannot_update_their_profile_or_ack_our_messages() {
2208+ let app = Skrepka;
2209+ let mut m = with_identity();
2210+ let peer = peer_hex(9);
2211+ let mut contact = Contact::new(peer.clone(), "Bob".into(), 0);
2212+ contact.blocked = true;
2213+ contact.display_name = "Bob".into();
2214+ m.contacts.insert(peer.clone(), contact);
2215+ m.messages.insert(
2216+ peer.clone(),
2217+ vec![StoredMessage {
2218+ id: "mine".into(),
2219+ body: "hi".into(),
2220+ ts: 1,
2221+ outgoing: true,
2222+ delivered: false,
2223+ }],
2224+ );
2225+
2226+ let profile = Payload::Profile {
2227+ display_name: "Mallory".into(),
2228+ bio: "spam".into(),
2229+ photo: Some("aGk=".into()),
2230+ };
2231+ let page = page_from(9, &m, &profile, now_ms());
2232+ let _ = app.ingest_poll(&mut m, page);
2233+ assert_eq!(
2234+ m.contacts[&peer].display_name, "Bob",
2235+ "a blocked peer cannot rewrite their entry in our contact list"
2236+ );
2237+
2238+ let ack = Payload::DeliveryAck {
2239+ ack_ids: vec!["mine".into()],
2240+ };
2241+ let page = page_from(9, &m, &ack, now_ms());
2242+ let _ = app.ingest_poll(&mut m, page);
2243+ assert!(
2244+ !m.messages[&peer][0].delivered,
2245+ "nor tell us they read what we sent before the block"
2246+ );
2247+ }
2248+
2249+ /// Blocking someone and then updating our profile must not hand them our new
2250+ /// display name, avatar, and a liveness signal.
2251+ #[test]
2252+ fn save_profile_does_not_broadcast_to_blocked_contacts() {
2253+ let app = Skrepka;
2254+ let mut m = with_identity();
2255+ let friend = peer_hex(9);
2256+ let enemy = peer_hex(11);
2257+ m.contacts
2258+ .insert(friend.clone(), Contact::new(friend.clone(), "Bob".into(), 0));
2259+ let mut blocked = Contact::new(enemy.clone(), "Mallory".into(), 0);
2260+ blocked.blocked = true;
2261+ m.contacts.insert(enemy.clone(), blocked);
2262+
2263+ let _ = app.update(
2264+ Event::SaveProfile {
2265+ display_name: "Alice".into(),
2266+ bio: String::new(),
2267+ photo: None,
2268+ },
2269+ &mut m,
2270+ );
2271+
2272+ let recipients: Vec<&str> = m.outbox.iter().map(|o| o.recipient.as_str()).collect();
2273+ assert_eq!(recipients, vec![friend.as_str()]);
2274+ }
2275+
2276+ /// Any stranger can create a contact just by writing to us, so the list needs
2277+ /// a ceiling — and past it, the message goes too: a stored conversation with
2278+ /// no contact entry is one nothing can open and nothing ever trims.
2279+ #[test]
2280+ fn contacts_from_strangers_are_capped() {
2281+ let app = Skrepka;
2282+ let mut m = with_identity();
2283+ for i in 0..MAX_CONTACTS {
2284+ let k = format!("{i:064x}");
2285+ m.contacts.insert(k.clone(), Contact::new(k, String::new(), 0));
2286+ }
2287+ assert_eq!(m.contacts.len(), MAX_CONTACTS);
2288+
2289+ let peer = peer_hex(9);
2290+ let page = page_from(9, &m, &text("m1", "hi"), now_ms());
2291+ let _ = app.ingest_poll(&mut m, page);
2292+
2293+ assert_eq!(m.contacts.len(), MAX_CONTACTS, "the cap holds");
2294+ assert!(!m.contacts.contains_key(&peer));
2295+ assert!(m.messages.get(&peer).is_none_or(Vec::is_empty));
2296+ assert!(m.outbox.is_empty(), "and nothing is acked back");
2297+ }
2298+
2299+ /// A bare profile is an unsolicited push with no message attached. Honouring
2300+ /// it would let anyone holding our pubkey install an entry — with an avatar —
2301+ /// in our contact list without ever saying a word.
2302+ #[test]
2303+ fn a_profile_from_a_stranger_does_not_create_a_contact() {
2304+ let app = Skrepka;
2305+ let mut m = with_identity();
2306+ let peer = peer_hex(9);
2307+
2308+ let profile = Payload::Profile {
2309+ display_name: "Mallory".into(),
2310+ bio: String::new(),
2311+ photo: Some("aGk=".into()),
2312+ };
2313+ let page = page_from(9, &m, &profile, now_ms());
2314+ let _ = app.ingest_poll(&mut m, page);
2315+ assert!(m.contacts.is_empty(), "no contact from a bare profile");
2316+
2317+ // But a profile from someone we *do* know still applies.
2318+ m.contacts
2319+ .insert(peer.clone(), Contact::new(peer.clone(), String::new(), 0));
2320+ let page = page_from(9, &m, &profile, now_ms());
2321+ let _ = app.ingest_poll(&mut m, page);
2322+ assert_eq!(m.contacts[&peer].display_name, "Mallory");
2323+ }
2324+
2325+ /// `polling` gates the whole long-poll loop and is cleared only by a
2326+ /// `PollResult`. If the shell ever drops the HTTP effect, the client goes
2327+ /// silent forever with no visible error.
2328+ #[test]
2329+ fn the_watchdog_unwedges_a_poll_that_never_resolved() {
2330+ let app = Skrepka;
2331+ let mut m = with_identity();
2332+ m.token = Some("t".into());
2333+
2334+ let _ = app.update(Event::Poll, &mut m);
2335+ assert!(m.polling);
2336+ let wedged = m.poll_gen;
2337+
2338+ // The watchdog armed for *this* poll releases the guard and restarts the
2339+ // loop — which issues a fresh poll under a new generation.
2340+ dispatch(&app, Event::PollWatchdog(wedged), &mut m);
2341+ assert!(m.polling, "the loop restarted");
2342+ assert_ne!(m.poll_gen, wedged, "under a new generation");
2343+ }
2344+
2345+ /// A watchdog is armed per poll and nothing cancels the earlier ones, so a
2346+ /// timer for a poll that has already completed keeps firing. It must not
2347+ /// touch the healthy poll that replaced it.
2348+ #[test]
2349+ fn a_stale_watchdog_leaves_the_current_poll_alone() {
2350+ let app = Skrepka;
2351+ let mut m = with_identity();
2352+ m.token = Some("t".into());
2353+
2354+ let _ = app.update(Event::Poll, &mut m);
2355+ let first = m.poll_gen;
2356+ let resp = crux_http::testing::ResponseBuilder::ok().body(Vec::new()).build();
2357+ let _ = app.update(Event::PollResult(first, Ok(resp)), &mut m);
2358+ let _ = app.update(Event::Poll, &mut m);
2359+ let current = m.poll_gen;
2360+ assert!(m.polling);
2361+
2362+ let _ = app.update(Event::PollWatchdog(first), &mut m);
2363+ assert!(m.polling, "the live poll is untouched");
2364+ assert_eq!(m.poll_gen, current, "and not superseded");
2365+ }
2366+
2367+ /// The regression the generation exists for.
2368+ ///
2369+ /// An HTTP effect cannot be cancelled, only abandoned — and it stays pending
2370+ /// across an iOS suspension while the wall clock runs on, so on resume the
2371+ /// watchdog gives up on a poll the shell is still about to resolve. If that
2372+ /// late `PollResult` were honoured it would clear `polling` — a guard that by
2373+ /// then belongs to the *replacement* poll — and chain a re-poll of its own.
2374+ /// Two loops, each re-polling itself forever, doubling on every recurrence.
2375+ #[test]
2376+ fn a_poll_abandoned_by_the_watchdog_cannot_fork_the_loop_when_it_lands() {
2377+ let app = Skrepka;
2378+ let mut m = with_identity();
2379+ m.token = Some("t".into());
2380+
2381+ let _ = app.update(Event::Poll, &mut m);
2382+ let abandoned = m.poll_gen;
2383+
2384+ // The watchdog gives up on it; the loop restarts under a new generation.
2385+ dispatch(&app, Event::PollWatchdog(abandoned), &mut m);
2386+ let live = m.poll_gen;
2387+ assert!(m.polling && live != abandoned);
2388+
2389+ // Now the abandoned request finally resolves. It must be ignored outright.
2390+ let resp = crux_http::testing::ResponseBuilder::ok().body(Vec::new()).build();
2391+ let mut cmd = app.update(Event::PollResult(abandoned, Ok(resp)), &mut m);
2392+
2393+ assert!(m.polling, "the live poll still holds the guard");
2394+ assert_eq!(m.poll_gen, live);
2395+ assert_eq!(
2396+ cmd.events().count(),
2397+ 0,
2398+ "and the corpse chains nothing — no second loop"
2399+ );
2400+ }
2401+
2402+ /// Dropping an abandoned page loses nothing: its cursor was never written, so
2403+ /// the relay still holds every message in it and re-delivers them.
2404+ #[test]
2405+ fn an_abandoned_poll_page_is_not_ingested() {
2406+ let app = Skrepka;
2407+ let mut m = with_identity();
2408+ m.token = Some("t".into());
2409+ let peer = peer_hex(9);
2410+
2411+ let _ = app.update(Event::Poll, &mut m);
2412+ let abandoned = m.poll_gen;
2413+ let page = page_from(9, &m, &text("m1", "hi"), now_ms());
2414+ let body = serde_json::to_vec(&serde_json::json!({
2415+ "events": [{"encryptedBlob": page.events[0].encrypted_blob}],
2416+ "cursor": 12345,
2417+ }))
2418+ .unwrap();
2419+
2420+ dispatch(&app, Event::PollWatchdog(abandoned), &mut m);
2421+ let resp = crux_http::testing::ResponseBuilder::ok().body(body).build();
2422+ let _ = app.update(Event::PollResult(abandoned, Ok(resp)), &mut m);
2423+
2424+ assert!(m.messages.get(&peer).is_none_or(Vec::is_empty));
2425+ assert_eq!(m.cursor, 0, "and the stale cursor is not adopted");
2426+ }
2427+
2428+ /// `view()` runs per keystroke (`ComposeChanged` renders), so cloning every
2429+ /// contact's base64 avatar each time meant megabytes of allocation to type one
2430+ /// message. Only the chat page has a composer, and while it is up the only
2431+ /// avatar on screen is the peer we are talking to — so that is the only page
2432+ /// that withholds anything.
2433+ ///
2434+ /// The conversations list draws an avatar per row (`ConversationsView.row`).
2435+ /// Blanking photos there buys nothing — no keystroke renders that page — and
2436+ /// costs every contact their picture.
2437+ #[test]
2438+ fn photos_are_withheld_on_the_chat_page_and_nowhere_else() {
2439+ let app = Skrepka;
2440+ let mut m = with_identity();
2441+ let a = peer_hex(9);
2442+ let b = peer_hex(11);
2443+ for peer in [&a, &b] {
2444+ let mut c = Contact::new(peer.clone(), String::new(), 0);
2445+ c.photo = Some("aGVsbG8=".into());
2446+ m.contacts.insert(peer.clone(), c);
2447+ }
2448+ let photo_of = |vm: &ViewModel, k: &str| {
2449+ vm.contacts
2450+ .iter()
2451+ .find(|c| c.pubkey == k)
2452+ .map(|c| c.photo.clone())
2453+ .unwrap()
2454+ };
2455+
2456+ // Conversations list: every row keeps its avatar.
2457+ let vm = app.view(&m);
2458+ assert_eq!(photo_of(&vm, &a), "aGVsbG8=");
2459+ assert_eq!(photo_of(&vm, &b), "aGVsbG8=");
2460+
2461+ // Open a chat: exactly that peer's photo, and no one else's — this is the
2462+ // page that re-renders on every keystroke.
2463+ m.page = Page::Chat;
2464+ m.active_peer = Some(a.clone());
2465+ let vm = app.view(&m);
2466+ assert_eq!(photo_of(&vm, &a), "aGVsbG8=");
2467+ assert!(photo_of(&vm, &b).is_empty());
2468+ }
2469+
2470+ /// `parse_payload` drops a `delivery.ack` with more than `MAX_ACK_IDS` ids, and
2471+ /// a peer running this code applies that cap to us. One ack payload per sender
2472+ /// therefore stops working the moment a peer has more than the cap waiting for
2473+ /// us — a long offline stretch — and those messages stay un-delivered on their
2474+ /// side forever, because a page is only ever acked once.
2475+ #[test]
2476+ fn acks_are_batched_so_the_peer_can_actually_parse_them() {
2477+ let app = Skrepka;
2478+ let mut m = with_identity();
2479+ let peer = peer_hex(9);
2480+ m.contacts
2481+ .insert(peer.clone(), Contact::new(peer.clone(), String::new(), 0));
2482+
2483+ // One page holding more messages from one sender than fits in a single ack.
2484+ let count = protocol::MAX_ACK_IDS + 5;
2485+ let sender = Identity::from_seed(&[9u8; 32]);
2486+ let me = Identity::from_secret_bytes(m.secret_key.as_ref().unwrap()).unwrap();
2487+ let events = (0..count)
2488+ .map(|i| {
2489+ let json = protocol::serialize_payload(&text(&format!("m{i}"), "hi"), now_ms());
2490+ let blob = crate::crypto::encrypt(
2491+ &mut rand_core::OsRng,
2492+ &sender,
2493+ &me.public_key(),
2494+ json.as_bytes(),
2495+ )
2496+ .unwrap();
2497+ PollEvent {
2498+ encrypted_blob: hex::encode(blob),
2499+ }
2500+ })
2501+ .collect();
2502+ let _ = app.ingest_poll(&mut m, PollResp { events, cursor: 1 });
2503+
2504+ // Local history is capped as always — but every message we *received* is
2505+ // acked, whether or not we chose to keep it.
2506+ assert_eq!(m.messages[&peer].len(), MAX_MESSAGES_PER_PEER);
2507+ assert_eq!(m.outbox.len(), 2, "the acks are split into parseable batches");
2508+
2509+ // And every batch really does survive the recipient's parser.
2510+ let mut acked = 0;
2511+ for item in &m.outbox {
2512+ assert_eq!(item.recipient, peer);
2513+ let parsed = protocol::parse_payload(item.envelope_json.as_bytes())
2514+ .expect("a peer must be able to parse the ack we send it");
2515+ match parsed.payload {
2516+ Payload::DeliveryAck { ack_ids } => {
2517+ assert!(ack_ids.len() <= protocol::MAX_ACK_IDS);
2518+ acked += ack_ids.len();
2519+ }
2520+ _ => panic!("expected a delivery.ack"),
2521+ }
2522+ }
2523+ assert_eq!(acked, count, "and between them they ack the whole page");
2524+ }
2525+
14802526 #[test]
14812527 fn view_exposes_identity_and_ob() {
14822528 let app = Skrepka;