@@ -73,6 +73,24 @@ const MAX_POLL_TOTAL_BYTES: usize = 64 * 1024 * 1024;
7373/// relay, so leave the margin.
7474const POLL_WATCHDOG_MS: u64 = 90_000;
7575
76+/// Messages per `/messages` request.
77+///
78+/// The relay rejects a longer batch outright (`413 batch_too_large`,
79+/// PROTOCOL.md §7), so this is a hard ceiling, not a tuning knob.
80+const MAX_SEND_BATCH: usize = 100;
81+
82+/// ...and a byte ceiling on the same batch, counted in hex characters — i.e. in
83+/// the units the request body is actually built from.
84+///
85+/// The count limit alone is not enough: a hundred profile broadcasts each
86+/// carrying a 64 KiB photo would build a request far past the 16 MiB body cap a
87+/// relay has by default (`--http-max-body-bytes`, PROTOCOL.md §7), and the whole
88+/// batch would come back `413` — which we read as a transient failure and retry,
89+/// forever. Bounding the body here keeps the batch inside what any conforming
90+/// relay accepts. A single item over the budget still goes out alone: it is the
91+/// relay's job to reject it, and its rejection is per-message and permanent.
92+const MAX_SEND_BATCH_BYTES: usize = 4 * 1024 * 1024;
93+
7694// kv keys
7795const K_SETTINGS: &str = "settings";
7896const K_PROFILE: &str = "profile";
@@ -120,6 +138,10 @@ pub enum Event {
120138 SendText,
121139 AddContact { input: String, nickname: String },
122140 SetBlocked { peer: String, blocked: bool },
141+ /// Forget a peer entirely: the contact, the conversation, and anything still
142+ /// queued for them. Blocking only silences a peer; the entry stays in the
143+ /// list forever, and there was no way to be rid of it.
144+ DeleteContact { peer: String },
123145 SaveProfile {
124146 display_name: String,
125147 bio: String,
@@ -220,6 +242,18 @@ pub enum Page {
220242/// `cursor`, `outbox`). `Connect` waits for all of them.
221243const STARTUP_LOADS: u8 = 5;
222244
245+/// The batch a live `/messages` request is carrying: the first `count` items of
246+/// the outbox, all bound for `recipient`.
247+///
248+/// `SendResult` needs both. The count is what it pops on success and what it
249+/// charges a retry against on a transient failure; the recipient is what tells
250+/// `SetBlocked` and `DeleteContact` whether the items they are about to remove
251+/// are the very ones this send is going to pop.
252+struct InFlight {
253+ recipient: String,
254+ count: usize,
255+}
256+
223257pub struct Model {
224258 /// Zeroized on drop: this is the Ed25519 seed, the one secret whose leak
225259 /// costs the user their identity.
@@ -261,7 +295,16 @@ pub struct Model {
261295 /// flight from before the app was backgrounded must not install a token or
262296 /// overwrite `conn` on top of the attempt that replaced it.
263297 auth_gen: u64,
298+ /// A `/messages` request is in flight. One at a time: `SendResult` pops the
299+ /// items it sent off the head of the outbox, so two overlapping sends would
300+ /// pop each other's messages.
264301 flushing: bool,
302+ /// What the in-flight send will pop off the head of the outbox when it
303+ /// succeeds — see `InFlight`. `None` while nothing is in flight, and also
304+ /// while a send is in flight whose items have since been removed from under
305+ /// it (`SetBlocked`, `DeleteContact`): there is then nothing left to pop, and
306+ /// popping by count anyway would eat whichever messages slid into their place.
307+ in_flight: Option<InFlight>,
265308 /// Startup kv loads still outstanding. `Connect` only fires at 0, so a poll
266309 /// can't land and be persisted against a half-loaded model — which would
267310 /// then be clobbered by the loads still in flight.
@@ -291,6 +334,7 @@ impl Default for Model {
291334 authenticating: false,
292335 auth_gen: 0,
293336 flushing: false,
337+ in_flight: None,
294338 loads_pending: 0,
295339 page: Page::default(),
296340 active_peer: None,
@@ -343,6 +387,26 @@ impl Model {
343387 self.authenticating = false;
344388 self.auth_gen = self.auth_gen.wrapping_add(1);
345389 }
390+
391+ /// Every item bound for `peer` is about to be removed from the outbox. If the
392+ /// send in flight is the one carrying them, forget what it was going to pop:
393+ /// those items will not be there when it lands, and popping by count anyway
394+ /// would take whatever slid into their place — another peer's mail.
395+ ///
396+ /// The `flushing` guard stays up. Nothing can cancel the request; it is still
397+ /// out there, and a second send started underneath it would have its own items
398+ /// popped by the first one's result.
399+ fn detach_in_flight(&mut self, peer: &str) {
400+ if self.in_flight.as_ref().is_some_and(|f| f.recipient == peer) {
401+ self.in_flight = None;
402+ }
403+ }
404+
405+ /// How many items at the head of the outbox are already on the wire, and so
406+ /// must not be rewritten in place (their ciphertext is built and gone).
407+ fn in_flight_count(&self) -> usize {
408+ self.in_flight.as_ref().map_or(0, |f| f.count)
409+ }
346410}
347411
348412// ---------------------------------------------------------------------------
@@ -693,11 +757,34 @@ impl App for Skrepka {
693757 // signal the block is supposed to cut off. Drop it all: once
694758 // blocked, nothing more goes to this peer.
695759 model.outbox.retain(|item| item.recipient != peer);
760+ model.detach_in_flight(&peer);
696761 return Command::all([self.persist_contacts(model), self.persist_outbox(model)])
697762 .and(render());
698763 }
699764 self.persist_contacts(model).and(render())
700765 }
766+ Event::DeleteContact { peer } => {
767+ model.contacts.remove(&peer);
768+ model.messages.remove(&peer);
769+ model.outbox.retain(|item| item.recipient != peer);
770+ // The send in flight may be carrying the very items we just
771+ // dropped; if so its result must pop nothing.
772+ model.detach_in_flight(&peer);
773+ if model.active_peer.as_deref() == Some(peer.as_str()) {
774+ model.active_peer = None;
775+ model.page = Page::Conversations;
776+ }
777+ model.error = None;
778+ Command::all([
779+ self.persist_contacts(model),
780+ // The conversation is gone from the model, but `persist_messages`
781+ // would leave a `messages:<peer>` blob behind holding every
782+ // message the user just asked us to forget. Delete the key.
783+ KeyValue::delete(k_messages(&peer)).then_send(Event::Saved),
784+ self.persist_outbox(model),
785+ ])
786+ .and(render())
787+ }
701788 Event::SaveProfile {
702789 display_name,
703790 bio,
@@ -720,11 +807,12 @@ impl App for Skrepka {
720807 // profile broadcast would keep feeding them our display name,
721808 // avatar, and a liveness signal.
722809 //
723− // Each item carries its own copy of the serialized payload, so a
724− // profile with a photo is duplicated (base64 and all) once per
725− // contact, in memory and in the `outbox` kv blob. Acceptable at
726− // today's contact counts; the fix is to intern the payload and
727− // have `OutboxItem` reference it by id.
810+ // One `Arc<String>` for the whole fan-out: the payload carries a
811+ // base64 photo, and a copy per contact would be megabytes in memory
812+ // for a list of any size. (The `outbox` kv blob still serializes the
813+ // string once per item — serde has no idea the `Arc`s are shared —
814+ // but the supersede-in-place below caps that at one pending profile
815+ // per contact, which is the bound that actually matters.)
728816 let recipients: Vec<String> = model
729817 .contacts
730818 .values()
@@ -732,11 +820,28 @@ impl App for Skrepka {
732820 .map(|c| c.pubkey.clone())
733821 .collect();
734822 let payload_json = Arc::new(protocol::serialize_payload(&payload, ts));
823+ // Items already on the wire cannot be rewritten: their ciphertext is
824+ // built and gone, and `SendResult` will pop them on success — so a
825+ // payload swapped into one now would be dropped having never been
826+ // sent. Supersede only what is still behind them.
827+ let live = model.in_flight_count();
735828 for peer in recipients {
736− model.outbox.push_back(OutboxItem {
737− recipient: peer,
738− envelope_json: Arc::clone(&payload_json),
739− });
829+ // A profile is state, not an event: only the newest one means
830+ // anything. Overwrite the pending one for this peer instead of
831+ // queueing a second — five quick edits to a profile with a photo
832+ // would otherwise be five payloads per contact, each of which we
833+ // then encrypt and send.
834+ match model
835+ .outbox
836+ .iter_mut()
837+ .skip(live)
838+ .find(|i| i.is_profile() && i.recipient == peer)
839+ {
840+ Some(pending) => pending.envelope_json = Arc::clone(&payload_json),
841+ None => model
842+ .outbox
843+ .push_back(OutboxItem::profile(peer, Arc::clone(&payload_json))),
844+ }
740845 }
741846 Command::all([
742847 KeyValue::set(K_PROFILE, json_bytes(&model.profile)).then_send(Event::Saved),
@@ -957,22 +1062,30 @@ impl App for Skrepka {
9571062 // The in-flight send is over; `flush_next` refuses to run while
9581063 // this is set, so it must be cleared before *any* branch below.
9591064 model.flushing = false;
1065+ let batch = model.in_flight.take();
9601066 let status = u16::from(resp.status());
9611067 if status == 401 {
9621068 model.token = None;
1069+ // The batch stays exactly as it is — re-auth, then re-send it.
9631070 return Command::event(Event::Authenticate);
9641071 }
9651072 if (200..300).contains(&status) || status == 400 {
966− // success, or a permanent rejection (self_send/invalid) — drop it.
967− model.outbox.pop_front();
1073+ // The batch was accepted, or rejected outright (`self_send` /
1074+ // `invalid_message` — both permanent, and the relay takes a
1075+ // batch all-or-nothing). Either way it is finished with.
1076+ for _ in 0..batch.map_or(0, |b| b.count) {
1077+ model.outbox.pop_front();
1078+ }
9681079 return self.persist_outbox(model).and(self.flush_next(model));
9691080 }
970− // transient error — stop; will resume on next connect/poll.
971− render()
1081+ // A transient failure: a 5xx, a rate limit, a relay restarting.
1082+ self.retry_batch(model, batch)
9721083 }
9731084 Event::SendResult(Err(_)) => {
1085+ // A transport error is the same transient failure as a 5xx.
9741086 model.flushing = false;
975− render()
1087+ let batch = model.in_flight.take();
1088+ self.retry_batch(model, batch)
9761089 }
9771090 }
9781091 }
@@ -1145,13 +1258,19 @@ impl Skrepka {
11451258
11461259 let convo = model.messages.entry(peer.clone()).or_default();
11471260 let first_to_peer = !convo.iter().any(|m| m.outgoing);
1148− convo.push(StoredMessage {
1149− id: id.clone(),
1150− body: body.clone(),
1151− ts,
1152− outgoing: true,
1153− delivered: false,
1154− });
1261+ // Insert in order rather than appending: an incoming message may hold a
1262+ // `ts` up to `MAX_FUTURE_SKEW_MS` ahead of ours, so appending would leave
1263+ // the conversation unsorted — and `trim_history` ages out the *front*.
1264+ insert_sorted(
1265+ convo,
1266+ StoredMessage {
1267+ id: id.clone(),
1268+ body: body.clone(),
1269+ ts,
1270+ outgoing: true,
1271+ delivered: false,
1272+ },
1273+ );
11551274 trim_history(convo);
11561275
11571276 // Auto-create a bare contact if messaging a brand-new key.
@@ -1161,21 +1280,23 @@ impl Skrepka {
11611280 .or_insert_with(|| Contact::new(peer.clone(), String::new(), ts));
11621281
11631282 let text = Payload::Text { id, body };
1164− model.outbox.push_back(OutboxItem {
1165− recipient: peer.clone(),
1166− envelope_json: Arc::new(protocol::serialize_payload(&text, ts)),
1167− });
1168− // On first contact, also share our profile.
1283+ model.outbox.push_back(OutboxItem::new(
1284+ peer.clone(),
1285+ Arc::new(protocol::serialize_payload(&text, ts)),
1286+ ));
1287+ // On first contact, also share our profile. Marked as a profile like any
1288+ // other, so a `SaveProfile` before this ever goes out supersedes it rather
1289+ // than queueing a second one behind it.
11691290 if first_to_peer {
11701291 let profile = Payload::Profile {
11711292 display_name: model.profile.display_name.clone(),
11721293 bio: model.profile.bio.clone(),
11731294 photo: model.profile.photo.clone(),
11741295 };
1175− model.outbox.push_back(OutboxItem {
1176− recipient: peer.clone(),
1177− envelope_json: Arc::new(protocol::serialize_payload(&profile, ts)),
1178− });
1296+ model.outbox.push_back(OutboxItem::profile(
1297+ peer.clone(),
1298+ Arc::new(protocol::serialize_payload(&profile, ts)),
1299+ ));
11791300 }
11801301
11811302 Command::all([
@@ -1187,8 +1308,23 @@ impl Skrepka {
11871308 .and(render())
11881309 }
11891310
1190− /// Encrypt and send the head of the outbox, discarding any unsendable items
1191− /// ahead of it.
1311+ /// Encrypt and send a batch from the head of the outbox, discarding any
1312+ /// unsendable items in the way.
1313+ ///
1314+ /// The batch is the longest run of items at the head that share a recipient,
1315+ /// bounded by `MAX_SEND_BATCH` and `MAX_SEND_BATCH_BYTES`. Sending them one
1316+ /// at a time meant an HTTP round-trip per message — and per *ack*, which
1317+ /// `ingest_poll` queues one of per sender per page, so catching up after a
1318+ /// long offline stretch was one request per message received. The relay takes
1319+ /// up to a hundred per request (PROTOCOL.md §7); take it up on that.
1320+ ///
1321+ /// Stopping at the first recipient change keeps the outbox's FIFO ordering
1322+ /// intact — a batch is delivered as a unit, so hoisting a later item for the
1323+ /// same peer over an earlier one for a different peer could only reorder
1324+ /// *across* conversations, but it would also let one busy peer starve the
1325+ /// rest of the queue. Consecutive-only is the cheap, obviously-fair rule, and
1326+ /// it already collapses the two cases that matter: a burst of texts to the
1327+ /// open chat, and a page's worth of acks back to one sender.
11921328 ///
11931329 /// The discards happen in one pass, followed by a single `persist_outbox`.
11941330 /// The previous shape recursed once per bad item and emitted a full outbox
@@ -1203,66 +1339,145 @@ impl Skrepka {
12031339 return Command::done();
12041340 };
12051341
1206− let mut dropped = false;
1207− let sendable = loop {
1208− let Some(item) = model.outbox.front() else {
1209− break None;
1342+ let now = now_ms();
1343+ let mut dirty = false;
1344+ let mut recipient: Option<String> = None;
1345+ let mut messages: Vec<Envelope> = Vec::new();
1346+ let mut bytes = 0usize;
1347+
1348+ while messages.len() < MAX_SEND_BATCH {
1349+ let Some(item) = model.outbox.get(messages.len()) else {
1350+ break;
12101351 };
1211− match encrypt_for(&id, item) {
1212− Some(blob) => break Some((item.recipient.clone(), blob)),
1213− None => {
1214− // An unusable recipient key can never be encrypted to, so the
1215− // item has to go or it blocks the head of the outbox forever —
1216− // but say so, rather than letting the message vanish.
1217− model.error = Some(format!(
1218− "cannot send to {}: invalid key",
1219− trunc_ob(&item.recipient)
1220− ));
1221− model.outbox.pop_front();
1222− dropped = true;
1223− }
1352+ if recipient.as_ref().is_some_and(|r| r != &item.recipient) {
1353+ break;
12241354 }
1225− };
1226−
1227− let persist = if dropped {
1228− self.persist_outbox(model)
1229− } else {
1230− Command::done()
1231− };
1355+ let Some(blob) = encrypt_for(&id, item) else {
1356+ // An unusable recipient key can never be encrypted to, so the item
1357+ // has to go or it blocks the head of the outbox forever — but say
1358+ // so, rather than letting the message vanish. (Only reachable at
1359+ // the head: an item deeper in the batch shares the recipient of one
1360+ // we have already encrypted to, so its key is good by construction.)
1361+ model.error = Some(format!(
1362+ "cannot send to {}: invalid key",
1363+ trunc_ob(&item.recipient)
1364+ ));
1365+ model.outbox.remove(messages.len());
1366+ dirty = true;
1367+ continue;
1368+ };
1369+ let hex_blob = hex::encode(blob);
1370+ // Always take the first item, whatever it weighs — a lone oversized
1371+ // payload must still reach the relay, which rejects it permanently
1372+ // (`400`) and so gets it out of the queue for good. Past the first,
1373+ // stop before the body outgrows what a relay will accept.
1374+ if !messages.is_empty() && bytes + hex_blob.len() > MAX_SEND_BATCH_BYTES {
1375+ break;
1376+ }
1377+ bytes += hex_blob.len();
1378+ recipient = Some(item.recipient.clone());
1379+ messages.push(Envelope {
1380+ to: item.recipient.clone(),
1381+ encrypted_blob: hex_blob,
1382+ });
1383+ }
12321384
1233− let Some((recipient, blob)) = sendable else {
1385+ let Some(recipient) = recipient else {
12341386 // Nothing left to send. Render only if we dropped something, so the
12351387 // user sees why their message went away.
1236− return if dropped {
1237− persist.and(render())
1388+ return if dirty {
1389+ self.persist_outbox(model).and(render())
12381390 } else {
12391391 Command::done()
12401392 };
12411393 };
12421394
1243− model.flushing = true;
1244− let batch = SendBatch {
1245− messages: vec![Envelope {
1246− to: recipient,
1247− encrypted_blob: hex::encode(blob),
1248− }],
1395+ // Stamp the items now going out for the first time. This is what the TTL
1396+ // in `OutboxItem::is_expired` measures from, so it has to be durable —
1397+ // otherwise a relaunch resets the clock and a permanently-stuck item is
1398+ // immortal again.
1399+ for item in model.outbox.iter_mut().take(messages.len()) {
1400+ if item.first_attempt == 0 {
1401+ item.first_attempt = now;
1402+ dirty = true;
1403+ }
1404+ }
1405+ let persist = if dirty {
1406+ self.persist_outbox(model)
1407+ } else {
1408+ Command::done()
12491409 };
1410+
1411+ model.flushing = true;
1412+ model.in_flight = Some(InFlight {
1413+ recipient,
1414+ count: messages.len(),
1415+ });
12501416 let url = format!("{}/messages", model.settings.server_url);
12511417 match Http::post(url)
12521418 .header("authorization", format!("Bearer {token}"))
1253− .body_json(&batch)
1419+ .body_json(&SendBatch { messages })
12541420 {
12551421 Ok(req) => persist.and(req.build().then_send(Event::SendResult)),
12561422 Err(_) => {
12571423 // Not a dead loop: the next StartFlush (from a poll page or a
12581424 // fresh send) retries the head of the outbox.
12591425 model.flushing = false;
1426+ model.in_flight = None;
12601427 model.error = Some("could not build send request".into());
12611428 persist.and(render())
12621429 }
12631430 }
12641431 }
12651432
1433+ /// A send failed transiently (a 5xx, a rate limit, a dropped connection). The
1434+ /// batch stays queued and goes out again on the next `StartFlush` — but not
1435+ /// forever.
1436+ ///
1437+ /// The outbox is a strict FIFO, so an item that can never be delivered blocks
1438+ /// every message queued behind it, permanently: a single unreachable recipient
1439+ /// relay used to be enough to silence the client for good. Charge every item in
1440+ /// the failed batch a retry and drop the ones that have exhausted their budget
1441+ /// (`OutboxItem::is_expired`) — with an error, rather than letting them vanish.
1442+ fn retry_batch(&self, model: &mut Model, batch: Option<InFlight>) -> Command<Effect, Event> {
1443+ // `None` means the items were removed out from under the send (the peer was
1444+ // blocked, or deleted). There is nothing left to charge or to drop.
1445+ let Some(batch) = batch else {
1446+ return render();
1447+ };
1448+
1449+ let now = now_ms();
1450+ let mut expired = 0usize;
1451+ let mut kept: VecDeque<OutboxItem> = VecDeque::with_capacity(model.outbox.len());
1452+ for (i, mut item) in std::mem::take(&mut model.outbox).into_iter().enumerate() {
1453+ if i < batch.count {
1454+ item.retries = item.retries.saturating_add(1);
1455+ if item.is_expired(now) {
1456+ expired += 1;
1457+ continue;
1458+ }
1459+ }
1460+ kept.push_back(item);
1461+ }
1462+ model.outbox = kept;
1463+
1464+ if expired == 0 {
1465+ // Nothing dropped, but the retry counters still have to be durable: a
1466+ // relaunch that handed every stuck item a fresh budget would wedge the
1467+ // outbox across restarts exactly as it did before there was a budget.
1468+ return self.persist_outbox(model).and(render());
1469+ }
1470+ model.error = Some(format!(
1471+ "gave up sending {expired} message(s) to {}",
1472+ trunc_ob(&batch.recipient)
1473+ ));
1474+ // The head of the outbox has moved, so whatever was stuck behind the items
1475+ // we just dropped can finally go out.
1476+ self.persist_outbox(model)
1477+ .and(self.flush_next(model))
1478+ .and(render())
1479+ }
1480+
12661481 /// Decrypt and fold a poll page into the model; queue delivery acks.
12671482 fn ingest_poll(&self, model: &mut Model, page: PollResp) -> Command<Effect, Event> {
12681483 if page.cursor > model.cursor {
@@ -1349,15 +1564,21 @@ impl Skrepka {
13491564 });
13501565 // dedup by id — O(1) via HashSet
13511566 if !id_set.contains(&msg_id) {
1352− convo.push(StoredMessage {
1353− id: msg_id.clone(),
1354− body,
1355− ts,
1356− outgoing: false,
1357− delivered: false,
1358− });
1567+ // The conversation is *kept* sorted rather than re-sorted:
1568+ // a `sort_by_key` per arrival is O(n log n) on a list of up
1569+ // to `MAX_MESSAGES_PER_PEER`, every time, and a poll page
1570+ // can carry hundreds of messages.
1571+ insert_sorted(
1572+ convo,
1573+ StoredMessage {
1574+ id: msg_id.clone(),
1575+ body,
1576+ ts,
1577+ outgoing: false,
1578+ delivered: false,
1579+ },
1580+ );
13591581 id_set.insert(msg_id.clone());
1360− convo.sort_by_key(|m| m.ts);
13611582 trim_history(convo);
13621583 touched_convos.insert(sender.clone());
13631584 }
@@ -1373,9 +1594,15 @@ impl Skrepka {
13731594 if stale {
13741595 continue;
13751596 }
1376− // Update last_ack_ts to the clamped ts.
1597+ // The watermark is clamped to *now*, not merely into the skew
1598+ // window: `ts` is the peer's to choose, and `last_ack_ts` is
1599+ // what every later ack is measured against. A single ack at
1600+ // `ts = i64::MAX` would otherwise park the watermark a minute
1601+ // into the future, and every honest ack that followed — the
1602+ // ones that actually mark our messages delivered — would read
1603+ // as stale and be dropped.
13771604 if let Some(c) = model.contacts.get_mut(&sender) {
1378− c.last_ack_ts = ts;
1605+ c.last_ack_ts = ts.min(now);
13791606 contacts_dirty = true;
13801607 }
13811608 let ack_set: HashSet<String> = ack_ids.iter().cloned().collect();
@@ -1404,14 +1631,16 @@ impl Skrepka {
14041631 let Some(c) = model.contacts.get_mut(&sender) else {
14051632 continue;
14061633 };
1407− // Ignore stale profile replays (PROTOCOL.md §4). The clamped
1408− // `ts` is what gets recorded, so a peer cannot park
1409− // `last_profile_ts` at `i64::MAX` and freeze its own profile.
1634+ // Ignore stale profile replays (PROTOCOL.md §4). The watermark
1635+ // is clamped to *now* for the same reason `last_ack_ts` is: a
1636+ // peer that parked it in the future would freeze its own
1637+ // profile, since every honest update it sent afterwards would
1638+ // read as a stale replay of itself.
14101639 if ts >= c.last_profile_ts {
14111640 c.display_name = display_name;
14121641 c.bio = bio;
14131642 c.photo = photo;
1414− c.last_profile_ts = ts;
1643+ c.last_profile_ts = ts.min(now);
14151644 contacts_dirty = true;
14161645 }
14171646 }
@@ -1433,10 +1662,12 @@ impl Skrepka {
14331662 let payload = Payload::DeliveryAck {
14341663 ack_ids: batch.to_vec(),
14351664 };
1436− model.outbox.push_back(OutboxItem {
1437− recipient: sender.clone(),
1438− envelope_json: Arc::new(protocol::serialize_payload(&payload, ack_ts)),
1439− });
1665+ // Consecutive and same-recipient, so `flush_next` puts every one of
1666+ // a sender's ack batches into a single `/messages` request.
1667+ model.outbox.push_back(OutboxItem::new(
1668+ sender.clone(),
1669+ Arc::new(protocol::serialize_payload(&payload, ack_ts)),
1670+ ));
14401671 }
14411672 }
14421673
@@ -1468,6 +1699,20 @@ fn backoff_ms(retries: u32) -> u64 {
14681699 (3000u64.saturating_mul(1u64 << retries.min(4))).min(30_000)
14691700}
14701701
1702+/// Insert a message into a conversation held in `ts` order.
1703+///
1704+/// `partition_point` lands *after* every message sharing the new one's `ts`, so
1705+/// ties keep arrival order — exactly what the stable `sort_by_key` this replaces
1706+/// produced, at O(log n) + a shift instead of O(n log n) on every message.
1707+///
1708+/// This is what makes "sorted by `ts`" an invariant rather than something
1709+/// `ingest_poll` restored after the fact, which is what `trim_history` has always
1710+/// assumed (it ages out the *front*) and what `view()` renders in order.
1711+fn insert_sorted(convo: &mut Vec<StoredMessage>, msg: StoredMessage) {
1712+ let pos = convo.partition_point(|m| m.ts <= msg.ts);
1713+ convo.insert(pos, msg);
1714+}
1715+
14711716/// Age out local history: keep only the most recent `MAX_MESSAGES_PER_PEER`
14721717/// messages of a conversation (PROTOCOL.md §9). Callers keep `convo` sorted by
14731718/// `ts`, so the oldest messages are at the front. The subsequent kv write
@@ -1500,6 +1745,7 @@ fn parse_kv<T: for<'de> Deserialize<'de>>(res: KvData) -> Option<T> {
15001745mod tests {
15011746 use super::*;
15021747 use crate::crypto::Identity;
1748+ use crate::model::{MAX_OUTBOX_RETRIES, OUTBOX_TTL_MS};
15031749
15041750 fn with_identity() -> Model {
15051751 let id = Identity::from_seed(&[3u8; 32]);
@@ -1559,6 +1805,627 @@ mod tests {
15591805 }
15601806 }
15611807
1808+ fn queued(recipient: &str, body: &str) -> OutboxItem {
1809+ OutboxItem::new(
1810+ recipient.to_string(),
1811+ Arc::new(protocol::serialize_payload(&text(body, body), 1)),
1812+ )
1813+ }
1814+
1815+ /// The `/messages` batches a command actually puts on the wire. Reading them
1816+ /// off the effect is the only way to see what was *sent*, as opposed to what
1817+ /// the model thinks it sent.
1818+ fn sent_batches(cmd: &mut Command<Effect, Event>) -> Vec<Vec<Envelope>> {
1819+ #[derive(Deserialize)]
1820+ struct Sent {
1821+ messages: Vec<Envelope>,
1822+ }
1823+ cmd.effects()
1824+ .filter_map(|e| match e {
1825+ Effect::Http(req) if req.operation.url.ends_with("/messages") => {
1826+ serde_json::from_slice::<Sent>(&req.operation.body)
1827+ .ok()
1828+ .map(|b| b.messages)
1829+ }
1830+ _ => None,
1831+ })
1832+ .collect()
1833+ }
1834+
1835+ fn ok_response(status: u16) -> crux_http::Response<Vec<u8>> {
1836+ let status = crux_http::http::StatusCode::try_from(status).unwrap();
1837+ crux_http::testing::ResponseBuilder::with_status(status)
1838+ .body(Vec::new())
1839+ .build()
1840+ }
1841+
1842+ /// One full send round-trip: flush the outbox, then answer with `status`.
1843+ fn send_and_answer(app: &Skrepka, m: &mut Model, status: u16) {
1844+ let _ = app.update(Event::StartFlush, m);
1845+ let _ = app.update(Event::SendResult(Ok(ok_response(status))), m);
1846+ }
1847+
1848+ // -----------------------------------------------------------------------
1849+ // outbox: retry budget and TTL
1850+ // -----------------------------------------------------------------------
1851+
1852+ /// The outbox is a strict FIFO, so an item that can never be delivered used to
1853+ /// block every message queued behind it — forever.
1854+ #[test]
1855+ fn an_undeliverable_item_is_dropped_once_it_runs_out_of_retries() {
1856+ let app = Skrepka;
1857+ let mut m = with_identity();
1858+ m.token = Some("t".into());
1859+ let stuck = peer_hex(9);
1860+ let behind = peer_hex(11);
1861+ m.outbox.push_back(queued(&stuck, "wedged"));
1862+ m.outbox.push_back(queued(&behind, "stranded"));
1863+
1864+ // The relay keeps 500-ing on the head of the queue.
1865+ for i in 1..MAX_OUTBOX_RETRIES {
1866+ send_and_answer(&app, &mut m, 503);
1867+ assert_eq!(m.outbox.len(), 2, "still queued after {i} failure(s)");
1868+ assert_eq!(m.outbox[0].retries, i, "and its budget is being spent");
1869+ }
1870+
1871+ // The last one exhausts the budget: the head goes, and what was stuck
1872+ // behind it goes out.
1873+ let _ = app.update(Event::StartFlush, &mut m);
1874+ let mut cmd = app.update(Event::SendResult(Ok(ok_response(503))), &mut m);
1875+
1876+ assert_eq!(m.outbox.len(), 1, "the undeliverable item is dropped");
1877+ assert_eq!(m.outbox[0].recipient, behind, "and the queue moves on");
1878+ assert_eq!(
1879+ sent_batches(&mut cmd).len(),
1880+ 1,
1881+ "the message behind it is sent immediately, not left for the next flush"
1882+ );
1883+ let err = m.error.expect("the user is told the message was given up on");
1884+ assert!(err.contains("gave up sending 1"), "got: {err}");
1885+ }
1886+
1887+ /// The counter alone is not enough: retries only accrue when a send is
1888+ /// attempted, so a client that goes online once a day would take a fortnight to
1889+ /// spend a ten-retry budget. The TTL bounds it in wall-clock time instead.
1890+ #[test]
1891+ fn an_item_older_than_the_ttl_is_dropped_on_its_next_failure() {
1892+ let app = Skrepka;
1893+ let mut m = with_identity();
1894+ m.token = Some("t".into());
1895+ let peer = peer_hex(9);
1896+
1897+ let mut old = queued(&peer, "ancient");
1898+ old.first_attempt = now_ms() - OUTBOX_TTL_MS - 1;
1899+ old.retries = 1; // nowhere near the retry cap
1900+ m.outbox.push_back(old);
1901+
1902+ send_and_answer(&app, &mut m, 503);
1903+
1904+ assert!(m.outbox.is_empty(), "the TTL alone is enough to drop it");
1905+ assert!(m.error.is_some_and(|e| e.contains("gave up sending")));
1906+ }
1907+
1908+ /// An item queued during a week offline must not be dropped the moment the
1909+ /// network returns: the TTL runs from the first *attempt*, not from when the
1910+ /// message was written. An outbox blob from a build with no `first_attempt`
1911+ /// field deserializes to 0, which must mean "never attempted", not "epoch".
1912+ #[test]
1913+ fn an_item_that_was_never_attempted_has_no_ttl_to_outlive() {
1914+ let item = OutboxItem::new(peer_hex(9), Arc::new("{}".into()));
1915+ assert_eq!(item.first_attempt, 0);
1916+ assert!(
1917+ !item.is_expired(now_ms()),
1918+ "an unsent item must not age out against the epoch"
1919+ );
1920+ }
1921+
1922+ /// The TTL clock has to survive a relaunch, or a permanently-stuck item is
1923+ /// handed a fresh 24 hours on every launch and the outbox wedges forever again.
1924+ #[test]
1925+ fn the_first_attempt_stamp_is_persisted_when_the_item_goes_out() {
1926+ let app = Skrepka;
1927+ let mut m = with_identity();
1928+ m.token = Some("t".into());
1929+ m.outbox.push_back(queued(&peer_hex(9), "hi"));
1930+ let before = now_ms();
1931+
1932+ let mut cmd = app.update(Event::StartFlush, &mut m);
1933+
1934+ assert!(m.outbox[0].first_attempt >= before, "the item is stamped");
1935+ // ...and the stamp is written to the `outbox` key, not just to memory.
1936+ let wrote_outbox = cmd.effects().any(|e| {
1937+ matches!(e, Effect::KeyValue(req)
1938+ if matches!(&req.operation, crux_kv::KeyValueOperation::Set { key, .. } if key == K_OUTBOX))
1939+ });
1940+ assert!(wrote_outbox, "the stamped outbox is persisted");
1941+ }
1942+
1943+ /// A transient failure below the budget changes nothing but the counter — the
1944+ /// message stays queued and goes out on the next flush.
1945+ #[test]
1946+ fn a_transient_failure_within_budget_keeps_the_message() {
1947+ let app = Skrepka;
1948+ let mut m = with_identity();
1949+ m.token = Some("t".into());
1950+ m.outbox.push_back(queued(&peer_hex(9), "hi"));
1951+
1952+ send_and_answer(&app, &mut m, 503);
1953+ assert_eq!(m.outbox.len(), 1);
1954+ assert_eq!(m.outbox[0].retries, 1);
1955+ assert!(m.error.is_none(), "a single 5xx is not worth shouting about");
1956+ assert!(!m.flushing, "and the guard is released so the next flush can run");
1957+
1958+ // The relay recovers; the message goes out and the item is gone.
1959+ send_and_answer(&app, &mut m, 200);
1960+ assert!(m.outbox.is_empty());
1961+ }
1962+
1963+ /// A 401 is not a delivery failure — it means our token expired. The batch must
1964+ /// not be charged a retry for it, or a long-lived client burns its whole budget
1965+ /// on routine re-auths.
1966+ #[test]
1967+ fn a_401_does_not_spend_the_retry_budget() {
1968+ let app = Skrepka;
1969+ let mut m = with_identity();
1970+ m.token = Some("t".into());
1971+ m.outbox.push_back(queued(&peer_hex(9), "hi"));
1972+
1973+ send_and_answer(&app, &mut m, 401);
1974+
1975+ assert_eq!(m.outbox.len(), 1, "the message is still queued");
1976+ assert_eq!(m.outbox[0].retries, 0, "and unpunished");
1977+ assert!(m.token.is_none(), "but the token is thrown away");
1978+ }
1979+
1980+ // -----------------------------------------------------------------------
1981+ // outbox: batching
1982+ // -----------------------------------------------------------------------
1983+
1984+ /// Sending one message per request meant a round-trip per *ack* too, so
1985+ /// catching up on a hundred messages cost a hundred requests.
1986+ #[test]
1987+ fn consecutive_items_for_one_recipient_go_out_in_a_single_request() {
1988+ let app = Skrepka;
1989+ let mut m = with_identity();
1990+ m.token = Some("t".into());
1991+ let peer = peer_hex(9);
1992+ for i in 0..5 {
1993+ m.outbox.push_back(queued(&peer, &format!("m{i}")));
1994+ }
1995+
1996+ let mut cmd = app.update(Event::StartFlush, &mut m);
1997+
1998+ let batches = sent_batches(&mut cmd);
1999+ assert_eq!(batches.len(), 1, "one request, not five");
2000+ assert_eq!(batches[0].len(), 5, "carrying every message");
2001+ assert!(batches[0].iter().all(|e| e.to == peer));
2002+
2003+ // ...and its success clears the whole batch, not just the head.
2004+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
2005+ assert!(m.outbox.is_empty());
2006+ }
2007+
2008+ /// The batch stops at the first recipient change: a batch is delivered as a
2009+ /// unit, and hoisting a later item for the same peer over an earlier one for a
2010+ /// different peer would let one busy conversation starve the rest of the queue.
2011+ #[test]
2012+ fn a_batch_stops_at_the_first_recipient_change() {
2013+ let app = Skrepka;
2014+ let mut m = with_identity();
2015+ m.token = Some("t".into());
2016+ let a = peer_hex(9);
2017+ let b = peer_hex(11);
2018+ m.outbox.push_back(queued(&a, "a1"));
2019+ m.outbox.push_back(queued(&a, "a2"));
2020+ m.outbox.push_back(queued(&b, "b1"));
2021+ m.outbox.push_back(queued(&a, "a3"));
2022+
2023+ let mut cmd = app.update(Event::StartFlush, &mut m);
2024+ let batches = sent_batches(&mut cmd);
2025+ assert_eq!(batches.len(), 1);
2026+ assert_eq!(batches[0].len(), 2, "only the run at the head");
2027+ assert!(batches[0].iter().all(|e| e.to == a));
2028+
2029+ let mut cmd = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
2030+ assert_eq!(m.outbox.len(), 2);
2031+ let batches = sent_batches(&mut cmd);
2032+ assert_eq!(batches[0].len(), 1, "then b's single message");
2033+ assert_eq!(batches[0][0].to, b);
2034+ }
2035+
2036+ /// The relay answers `413 batch_too_large` past a hundred (PROTOCOL.md §7) —
2037+ /// which we would read as a transient failure and retry forever.
2038+ #[test]
2039+ fn a_batch_is_capped_at_the_relays_limit() {
2040+ let app = Skrepka;
2041+ let mut m = with_identity();
2042+ m.token = Some("t".into());
2043+ let peer = peer_hex(9);
2044+ for i in 0..(MAX_SEND_BATCH + 20) {
2045+ m.outbox.push_back(queued(&peer, &format!("m{i}")));
2046+ }
2047+
2048+ let mut cmd = app.update(Event::StartFlush, &mut m);
2049+ let batches = sent_batches(&mut cmd);
2050+ assert_eq!(batches[0].len(), MAX_SEND_BATCH);
2051+
2052+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
2053+ assert_eq!(m.outbox.len(), 20, "the rest stays queued for the next batch");
2054+ }
2055+
2056+ /// A hundred profile broadcasts each carrying a 64 KiB photo would build a
2057+ /// request far past the body-size cap a relay has by default — and the `413`
2058+ /// that came back would look transient, so we would retry it forever.
2059+ #[test]
2060+ fn a_batch_is_capped_by_body_size_as_well_as_count() {
2061+ let app = Skrepka;
2062+ let mut m = with_identity();
2063+ m.token = Some("t".into());
2064+ let peer = peer_hex(9);
2065+ // A body at the cap, and incompressible — `crypto::encrypt` deflates the
2066+ // plaintext, so a run of one character would ride out to nearly nothing and
2067+ // prove no cap at all.
2068+ let mut seed = 1u64;
2069+ let big: String = std::iter::repeat_with(|| {
2070+ seed = seed.wrapping_mul(6_364_136_223_846_793_005).wrapping_add(1);
2071+ char::from(b'a' + (seed >> 33) as u8 % 26)
2072+ })
2073+ .take(protocol::MAX_BODY_LEN)
2074+ .collect();
2075+ for i in 0..MAX_SEND_BATCH {
2076+ m.outbox.push_back(OutboxItem::new(
2077+ peer.clone(),
2078+ Arc::new(protocol::serialize_payload(&text(&format!("m{i}"), &big), 1)),
2079+ ));
2080+ }
2081+
2082+ let mut cmd = app.update(Event::StartFlush, &mut m);
2083+ let batches = sent_batches(&mut cmd);
2084+ let body: usize = batches[0].iter().map(|e| e.encrypted_blob.len()).sum();
2085+
2086+ assert!(batches[0].len() < MAX_SEND_BATCH, "the count cap is not the binding one");
2087+ assert!(!batches[0].is_empty(), "but something still goes out");
2088+ assert!(body <= MAX_SEND_BATCH_BYTES, "the body stays inside the budget");
2089+ }
2090+
2091+ /// An unusable key at the head still has to be discarded, batching or not —
2092+ /// and the discard must not take the rest of the queue with it.
2093+ #[test]
2094+ fn an_unsendable_head_is_dropped_and_the_batch_behind_it_still_goes_out() {
2095+ let app = Skrepka;
2096+ let mut m = with_identity();
2097+ m.token = Some("t".into());
2098+ let good = peer_hex(9);
2099+ m.outbox.push_back(queued(&hex::encode([0xabu8; 32]), "doomed"));
2100+ m.outbox.push_back(queued(&good, "fine"));
2101+ m.outbox.push_back(queued(&good, "also fine"));
2102+
2103+ let mut cmd = app.update(Event::StartFlush, &mut m);
2104+
2105+ assert_eq!(m.outbox.len(), 2, "only the unsendable item is dropped");
2106+ let batches = sent_batches(&mut cmd);
2107+ assert_eq!(batches[0].len(), 2, "and the batch behind it goes out whole");
2108+ assert!(m.error.is_some_and(|e| e.contains("invalid key")));
2109+ }
2110+
2111+ // -----------------------------------------------------------------------
2112+ // outbox: profile supersede
2113+ // -----------------------------------------------------------------------
2114+
2115+ /// A profile is state, not an event: only the newest one means anything. Five
2116+ /// quick edits used to queue five payloads per contact — each with the photo.
2117+ #[test]
2118+ fn repeated_profile_edits_supersede_rather_than_pile_up() {
2119+ let app = Skrepka;
2120+ let mut m = with_identity();
2121+ let a = peer_hex(9);
2122+ let b = peer_hex(11);
2123+ for peer in [&a, &b] {
2124+ m.contacts
2125+ .insert(peer.clone(), Contact::new(peer.clone(), String::new(), 0));
2126+ }
2127+
2128+ for name in ["One", "Two", "Three"] {
2129+ let _ = app.update(
2130+ Event::SaveProfile {
2131+ display_name: name.into(),
2132+ bio: String::new(),
2133+ photo: None,
2134+ },
2135+ &mut m,
2136+ );
2137+ }
2138+
2139+ assert_eq!(m.outbox.len(), 2, "one pending profile per contact, not three");
2140+ for item in &m.outbox {
2141+ assert!(item.is_profile());
2142+ let parsed = protocol::parse_payload(item.envelope_json.as_bytes()).unwrap();
2143+ match parsed.payload {
2144+ Payload::Profile { display_name, .. } => {
2145+ assert_eq!(display_name, "Three", "and it is the latest one");
2146+ }
2147+ _ => panic!("expected a profile"),
2148+ }
2149+ }
2150+ }
2151+
2152+ /// Superseding must not reach into a payload that is already on the wire: its
2153+ /// ciphertext is built and gone, and `SendResult` pops it on success — so a
2154+ /// profile swapped in now would be dropped having never been sent.
2155+ #[test]
2156+ fn a_profile_already_in_flight_is_not_rewritten_under_the_send() {
2157+ let app = Skrepka;
2158+ let mut m = with_identity();
2159+ m.token = Some("t".into());
2160+ let peer = peer_hex(9);
2161+ m.contacts
2162+ .insert(peer.clone(), Contact::new(peer.clone(), String::new(), 0));
2163+
2164+ let _ = app.update(
2165+ Event::SaveProfile {
2166+ display_name: "Old".into(),
2167+ bio: String::new(),
2168+ photo: None,
2169+ },
2170+ &mut m,
2171+ );
2172+ // It goes out and is still in flight.
2173+ let _ = app.update(Event::StartFlush, &mut m);
2174+ assert!(m.flushing);
2175+
2176+ // A second edit lands mid-send. It must queue a *new* item rather than
2177+ // rewrite the one whose ciphertext is already gone.
2178+ let _ = app.update(
2179+ Event::SaveProfile {
2180+ display_name: "New".into(),
2181+ bio: String::new(),
2182+ photo: None,
2183+ },
2184+ &mut m,
2185+ );
2186+ assert_eq!(m.outbox.len(), 2, "the in-flight item is left alone");
2187+
2188+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
2189+ assert_eq!(m.outbox.len(), 1, "the sent one is popped");
2190+ let parsed = protocol::parse_payload(m.outbox[0].envelope_json.as_bytes()).unwrap();
2191+ match parsed.payload {
2192+ Payload::Profile { display_name, .. } => assert_eq!(
2193+ display_name, "New",
2194+ "and the newer profile survives to be sent"
2195+ ),
2196+ _ => panic!("expected a profile"),
2197+ }
2198+ }
2199+
2200+ // -----------------------------------------------------------------------
2201+ // in-flight bookkeeping
2202+ // -----------------------------------------------------------------------
2203+
2204+ /// `SendResult` pops by count off the head of the outbox. If the items it sent
2205+ /// were removed from under it — the user blocked the peer mid-send — popping by
2206+ /// count anyway takes whatever slid into their place: another peer's mail.
2207+ #[test]
2208+ fn blocking_the_peer_mid_send_does_not_pop_someone_elses_messages() {
2209+ let app = Skrepka;
2210+ let mut m = with_identity();
2211+ m.token = Some("t".into());
2212+ let enemy = peer_hex(9);
2213+ let friend = peer_hex(11);
2214+ m.contacts
2215+ .insert(enemy.clone(), Contact::new(enemy.clone(), String::new(), 0));
2216+ m.outbox.push_back(queued(&enemy, "e1"));
2217+ m.outbox.push_back(queued(&enemy, "e2"));
2218+ m.outbox.push_back(queued(&friend, "f1"));
2219+
2220+ let _ = app.update(Event::StartFlush, &mut m);
2221+ assert_eq!(m.in_flight.as_ref().unwrap().count, 2);
2222+
2223+ // Blocked mid-flight: the two items in the air are dropped from the queue.
2224+ let _ = app.update(
2225+ Event::SetBlocked {
2226+ peer: enemy.clone(),
2227+ blocked: true,
2228+ },
2229+ &mut m,
2230+ );
2231+ assert_eq!(m.outbox.len(), 1);
2232+
2233+ // The send lands. It must pop nothing.
2234+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
2235+ assert_eq!(m.outbox.len(), 1, "the friend's message is untouched");
2236+ assert_eq!(m.outbox[0].recipient, friend);
2237+ }
2238+
2239+ // -----------------------------------------------------------------------
2240+ // delete contact
2241+ // -----------------------------------------------------------------------
2242+
2243+ /// Blocking only silences a peer; the entry stayed in the list forever.
2244+ #[test]
2245+ fn deleting_a_contact_forgets_the_contact_the_history_and_the_queue() {
2246+ let app = Skrepka;
2247+ let mut m = with_identity();
2248+ let peer = peer_hex(9);
2249+ let other = peer_hex(11);
2250+ for p in [&peer, &other] {
2251+ m.contacts
2252+ .insert(p.clone(), Contact::new(p.clone(), String::new(), 0));
2253+ m.messages.insert(
2254+ p.clone(),
2255+ vec![StoredMessage {
2256+ id: "m1".into(),
2257+ body: "hi".into(),
2258+ ts: 1,
2259+ outgoing: false,
2260+ delivered: false,
2261+ }],
2262+ );
2263+ m.outbox.push_back(queued(p, "queued"));
2264+ }
2265+ m.active_peer = Some(peer.clone());
2266+ m.page = Page::Chat;
2267+
2268+ let mut cmd = app.update(Event::DeleteContact { peer: peer.clone() }, &mut m);
2269+
2270+ assert!(!m.contacts.contains_key(&peer));
2271+ assert!(!m.messages.contains_key(&peer));
2272+ assert!(
2273+ m.outbox.iter().all(|i| i.recipient != peer),
2274+ "nothing more goes to a peer we have forgotten"
2275+ );
2276+ assert_eq!(m.page, Page::Conversations, "the open chat is closed");
2277+ assert!(m.active_peer.is_none());
2278+
2279+ // Everyone else is untouched.
2280+ assert!(m.contacts.contains_key(&other));
2281+ assert_eq!(m.messages[&other].len(), 1);
2282+ assert_eq!(m.outbox.len(), 1);
2283+
2284+ // The history blob is *deleted*, not left behind holding what the user just
2285+ // asked us to forget.
2286+ let deleted = cmd.effects().any(|e| {
2287+ matches!(e, Effect::KeyValue(req)
2288+ if matches!(&req.operation, crux_kv::KeyValueOperation::Delete { key } if key == &k_messages(&peer)))
2289+ });
2290+ assert!(deleted, "the messages:<peer> key is removed");
2291+ }
2292+
2293+ /// Deleting a peer we are not looking at must not throw the user out of the
2294+ /// chat they *are* looking at.
2295+ #[test]
2296+ fn deleting_another_contact_leaves_the_open_chat_alone() {
2297+ let app = Skrepka;
2298+ let mut m = with_identity();
2299+ let open = peer_hex(9);
2300+ let doomed = peer_hex(11);
2301+ for p in [&open, &doomed] {
2302+ m.contacts
2303+ .insert(p.clone(), Contact::new(p.clone(), String::new(), 0));
2304+ }
2305+ m.active_peer = Some(open.clone());
2306+ m.page = Page::Chat;
2307+
2308+ let _ = app.update(Event::DeleteContact { peer: doomed }, &mut m);
2309+
2310+ assert_eq!(m.page, Page::Chat);
2311+ assert_eq!(m.active_peer.as_deref(), Some(open.as_str()));
2312+ }
2313+
2314+ // -----------------------------------------------------------------------
2315+ // watermark clamping
2316+ // -----------------------------------------------------------------------
2317+
2318+ /// An ack at `ts = i64::MAX` used to park `last_ack_ts` a minute into the
2319+ /// future — and every honest ack that followed, the ones that actually mark our
2320+ /// messages delivered, read as stale and was dropped. A peer could silence its
2321+ /// own delivery receipts forever with one payload.
2322+ #[test]
2323+ fn a_far_future_ack_cannot_suppress_the_acks_that_follow_it() {
2324+ let app = Skrepka;
2325+ let mut m = with_identity();
2326+ let peer = peer_hex(9);
2327+ m.contacts
2328+ .insert(peer.clone(), Contact::new(peer.clone(), String::new(), 0));
2329+ m.messages.insert(
2330+ peer.clone(),
2331+ vec![StoredMessage {
2332+ id: "m1".into(),
2333+ body: "hi".into(),
2334+ ts: 1,
2335+ outgoing: true,
2336+ delivered: false,
2337+ }],
2338+ );
2339+
2340+ // The poisoned ack: acks nothing we sent, but sets the watermark.
2341+ let ack = Payload::DeliveryAck {
2342+ ack_ids: vec!["nothing".into()],
2343+ };
2344+ let page = page_from(9, &m, &ack, i64::MAX);
2345+ let _ = app.ingest_poll(&mut m, page);
2346+ assert!(
2347+ m.contacts[&peer].last_ack_ts <= now_ms(),
2348+ "the watermark cannot be pushed into the future"
2349+ );
2350+
2351+ // A perfectly ordinary ack, sent now. It must still land.
2352+ let ack = Payload::DeliveryAck {
2353+ ack_ids: vec!["m1".into()],
2354+ };
2355+ let page = page_from(9, &m, &ack, now_ms());
2356+ let _ = app.ingest_poll(&mut m, page);
2357+ assert!(
2358+ m.messages[&peer][0].delivered,
2359+ "an honest ack after a far-future one must still be honoured"
2360+ );
2361+ }
2362+
2363+ /// The same trick against `last_profile_ts`: a profile at `ts = i64::MAX` would
2364+ /// freeze the peer's own entry, since every honest update they sent afterwards
2365+ /// would read as a stale replay of itself.
2366+ #[test]
2367+ fn a_far_future_profile_cannot_freeze_the_peers_own_profile() {
2368+ let app = Skrepka;
2369+ let mut m = with_identity();
2370+ let peer = peer_hex(9);
2371+ m.contacts
2372+ .insert(peer.clone(), Contact::new(peer.clone(), String::new(), 0));
2373+
2374+ let profile = |name: &str| Payload::Profile {
2375+ display_name: name.into(),
2376+ bio: String::new(),
2377+ photo: None,
2378+ };
2379+ let page = page_from(9, &m, &profile("Far"), i64::MAX);
2380+ let _ = app.ingest_poll(&mut m, page);
2381+ assert!(m.contacts[&peer].last_profile_ts <= now_ms());
2382+
2383+ let page = page_from(9, &m, &profile("Now"), now_ms());
2384+ let _ = app.ingest_poll(&mut m, page);
2385+ assert_eq!(
2386+ m.contacts[&peer].display_name, "Now",
2387+ "a later, honest profile must still apply"
2388+ );
2389+ }
2390+
2391+ // -----------------------------------------------------------------------
2392+ // conversation ordering
2393+ // -----------------------------------------------------------------------
2394+
2395+ /// The conversation is held in `ts` order rather than re-sorted per arrival.
2396+ /// `trim_history` ages out the *front*, so an out-of-order insert does not just
2397+ /// render wrong — it throws away the wrong message.
2398+ #[test]
2399+ fn messages_are_inserted_in_timestamp_order() {
2400+ let app = Skrepka;
2401+ let mut m = with_identity();
2402+ let peer = peer_hex(9);
2403+
2404+ // Arrive out of order, including a tie.
2405+ for (i, ts) in [300i64, 100, 200, 200].into_iter().enumerate() {
2406+ let page = page_from(9, &m, &text(&format!("m{i}"), "x"), ts);
2407+ let _ = app.ingest_poll(&mut m, page);
2408+ }
2409+ // ...and an outgoing message, which used to be appended blindly.
2410+ m.active_peer = Some(peer.clone());
2411+ m.compose = "mine".into();
2412+ let _ = app.update(Event::SendText, &mut m);
2413+
2414+ let convo = &m.messages[&peer];
2415+ assert!(
2416+ convo.windows(2).all(|w| w[0].ts <= w[1].ts),
2417+ "the conversation is sorted: {:?}",
2418+ convo.iter().map(|s| s.ts).collect::<Vec<_>>()
2419+ );
2420+ // Ties keep arrival order, exactly as the stable sort used to give.
2421+ let tied: Vec<&str> = convo
2422+ .iter()
2423+ .filter(|s| s.ts == 200)
2424+ .map(|s| s.id.as_str())
2425+ .collect();
2426+ assert_eq!(tied, vec!["m2", "m3"]);
2427+ }
2428+
15622429 /// PROTOCOL.md §4: a blocked peer's messages are not shown — and acking them
15632430 /// would tell the blocked peer we are online and reading.
15642431 #[test]
@@ -1674,10 +2541,8 @@ mod tests {
16742541 let mut m = with_identity();
16752542 m.token = Some("t".into());
16762543 let bad = hex::encode([0xabu8; 32]); // 32 bytes, not a curve point
1677− m.outbox.push_back(OutboxItem {
1678− recipient: bad,
1679− envelope_json: Arc::new("{}".into()),
1680− });
2544+ m.outbox
2545+ .push_back(OutboxItem::new(bad, Arc::new("{}".into())));
16812546
16822547 let _ = app.update(Event::StartFlush, &mut m);
16832548
@@ -1855,15 +2720,15 @@ mod tests {
18552720
18562721 // An ack to the peer we are about to block, plus unrelated mail.
18572722 for recipient in [&enemy, &friend] {
1858− m.outbox.push_back(OutboxItem {
1859− recipient: recipient.clone(),
1860− envelope_json: Arc::new(protocol::serialize_payload(
2723+ m.outbox.push_back(OutboxItem::new(
2724+ recipient.clone(),
2725+ Arc::new(protocol::serialize_payload(
18612726 &Payload::DeliveryAck {
18622727 ack_ids: vec!["m1".into()],
18632728 },
18642729 1,
18652730 )),
1866− });
2731+ ));
18672732 }
18682733
18692734 let _ = app.update(
@@ -1952,11 +2817,11 @@ mod tests {
19522817 let app = Skrepka;
19532818 let mut m = with_identity();
19542819 m.token = Some("t".into());
1955− m.flushing = true;
1956− m.outbox.push_back(OutboxItem {
1957− recipient: peer_hex(9),
1958− envelope_json: Arc::new("{}".into()),
1959− });
2820+ m.outbox
2821+ .push_back(OutboxItem::new(peer_hex(9), Arc::new("{}".into())));
2822+
2823+ let _ = app.update(Event::StartFlush, &mut m);
2824+ assert!(m.flushing);
19602825
19612826 let resp = crux_http::testing::ResponseBuilder::ok().body(Vec::new()).build();
19622827 let _ = app.update(Event::SendResult(Ok(resp)), &mut m);
@@ -1970,11 +2835,10 @@ mod tests {
19702835 let app = Skrepka;
19712836 let mut m = with_identity();
19722837 m.token = Some("t".into());
1973− for _ in 0..2 {
1974− m.outbox.push_back(OutboxItem {
1975− recipient: peer_hex(9),
1976− envelope_json: Arc::new("{}".into()),
1977− });
2838+ // Two different peers, or the batching would send both at once.
2839+ for seed in [9u8, 11] {
2840+ m.outbox
2841+ .push_back(OutboxItem::new(peer_hex(seed), Arc::new("{}".into())));
19782842 }
19792843 // First send goes out and marks the model as flushing.
19802844 let _ = app.update(Event::StartFlush, &mut m);