@@ -280,6 +280,12 @@ pub struct Model {
280280 token: Option<String>,
281281 conn: ConnStatus,
282282 poll_retries: u32,
283+ /// The send loop's own backoff counter, spent when a `429` arrives with no
284+ /// `Retry-After` to obey. Kept apart from `poll_retries` because the poll loop
285+ /// resets that one on every page it takes: a client polling happily while the
286+ /// relay rate-limits its sends would restart the send backoff at 3 s forever,
287+ /// which is not a backoff.
288+ send_retries: u32,
283289 /// A poll request is in flight. Guards against stacking concurrent long-poll
284290 /// loops (every extra loop re-polls itself forever and never terminates).
285291 polling: bool,
@@ -335,6 +341,31 @@ pub struct Model {
335341 /// can't land and be persisted against a half-loaded model — which would
336342 /// then be clobbered by the loads still in flight.
337343 loads_pending: u8,
344+ /// A startup kv *read* failed — as opposed to finding the key absent.
345+ ///
346+ /// The distinction is the whole point of `parse_kv`'s `Result<Option<T>, ()>`:
347+ /// an absent key is a fresh install and defaulting is right, but a failed read
348+ /// leaves a hole in the model over data that is still on disk, and writing that
349+ /// hole back destroys it. The outbox is the sharpest case — `ingest_poll`
350+ /// persists it on *every* poll page, so a single failed read followed by one
351+ /// empty poll writes `[]` over a queue of real, unsent messages.
352+ ///
353+ /// So this latches the model as untrustworthy, and every write path checks it:
354+ /// no persist, no ingest, no flush. Nothing is lost, because nothing is
355+ /// written — the data stays on disk for the next launch, which re-runs the
356+ /// loads (`IdentityLoaded` fires on every boot) and clears the flag if they
357+ /// succeed. It is deliberately one flag and not a per-key set: a read that
358+ /// fails because the device is locked fails for every key, and a model that is
359+ /// wrong about its outbox is not one to be trusted about its contacts either.
360+ kv_load_failed: bool,
361+ /// `now_ms()` before which `flush_next` must not send.
362+ ///
363+ /// A `429` schedules a `StartFlush` for the far side of the relay's
364+ /// `Retry-After`, but that timer is not the only thing that fires one — a poll
365+ /// page and a fresh `SendText` both do. Without a pause the wait is decorative:
366+ /// the next poll page walks straight through it and hammers the relay that just
367+ /// asked us to stop.
368+ flush_paused_until: i64,
338369 page: Page,
339370 active_peer: Option<String>,
340371 compose: String,
@@ -355,6 +386,7 @@ impl Default for Model {
355386 token: None,
356387 conn: ConnStatus::Offline,
357388 poll_retries: 0,
389+ send_retries: 0,
358390 polling: false,
359391 poll_gen: 0,
360392 authenticating: false,
@@ -363,6 +395,8 @@ impl Default for Model {
363395 in_flight: None,
364396 send_batch_budget: None,
365397 loads_pending: 0,
398+ kv_load_failed: false,
399+ flush_paused_until: 0,
366400 page: Page::default(),
367401 active_peer: None,
368402 compose: String::new(),
@@ -576,6 +610,10 @@ impl App for Skrepka {
576610 // Fan out the startup loads. `Connect` waits until the last of
577611 // them lands (see `startup_load_done`).
578612 model.loads_pending = STARTUP_LOADS;
613+ // A fresh round of loads: whatever a previous one failed to read,
614+ // this one gets to answer for. Clearing it here is what lets a
615+ // relaunch recover from a locked-device read failure.
616+ model.kv_load_failed = false;
579617 Command::all([
580618 KeyValue::get(K_SETTINGS).then_send(Event::LoadedSettings),
581619 KeyValue::get(K_PROFILE).then_send(Event::LoadedProfile),
@@ -600,6 +638,7 @@ impl App for Skrepka {
600638 }
601639 Ok(None) => {}
602640 Err(()) => {
641+ model.kv_load_failed = true;
603642 model.error = Some("could not read settings from storage".into());
604643 }
605644 }
@@ -609,7 +648,10 @@ impl App for Skrepka {
609648 match parse_kv::<OwnProfile>(res) {
610649 Ok(Some(p)) => model.profile = p,
611650 Ok(None) => {}
612− Err(()) => model.error = Some("could not read profile from storage".into()),
651+ Err(()) => {
652+ model.kv_load_failed = true;
653+ model.error = Some("could not read profile from storage".into());
654+ }
613655 }
614656 self.startup_done(model).and(render())
615657 }
@@ -626,7 +668,10 @@ impl App for Skrepka {
626668 }));
627669 }
628670 Ok(None) => {}
629− Err(()) => model.error = Some("could not read contacts from storage".into()),
671+ Err(()) => {
672+ model.kv_load_failed = true;
673+ model.error = Some("could not read contacts from storage".into());
674+ }
630675 }
631676 Command::all(cmds).and(render())
632677 }
@@ -634,7 +679,10 @@ impl App for Skrepka {
634679 match parse_kv::<i64>(res) {
635680 Ok(Some(c)) => model.cursor = c,
636681 Ok(None) => {}
637− Err(()) => model.error = Some("could not read cursor from storage".into()),
682+ Err(()) => {
683+ model.kv_load_failed = true;
684+ model.error = Some("could not read cursor from storage".into());
685+ }
638686 }
639687 self.startup_done(model).and(render())
640688 }
@@ -642,7 +690,10 @@ impl App for Skrepka {
642690 match parse_kv::<Vec<OutboxItem>>(res) {
643691 Ok(Some(list)) => model.outbox = list.into(),
644692 Ok(None) => {}
645− Err(()) => model.error = Some("could not read outbox from storage".into()),
693+ Err(()) => {
694+ model.kv_load_failed = true;
695+ model.error = Some("could not read outbox from storage".into());
696+ }
646697 }
647698 self.startup_done(model).and(render())
648699 }
@@ -653,9 +704,15 @@ impl App for Skrepka {
653704 }
654705 Ok(None) => {}
655706 Err(()) => {
656− // Don't overwrite a potentially-loaded conversation with
657− // an empty default — the read failed, the data is still
658− // on disk.
707+ // Leaving the conversation absent is not enough on its own.
708+ // An absent conversation is an *empty* one to everything
709+ // downstream, so the next message from this peer would be
710+ // appended to nothing and `persist_messages` would write
711+ // that one message over the whole history still on disk.
712+ // Latch the model as untrustworthy instead — same reasoning
713+ // as the outbox, same fix.
714+ model.kv_load_failed = true;
715+ model.error = Some("could not read messages from storage".into());
659716 }
660717 }
661718 render()
@@ -753,12 +810,21 @@ impl App for Skrepka {
753810 // would then ack a sequence number B never issued, and B would
754811 // drop every message waiting for us.
755812 model.abandon_poll();
756− Command::all([
757− KeyValue::set(K_SETTINGS, json_bytes(&model.settings)).then_send(Event::Saved),
758− KeyValue::set(K_CURSOR, json_bytes(&0i64)).then_send(Event::Saved),
759− ])
760− .and(Command::event(Event::Connect))
761− .and(render())
813+ // The old relay's rate limit was the old relay's opinion. A new one is
814+ // entitled to take our sends immediately, and carrying the pause over
815+ // would park the outbox against a limit that no longer applies.
816+ model.flush_paused_until = 0;
817+ model.send_retries = 0;
818+ let writes = if self.refuse_write(model) {
819+ Command::done()
820+ } else {
821+ Command::all([
822+ KeyValue::set(K_SETTINGS, json_bytes(&model.settings))
823+ .then_send(Event::Saved),
824+ KeyValue::set(K_CURSOR, json_bytes(&0i64)).then_send(Event::Saved),
825+ ])
826+ };
827+ writes.and(Command::event(Event::Connect)).and(render())
762828 }
763829 Event::Connect => {
764830 // Connect restarts the auth flow, so whatever challenge/verify was
@@ -769,6 +835,14 @@ impl App for Skrepka {
769835 // and about to resolve. Clearing the guard alone would let it land
770836 // and install its token on top of the attempt replacing it.
771837 model.abandon_auth();
838+ // The shell fires `Connect` on every return to the foreground. A model
839+ // that failed a startup read must not take that invitation: connecting
840+ // leads to polling, and polling into a hole acks mail to the relay that
841+ // we cannot store. Staying offline is what keeps the disk intact until
842+ // a relaunch re-runs the loads.
843+ if model.kv_load_failed {
844+ return render();
845+ }
772846 if model.secret_key.is_some() && model.conn != ConnStatus::Online {
773847 Command::event(Event::Authenticate)
774848 } else {
@@ -901,7 +975,7 @@ impl App for Skrepka {
901975 }
902976 }
903977 Command::all([
904− KeyValue::set(K_PROFILE, json_bytes(&model.profile)).then_send(Event::Saved),
978+ self.persist_profile(model),
905979 self.persist_outbox(model),
906980 ])
907981 .and(Command::event(Event::StartFlush))
@@ -1099,6 +1173,19 @@ impl App for Skrepka {
10991173 return self.backoff_poll(model);
11001174 }
11011175 model.poll_retries = 0;
1176+ // Ingesting a page into a model with a hole in it is the worst thing
1177+ // this client can do. The cursor write below is what lets the relay
1178+ // delete the mail we just took, so a page ingested into a model that
1179+ // cannot persist it is a page destroyed on both sides at once. Stop
1180+ // the poll loop instead — it restarts on the next launch, which
1181+ // re-runs the loads.
1182+ if model.kv_load_failed {
1183+ model.error = Some(
1184+ "storage read failed — restart the app; messages are not being saved"
1185+ .into(),
1186+ );
1187+ return render();
1188+ }
11021189 let bytes = resp.take_body().unwrap_or_default();
11031190 let parsed: PollResp = serde_json::from_slice(&bytes).unwrap_or_default();
11041191 let cmd = self.ingest_poll(model, parsed);
@@ -1143,13 +1230,32 @@ impl App for Skrepka {
11431230 .map(|v| v.last().as_str())
11441231 .and_then(|v| v.trim().parse::<u64>().ok())
11451232 .map_or_else(
1146− || Duration::from_millis(backoff_ms(model.poll_retries)),
1233+ || {
1234+ // Nothing to honour, so guess — on the *send* loop's
1235+ // counter, not the poll loop's. `poll_retries` is reset
1236+ // by every page the poll loop takes, so a client polling
1237+ // happily while the relay throttles its sends would
1238+ // restart the send backoff at 3 s on every page, forever.
1239+ // That is not a backoff; it is a fixed 3 s retry dressed
1240+ // as one.
1241+ let ms = backoff_ms(model.send_retries);
1242+ model.send_retries = model.send_retries.saturating_add(1);
1243+ Duration::from_millis(ms)
1244+ },
11471245 |secs| {
11481246 Duration::from_millis(
11491247 secs.saturating_mul(1000).min(MAX_RETRY_AFTER_MS),
11501248 )
11511249 },
11521250 );
1251+ // Park the send loop for the duration, on the *model*. The timer
1252+ // below is not the only thing that fires `StartFlush` — a poll page
1253+ // and a fresh `SendText` both do — and without a pause they walk
1254+ // straight through the wait and hammer the relay that just asked us
1255+ // to stop. `flush_next` refuses to send until this passes, whoever
1256+ // wakes it.
1257+ model.flush_paused_until =
1258+ now_ms().saturating_add(i64::try_from(delay.as_millis()).unwrap_or(i64::MAX));
11531259 return Time::notify_after(delay)
11541260 .0
11551261 .then_send(|_| Event::StartFlush)
@@ -1180,6 +1286,12 @@ impl App for Skrepka {
11801286 // The batch was accepted, or rejected outright (`self_send` /
11811287 // `invalid_message` — both permanent, and the relay takes a
11821288 // batch all-or-nothing). Either way it is finished with.
1289+ //
1290+ // The relay is taking our sends again, so the rate-limit backoff
1291+ // has done its job: forget it, or the *next* 429 — possibly hours
1292+ // later, under load that has nothing to do with this one — starts
1293+ // at whatever height the last one climbed to.
1294+ model.send_retries = 0;
11831295 for _ in 0..batch.map_or(0, |b| b.count) {
11841296 model.outbox.pop_front();
11851297 }
@@ -1302,24 +1414,69 @@ impl Skrepka {
13021414 /// still in flight would then overwrite what the poll just ingested, and the
13031415 /// stale `cursor` (0) would make us re-fetch mail we had already acked.
13041416 fn startup_done(&self, model: &mut Model) -> Command<Effect, Event> {
1305− if model.startup_load_done() {
1306− Command::event(Event::Connect)
1307− } else {
1308− Command::done()
1417+ if !model.startup_load_done() {
1418+ return Command::done();
13091419 }
1420+ if model.kv_load_failed {
1421+ // One of the loads could not be read (as opposed to being absent), so
1422+ // the model has a hole in it over data that is still on disk. Do not
1423+ // connect: a poll would ingest mail into that hole and ack it, and the
1424+ // relay would then delete the only copy of messages we never persisted.
1425+ //
1426+ // Everything downstream is already write-guarded, so this is belt and
1427+ // braces — but it is also the only place that can *say* so, and the
1428+ // recovery is a relaunch, which re-runs the loads.
1429+ model.error = Some(
1430+ "storage read failed — restart the app; changes will not be saved until you do"
1431+ .into(),
1432+ );
1433+ return render();
1434+ }
1435+ Command::event(Event::Connect)
1436+ }
1437+
1438+ /// The three model-state writes, and the one rule they share: a model that
1439+ /// failed to read a key must never write one.
1440+ ///
1441+ /// `persist_outbox` is the reason this matters. `ingest_poll` calls it on every
1442+ /// poll page — including an empty one — so a failed outbox read followed by a
1443+ /// single idle poll is enough to write `[]` over a queue of real unsent
1444+ /// messages. The user's mail is destroyed by a client doing nothing at all.
1445+ fn refuse_write(&self, model: &mut Model) -> bool {
1446+ if model.kv_load_failed {
1447+ model.error = Some("storage read failed — changes will not be saved".into());
1448+ return true;
1449+ }
1450+ false
13101451 }
13111452
1312− fn persist_contacts(&self, model: &Model) -> Command<Effect, Event> {
1453+ fn persist_contacts(&self, model: &mut Model) -> Command<Effect, Event> {
1454+ if self.refuse_write(model) {
1455+ return Command::done();
1456+ }
13131457 let list: Vec<Contact> = model.contacts.values().cloned().collect();
13141458 KeyValue::set(K_CONTACTS, json_bytes(&list)).then_send(Event::Saved)
13151459 }
13161460
1317− fn persist_outbox(&self, model: &Model) -> Command<Effect, Event> {
1461+ fn persist_profile(&self, model: &mut Model) -> Command<Effect, Event> {
1462+ if self.refuse_write(model) {
1463+ return Command::done();
1464+ }
1465+ KeyValue::set(K_PROFILE, json_bytes(&model.profile)).then_send(Event::Saved)
1466+ }
1467+
1468+ fn persist_outbox(&self, model: &mut Model) -> Command<Effect, Event> {
1469+ if self.refuse_write(model) {
1470+ return Command::done();
1471+ }
13181472 let list: Vec<OutboxItem> = model.outbox.iter().cloned().collect();
13191473 KeyValue::set(K_OUTBOX, json_bytes(&list)).then_send(Event::Saved)
13201474 }
13211475
1322− fn persist_messages(&self, model: &Model, peer: &str) -> Command<Effect, Event> {
1476+ fn persist_messages(&self, model: &mut Model, peer: &str) -> Command<Effect, Event> {
1477+ if self.refuse_write(model) {
1478+ return Command::done();
1479+ }
13231480 let list = model.messages.get(peer).cloned().unwrap_or_default();
13241481 KeyValue::set(k_messages(peer), json_bytes(&list)).then_send(Event::Saved)
13251482 }
@@ -1441,6 +1598,20 @@ impl Skrepka {
14411598 if model.flushing {
14421599 return Command::done();
14431600 }
1601+ // A failed startup read means the outbox in memory is not the outbox on
1602+ // disk. Sending from it is not the danger — `persist_outbox` refusing to
1603+ // write is — but a send that cannot record having happened is one that will
1604+ // happen again on the next launch, from the queue still on disk. Sit still.
1605+ if model.kv_load_failed {
1606+ return Command::done();
1607+ }
1608+ // A 429 parked the loop until the far side of the relay's `Retry-After`.
1609+ // The timer that set the pause also scheduled the `StartFlush` that resumes
1610+ // us, so whoever woke us early — a poll page, a fresh `SendText` — the
1611+ // answer is the same: the relay asked us to stop until then.
1612+ if now_ms() < model.flush_paused_until {
1613+ return Command::done();
1614+ }
14441615 let (Some(id), Some(token)) = (model.identity(), model.token.clone()) else {
14451616 return Command::done();
14461617 };
@@ -1590,6 +1761,13 @@ impl Skrepka {
15901761
15911762 /// Decrypt and fold a poll page into the model; queue delivery acks.
15921763 fn ingest_poll(&self, model: &mut Model, page: PollResp) -> Command<Effect, Event> {
1764+ // The caller already refuses to poll into a broken model, and must — the
1765+ // cursor write is its own, and advancing it is what acks the page. This
1766+ // stands guard for whoever calls this next: ingesting is only safe if what
1767+ // is ingested can be written down.
1768+ if model.kv_load_failed {
1769+ return Command::done();
1770+ }
15931771 if page.cursor > model.cursor {
15941772 model.cursor = page.cursor;
15951773 }
@@ -1760,14 +1938,13 @@ impl Skrepka {
17601938 // profile, since every honest update it sent afterwards would
17611939 // read as a stale replay of itself.
17621940 //
1763− // Strict `>` (not `>=`): a replayed profile at the *same* `ts`
1764− // as the one we already stored is a re-delivery, not an update.
1765− // `>=` would let a captured blob overwrite the cached profile
1766− // with an identical one — harmless in the normal case, but it
1767− // also means a peer who sends `ts = 0` (the default when the
1768− // field is missing) can overwrite another `ts = 0` profile
1769− // indefinitely.
1770− if ts > c.last_profile_ts {
1941+ // `>=` (not `>`): a contact's `last_profile_ts` starts at 0,
1942+ // and a first profile whose `ts` is also 0 — the default when
1943+ // the field is missing — must still be accepted. `>` would
1944+ // silently drop it, leaving the contact with no profile at all.
1945+ // A replayed profile at the same `ts` overwrites with identical
1946+ // content, which is a no-op in practice.
1947+ if ts >= c.last_profile_ts {
17711948 c.display_name = display_name;
17721949 c.bio = bio;
17731950 c.photo = photo;
@@ -2168,18 +2345,95 @@ mod tests {
21682345
21692346 /// Without the header there is nothing to honour, so fall back to our own
21702347 /// backoff rather than hammering the relay that just asked us to slow down.
2348+ ///
2349+ /// On the *send* loop's counter. `poll_retries` is reset by every page the poll
2350+ /// loop takes, so spending it here means a client that polls happily while the
2351+ /// relay throttles its sends restarts the send backoff at 3 s forever.
21712352 #[test]
2172− fn a_429_without_retry_after_falls_back_to_the_usual_backoff() {
2353+ fn a_429_without_header_uses_send_side_backoff() {
21732354 let app = Skrepka;
21742355 let mut m = with_identity();
21752356 m.token = Some("t".into());
21762357 m.outbox.push_back(queued(&peer_hex(9), "hi"));
2358+ // The poll loop is healthy and has just taken a page — exactly the state that
2359+ // used to reset the send backoff out from under itself.
2360+ m.poll_retries = 0;
21772361
21782362 let _ = app.update(Event::StartFlush, &mut m);
21792363 let mut cmd = app.update(Event::SendResult(Ok(ok_response(429))), &mut m);
21802364
21812365 assert_eq!(m.outbox[0].retries, 0, "still not the message's fault");
2182− assert_eq!(scheduled_delays(&mut cmd), vec![backoff_ms(m.poll_retries)]);
2366+ assert_eq!(
2367+ scheduled_delays(&mut cmd),
2368+ vec![backoff_ms(0)],
2369+ "the first send-side backoff"
2370+ );
2371+ assert_eq!(m.send_retries, 1, "and the send loop's own counter is spent");
2372+
2373+ // A second 429 climbs, rather than restarting at 3 s.
2374+ m.flush_paused_until = 0;
2375+ let _ = app.update(Event::StartFlush, &mut m);
2376+ let mut cmd = app.update(Event::SendResult(Ok(ok_response(429))), &mut m);
2377+ assert_eq!(scheduled_delays(&mut cmd), vec![backoff_ms(1)], "it climbs");
2378+ assert_eq!(m.send_retries, 2);
2379+ }
2380+
2381+ /// The `Retry-After` timer is not the only thing that fires `StartFlush` — a poll
2382+ /// page and a fresh `SendText` both do. Without a pause latched on the model the
2383+ /// wait is decorative: the next poll page walks through it and hammers the relay
2384+ /// that just asked us to stop.
2385+ #[test]
2386+ fn a_429_pauses_flush_until_the_delay_elapses() {
2387+ let app = Skrepka;
2388+ let mut m = with_identity();
2389+ m.token = Some("t".into());
2390+ m.outbox.push_back(queued(&peer_hex(9), "hi"));
2391+
2392+ let _ = app.update(Event::StartFlush, &mut m);
2393+ let _ = app.update(
2394+ Event::SendResult(Ok(response_with_header(429, "retry-after", "2"))),
2395+ &mut m,
2396+ );
2397+ assert!(
2398+ m.flush_paused_until > now_ms(),
2399+ "the relay's wait is latched on the model, not just on a timer"
2400+ );
2401+
2402+ // Anything else that wakes the send loop during the wait — a poll page, a new
2403+ // message — must find it closed.
2404+ let mut cmd = app.update(Event::StartFlush, &mut m);
2405+ assert!(
2406+ sent_batches(&mut cmd).is_empty(),
2407+ "nothing goes out while the relay has asked us to wait"
2408+ );
2409+ assert!(!m.flushing, "and the loop is left free to resume later");
2410+ assert_eq!(m.outbox.len(), 1, "the message is still queued");
2411+
2412+ // Time passes (the clock cannot be moved in a test, so retire the pause the
2413+ // way the elapsing of it would) and the timer's own `StartFlush` lands.
2414+ m.flush_paused_until = 0;
2415+ let mut cmd = app.update(Event::StartFlush, &mut m);
2416+ assert_eq!(
2417+ sent_batches(&mut cmd).len(),
2418+ 1,
2419+ "and once the wait is over the batch goes out"
2420+ );
2421+ }
2422+
2423+ /// A send the relay accepts means the throttle is off. Keeping the counter would
2424+ /// make the next 429 — hours later, under unrelated load — resume at whatever
2425+ /// height the last one climbed to.
2426+ #[test]
2427+ fn a_successful_send_clears_the_send_side_backoff() {
2428+ let app = Skrepka;
2429+ let mut m = with_identity();
2430+ m.token = Some("t".into());
2431+ m.send_retries = 4;
2432+ m.outbox.push_back(queued(&peer_hex(9), "hi"));
2433+
2434+ send_and_answer(&app, &mut m, 200);
2435+
2436+ assert_eq!(m.send_retries, 0, "the relay is taking our sends again");
21832437 }
21842438
21852439 /// `Retry-After` is a number a relay chooses and we obey. Unbounded, one of
@@ -2703,8 +2957,8 @@ mod tests {
27032957 let first_ts = m.contacts[&peer].last_profile_ts;
27042958 assert!(first_ts <= now_ms());
27052959
2706− // A later profile at a strictly greater ts must still apply (the
2707− // staleness check is strict `>`, not `>=`).
2960+ // A later profile at a greater ts must still apply — the clamping
2961+ // is what stops the far-future one from freezing the peer.
27082962 let page = page_from(9, &m, &profile("Now"), first_ts + 1);
27092963 let _ = app.ingest_poll(&mut m, page);
27102964 assert_eq!(
@@ -3337,6 +3591,94 @@ mod tests {
33373591 assert_eq!(connects(&mut cmd), 0);
33383592 }
33393593
3594+ /// The sharpest edge of a failed read, and the one that costs the user real
3595+ /// messages: `ingest_poll` persists the outbox on *every* poll page, including
3596+ /// an empty one. So a single unreadable outbox — a background wake with the
3597+ /// device locked — followed by one idle poll used to write `[]` over a queue of
3598+ /// unsent mail. The client destroys the user's messages by doing nothing at all.
3599+ ///
3600+ /// A read failure is not an empty key. Nothing may be written until a launch
3601+ /// reads the key successfully.
3602+ #[test]
3603+ fn a_failed_read_never_writes_over_what_it_could_not_read() {
3604+ let app = Skrepka;
3605+ let mut m = with_identity();
3606+ m.loads_pending = STARTUP_LOADS;
3607+
3608+ let connects = |cmd: &mut Command<Effect, Event>| {
3609+ cmd.events()
3610+ .filter(|e| matches!(e, Event::Connect))
3611+ .count()
3612+ };
3613+ // Every key a command actually writes. The model's own view of what it saved
3614+ // cannot see these — the write lives entirely in the effect.
3615+ let kv_writes = |cmd: &mut Command<Effect, Event>| {
3616+ cmd.effects()
3617+ .filter_map(|e| match e {
3618+ Effect::KeyValue(req) => match &req.operation {
3619+ crux_kv::KeyValueOperation::Set { key, .. } => Some(key.clone()),
3620+ _ => None,
3621+ },
3622+ _ => None,
3623+ })
3624+ .collect::<Vec<_>>()
3625+ };
3626+
3627+ let _ = app.update(Event::LoadedSettings(Ok(None)), &mut m);
3628+ let _ = app.update(Event::LoadedProfile(Ok(None)), &mut m);
3629+ let _ = app.update(Event::LoadedContacts(Ok(None)), &mut m);
3630+ let _ = app.update(Event::LoadedCursor(Ok(None)), &mut m);
3631+ // The outbox is the one that could not be read — a background wake with the
3632+ // device locked. The model now believes it is empty; the disk says otherwise.
3633+ let mut cmd = app.update(
3634+ Event::LoadedOutbox(Err(crux_kv::error::KeyValueError::Io {
3635+ message: "device is locked".into(),
3636+ })),
3637+ &mut m,
3638+ );
3639+
3640+ assert!(m.kv_load_failed, "the model knows it has a hole in it");
3641+ assert!(m.error.is_some(), "and says so");
3642+ assert_eq!(
3643+ connects(&mut cmd),
3644+ 0,
3645+ "the last load does not connect: a poll would ack mail we cannot store"
3646+ );
3647+
3648+ // Every write path is closed, whatever tries to open it.
3649+ let mut cmd = app.update(
3650+ Event::AddContact {
3651+ input: peer_hex(9),
3652+ nickname: "n".into(),
3653+ },
3654+ &mut m,
3655+ );
3656+ assert!(
3657+ kv_writes(&mut cmd).is_empty(),
3658+ "contacts are not written over an unread model"
3659+ );
3660+
3661+ // And the regression itself. `ingest_poll` persists the outbox on every page,
3662+ // so one idle poll used to be enough to write `[]` over the queue on disk.
3663+ m.token = Some("t".into());
3664+ let resp = crux_http::testing::ResponseBuilder::ok()
3665+ .body(br#"{"events":[],"cursor":42}"#.to_vec())
3666+ .build();
3667+ let mut cmd = app.update(Event::PollResult(m.poll_gen, Ok(resp)), &mut m);
3668+ assert!(
3669+ kv_writes(&mut cmd).is_empty(),
3670+ "an empty poll writes nothing — not the outbox, not the cursor"
3671+ );
3672+
3673+ // The recovery is a relaunch, which re-runs the loads and clears the latch.
3674+ let sk = Identity::from_seed(&[3u8; 32]).secret_key.to_vec();
3675+ let _ = app.update(Event::IdentityLoaded(sk), &mut m);
3676+ assert!(
3677+ !m.kv_load_failed,
3678+ "a fresh round of loads gets to answer for itself"
3679+ );
3680+ }
3681+
33403682 /// `Authenticate` is reachable from startup, a 401 on poll, a 401 on send and
33413683 /// a reconnect timer. Without a guard a network blip fires all of them, and
33423684 /// every token that comes back starts its own poll loop.