@@ -73,6 +73,24 @@ const MAX_POLL_TOTAL_BYTES: usize = 64 * 1024 * 1024;
73/// relay, so leave the margin.
73/// relay, so leave the margin.
74const POLL_WATCHDOG_MS: u64 = 90_000;
74const POLL_WATCHDOG_MS: u64 = 90_000;
76+/// Messages per `/messages` request.
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;
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.
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;
77const K_SETTINGS: &str = "settings";
95const K_SETTINGS: &str = "settings";
78const K_PROFILE: &str = "profile";
96const K_PROFILE: &str = "profile";
@@ -120,6 +138,10 @@ pub enum Event {
120 SendText,
138 SendText,
121 AddContact { input: String, nickname: String },
139 AddContact { input: String, nickname: String },
122 SetBlocked { peer: String, blocked: bool },
140 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 },
123 SaveProfile {
145 SaveProfile {
124 display_name: String,
146 display_name: String,
125 bio: String,
147 bio: String,
@@ -220,6 +242,18 @@ pub enum Page {
220/// `cursor`, `outbox`). `Connect` waits for all of them.
242/// `cursor`, `outbox`). `Connect` waits for all of them.
221const STARTUP_LOADS: u8 = 5;
243const STARTUP_LOADS: u8 = 5;
245+/// The batch a live `/messages` request is carrying: the first `count` items of
246+/// the outbox, all bound for `recipient`.
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.
223pub struct Model {
257pub struct Model {
224 /// Zeroized on drop: this is the Ed25519 seed, the one secret whose leak
258 /// Zeroized on drop: this is the Ed25519 seed, the one secret whose leak
225 /// costs the user their identity.
259 /// costs the user their identity.
@@ -261,7 +295,16 @@ pub struct Model {
261 /// flight from before the app was backgrounded must not install a token or
295 /// flight from before the app was backgrounded must not install a token or
262 /// overwrite `conn` on top of the attempt that replaced it.
296 /// overwrite `conn` on top of the attempt that replaced it.
263 auth_gen: u64,
297 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.
264 flushing: bool,
301 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>,
265 /// Startup kv loads still outstanding. `Connect` only fires at 0, so a poll
308 /// Startup kv loads still outstanding. `Connect` only fires at 0, so a poll
266 /// can't land and be persisted against a half-loaded model — which would
309 /// can't land and be persisted against a half-loaded model — which would
267 /// then be clobbered by the loads still in flight.
310 /// then be clobbered by the loads still in flight.
@@ -291,6 +334,7 @@ impl Default for Model {
291 authenticating: false,
334 authenticating: false,
292 auth_gen: 0,
335 auth_gen: 0,
293 flushing: false,
336 flushing: false,
294 loads_pending: 0,
338 loads_pending: 0,
295 page: Page::default(),
339 page: Page::default(),
296 active_peer: None,
340 active_peer: None,
@@ -343,6 +387,26 @@ impl Model {
343 self.authenticating = false;
387 self.authenticating = false;
344 self.auth_gen = self.auth_gen.wrapping_add(1);
388 self.auth_gen = self.auth_gen.wrapping_add(1);
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.
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;
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)
348// ---------------------------------------------------------------------------
412// ---------------------------------------------------------------------------
@@ -693,11 +757,34 @@ impl App for Skrepka {
693 // signal the block is supposed to cut off. Drop it all: once
757 // signal the block is supposed to cut off. Drop it all: once
694 // blocked, nothing more goes to this peer.
758 // blocked, nothing more goes to this peer.
695 model.outbox.retain(|item| item.recipient != peer);
759 model.outbox.retain(|item| item.recipient != peer);
760+ model.detach_in_flight(&peer);
696 return Command::all([self.persist_contacts(model), self.persist_outbox(model)])
761 return Command::all([self.persist_contacts(model), self.persist_outbox(model)])
697 .and(render());
762 .and(render());
699 self.persist_contacts(model).and(render())
764 self.persist_contacts(model).and(render())
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;
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),
701 Event::SaveProfile {
788 Event::SaveProfile {
702 display_name,
789 display_name,
@@ -720,11 +807,12 @@ impl App for Skrepka {
720 // profile broadcast would keep feeding them our display name,
807 // profile broadcast would keep feeding them our display name,
721 // avatar, and a liveness signal.
808 // avatar, and a liveness signal.
723− // Each item carries its own copy of the serialized payload, so a
810+ // One `Arc<String>` for the whole fan-out: the payload carries a
724− // profile with a photo is duplicated (base64 and all) once per
811+ // base64 photo, and a copy per contact would be megabytes in memory
725− // contact, in memory and in the `outbox` kv blob. Acceptable at
812+ // for a list of any size. (The `outbox` kv blob still serializes the
726− // today's contact counts; the fix is to intern the payload and
813+ // string once per item — serde has no idea the `Arc`s are shared —
727− // have `OutboxItem` reference it by id.
814+ // but the supersede-in-place below caps that at one pending profile
815+ // per contact, which is the bound that actually matters.)
728 let recipients: Vec<String> = model
816 let recipients: Vec<String> = model
729 .contacts
817 .contacts
730 .values()
818 .values()
@@ -732,11 +820,28 @@ impl App for Skrepka {
732 .map(|c| c.pubkey.clone())
820 .map(|c| c.pubkey.clone())
733 .collect();
821 .collect();
734 let payload_json = Arc::new(protocol::serialize_payload(&payload, ts));
822 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();
735 for peer in recipients {
828 for peer in recipients {
736− model.outbox.push_back(OutboxItem {
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.
737− recipient: peer,
838+ .find(|i| i.is_profile() && i.recipient == peer)
840+ Some(pending) => pending.envelope_json = Arc::clone(&payload_json),
738− envelope_json: Arc::clone(&payload_json),
843+ .push_back(OutboxItem::profile(peer, Arc::clone(&payload_json))),
741 Command::all([
846 Command::all([
742 KeyValue::set(K_PROFILE, json_bytes(&model.profile)).then_send(Event::Saved),
847 KeyValue::set(K_PROFILE, json_bytes(&model.profile)).then_send(Event::Saved),
@@ -957,22 +1062,30 @@ impl App for Skrepka {
957 // The in-flight send is over; `flush_next` refuses to run while
1062 // The in-flight send is over; `flush_next` refuses to run while
958 // this is set, so it must be cleared before *any* branch below.
1063 // this is set, so it must be cleared before *any* branch below.
959 model.flushing = false;
1064 model.flushing = false;
1065+ let batch = model.in_flight.take();
960 let status = u16::from(resp.status());
1066 let status = u16::from(resp.status());
961 if status == 401 {
1067 if status == 401 {
962 model.token = None;
1068 model.token = None;
1069+ // The batch stays exactly as it is — re-auth, then re-send it.
963 return Command::event(Event::Authenticate);
1070 return Command::event(Event::Authenticate);
965 if (200..300).contains(&status) || status == 400 {
1072 if (200..300).contains(&status) || status == 400 {
966− // success, or a permanent rejection (self_send/invalid) — drop it.
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) {
967− model.outbox.pop_front();
1077+ model.outbox.pop_front();
968 return self.persist_outbox(model).and(self.flush_next(model));
1079 return self.persist_outbox(model).and(self.flush_next(model));
970− // transient error — stop; will resume on next connect/poll.
1081+ // A transient failure: a 5xx, a rate limit, a relay restarting.
971− render()
1082+ self.retry_batch(model, batch)
973 Event::SendResult(Err(_)) => {
1084 Event::SendResult(Err(_)) => {
1085+ // A transport error is the same transient failure as a 5xx.
974 model.flushing = false;
1086 model.flushing = false;
975− render()
1087+ let batch = model.in_flight.take();
1088+ self.retry_batch(model, batch)
@@ -1145,13 +1258,19 @@ impl Skrepka {
1146 let convo = model.messages.entry(peer.clone()).or_default();
1259 let convo = model.messages.entry(peer.clone()).or_default();
1147 let first_to_peer = !convo.iter().any(|m| m.outgoing);
1260 let first_to_peer = !convo.iter().any(|m| m.outgoing);
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*.
1148− convo.push(StoredMessage {
1266+ StoredMessage {
1149− id: id.clone(),
1267+ id: id.clone(),
1150− body: body.clone(),
1268+ body: body.clone(),
1152− outgoing: true,
1270+ outgoing: true,
1153− delivered: false,
1271+ delivered: false,
1155 trim_history(convo);
1274 trim_history(convo);
1157 // Auto-create a bare contact if messaging a brand-new key.
1276 // Auto-create a bare contact if messaging a brand-new key.
@@ -1161,21 +1280,23 @@ impl Skrepka {
1161 .or_insert_with(|| Contact::new(peer.clone(), String::new(), ts));
1280 .or_insert_with(|| Contact::new(peer.clone(), String::new(), ts));
1163 let text = Payload::Text { id, body };
1282 let text = Payload::Text { id, body };
1164− model.outbox.push_back(OutboxItem {
1283+ model.outbox.push_back(OutboxItem::new(
1165− recipient: peer.clone(),
1284+ peer.clone(),
1166− envelope_json: Arc::new(protocol::serialize_payload(&text, ts)),
1285+ Arc::new(protocol::serialize_payload(&text, ts)),
1168− // On first contact, also share our profile.
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.
1169 if first_to_peer {
1290 if first_to_peer {
1170 let profile = Payload::Profile {
1291 let profile = Payload::Profile {
1171 display_name: model.profile.display_name.clone(),
1292 display_name: model.profile.display_name.clone(),
1172 bio: model.profile.bio.clone(),
1293 bio: model.profile.bio.clone(),
1173 photo: model.profile.photo.clone(),
1294 photo: model.profile.photo.clone(),
1175− model.outbox.push_back(OutboxItem {
1296+ model.outbox.push_back(OutboxItem::profile(
1176− recipient: peer.clone(),
1297+ peer.clone(),
1177− envelope_json: Arc::new(protocol::serialize_payload(&profile, ts)),
1298+ Arc::new(protocol::serialize_payload(&profile, ts)),
1181 Command::all([
1302 Command::all([
@@ -1187,8 +1308,23 @@ impl Skrepka {
1187 .and(render())
1308 .and(render())
1190− /// Encrypt and send the head of the outbox, discarding any unsendable items
1311+ /// Encrypt and send a batch from the head of the outbox, discarding any
1191− /// ahead of it.
1312+ /// unsendable items in the way.
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.
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.
1193 /// The discards happen in one pass, followed by a single `persist_outbox`.
1329 /// The discards happen in one pass, followed by a single `persist_outbox`.
1194 /// The previous shape recursed once per bad item and emitted a full outbox
1330 /// The previous shape recursed once per bad item and emitted a full outbox
@@ -1203,66 +1339,145 @@ impl Skrepka {
1203 return Command::done();
1339 return Command::done();
1342+ let now = now_ms();
1206− let mut dropped = false;
1343+ let mut dirty = false;
1207− let sendable = loop {
1344+ let mut recipient: Option<String> = None;
1345+ let mut messages: Vec<Envelope> = Vec::new();
1346+ let mut bytes = 0usize;
1348+ while messages.len() < MAX_SEND_BATCH {
1208− let Some(item) = model.outbox.front() else {
1349+ let Some(item) = model.outbox.get(messages.len()) else {
1209− break None;
1350+ break;
1211− match encrypt_for(&id, item) {
1352+ if recipient.as_ref().is_some_and(|r| r != &item.recipient) {
1212− Some(blob) => break Some((item.recipient.clone(), blob)),
1353+ break;
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)
1221− model.outbox.pop_front();
1225− };
1355+ let Some(blob) = encrypt_for(&id, item) else {
1226−
1356+ // An unusable recipient key can never be encrypted to, so the item
1227− let persist = if dropped {
1357+ // has to go or it blocks the head of the outbox forever — but say
1228− self.persist_outbox(model)
1358+ // so, rather than letting the message vanish. (Only reachable at
1229− } else {
1359+ // the head: an item deeper in the batch shares the recipient of one
1230− Command::done()
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)
1365+ model.outbox.remove(messages.len());
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 {
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,
1233− let Some((recipient, blob)) = sendable else {
1385+ let Some(recipient) = recipient else {
1234 // Nothing left to send. Render only if we dropped something, so the
1386 // Nothing left to send. Render only if we dropped something, so the
1235 // user sees why their message went away.
1387 // user sees why their message went away.
1236− return if dropped {
1388+ return if dirty {
1237− persist.and(render())
1389+ self.persist_outbox(model).and(render())
1238 } else {
1390 } else {
1239 Command::done()
1391 Command::done()
1243− model.flushing = true;
1395+ // Stamp the items now going out for the first time. This is what the TTL
1244− let batch = SendBatch {
1396+ // in `OutboxItem::is_expired` measures from, so it has to be durable —
1245− messages: vec![Envelope {
1397+ // otherwise a relaunch resets the clock and a permanently-stuck item is
1246− to: recipient,
1398+ // immortal again.
1247− encrypted_blob: hex::encode(blob),
1399+ for item in model.outbox.iter_mut().take(messages.len()) {
1248− }],
1400+ if item.first_attempt == 0 {
1401+ item.first_attempt = now;
1405+ let persist = if dirty {
1406+ self.persist_outbox(model)
1411+ model.flushing = true;
1412+ model.in_flight = Some(InFlight {
1414+ count: messages.len(),
1250 let url = format!("{}/messages", model.settings.server_url);
1416 let url = format!("{}/messages", model.settings.server_url);
1251 match Http::post(url)
1417 match Http::post(url)
1252 .header("authorization", format!("Bearer {token}"))
1418 .header("authorization", format!("Bearer {token}"))
1253− .body_json(&batch)
1419+ .body_json(&SendBatch { messages })
1255 Ok(req) => persist.and(req.build().then_send(Event::SendResult)),
1421 Ok(req) => persist.and(req.build().then_send(Event::SendResult)),
1256 Err(_) => {
1422 Err(_) => {
1257 // Not a dead loop: the next StartFlush (from a poll page or a
1423 // Not a dead loop: the next StartFlush (from a poll page or a
1258 // fresh send) retries the head of the outbox.
1424 // fresh send) retries the head of the outbox.
1259 model.flushing = false;
1425 model.flushing = false;
1426+ model.in_flight = None;
1260 model.error = Some("could not build send request".into());
1427 model.error = Some("could not build send request".into());
1261 persist.and(render())
1428 persist.and(render())
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
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 {
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) {
1460+ kept.push_back(item);
1462+ model.outbox = kept;
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());
1470+ model.error = Some(format!(
1471+ "gave up sending {expired} message(s) to {}",
1472+ trunc_ob(&batch.recipient)
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))
1266 /// Decrypt and fold a poll page into the model; queue delivery acks.
1481 /// Decrypt and fold a poll page into the model; queue delivery acks.
1267 fn ingest_poll(&self, model: &mut Model, page: PollResp) -> Command<Effect, Event> {
1482 fn ingest_poll(&self, model: &mut Model, page: PollResp) -> Command<Effect, Event> {
1268 if page.cursor > model.cursor {
1483 if page.cursor > model.cursor {
@@ -1349,15 +1564,21 @@ impl Skrepka {
1350 // dedup by id — O(1) via HashSet
1565 // dedup by id — O(1) via HashSet
1351 if !id_set.contains(&msg_id) {
1566 if !id_set.contains(&msg_id) {
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.
1352− convo.push(StoredMessage {
1573+ StoredMessage {
1353− id: msg_id.clone(),
1574+ id: msg_id.clone(),
1356− outgoing: false,
1577+ outgoing: false,
1357− delivered: false,
1578+ delivered: false,
1359 id_set.insert(msg_id.clone());
1581 id_set.insert(msg_id.clone());
1360− convo.sort_by_key(|m| m.ts);
1361 trim_history(convo);
1582 trim_history(convo);
1362 touched_convos.insert(sender.clone());
1583 touched_convos.insert(sender.clone());
@@ -1373,9 +1594,15 @@ impl Skrepka {
1373 if stale {
1594 if stale {
1374 continue;
1595 continue;
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.
1377 if let Some(c) = model.contacts.get_mut(&sender) {
1604 if let Some(c) = model.contacts.get_mut(&sender) {
1378− c.last_ack_ts = ts;
1605+ c.last_ack_ts = ts.min(now);
1379 contacts_dirty = true;
1606 contacts_dirty = true;
1381 let ack_set: HashSet<String> = ack_ids.iter().cloned().collect();
1608 let ack_set: HashSet<String> = ack_ids.iter().cloned().collect();
@@ -1404,14 +1631,16 @@ impl Skrepka {
1404 let Some(c) = model.contacts.get_mut(&sender) else {
1631 let Some(c) = model.contacts.get_mut(&sender) else {
1405 continue;
1632 continue;
1407− // Ignore stale profile replays (PROTOCOL.md §4). The clamped
1634+ // Ignore stale profile replays (PROTOCOL.md §4). The watermark
1408− // `ts` is what gets recorded, so a peer cannot park
1635+ // is clamped to *now* for the same reason `last_ack_ts` is: a
1409− // `last_profile_ts` at `i64::MAX` and freeze its own profile.
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.
1410 if ts >= c.last_profile_ts {
1639 if ts >= c.last_profile_ts {
1411 c.display_name = display_name;
1640 c.display_name = display_name;
1412 c.bio = bio;
1641 c.bio = bio;
1413 c.photo = photo;
1642 c.photo = photo;
1414− c.last_profile_ts = ts;
1643+ c.last_profile_ts = ts.min(now);
1415 contacts_dirty = true;
1644 contacts_dirty = true;
@@ -1433,10 +1662,12 @@ impl Skrepka {
1433 let payload = Payload::DeliveryAck {
1662 let payload = Payload::DeliveryAck {
1434 ack_ids: batch.to_vec(),
1663 ack_ids: batch.to_vec(),
1665+ // Consecutive and same-recipient, so `flush_next` puts every one of
1666+ // a sender's ack batches into a single `/messages` request.
1436− model.outbox.push_back(OutboxItem {
1667+ model.outbox.push_back(OutboxItem::new(
1437− recipient: sender.clone(),
1668+ sender.clone(),
1438− envelope_json: Arc::new(protocol::serialize_payload(&payload, ack_ts)),
1669+ Arc::new(protocol::serialize_payload(&payload, ack_ts)),
@@ -1468,6 +1699,20 @@ fn backoff_ms(retries: u32) -> u64 {
1468 (3000u64.saturating_mul(1u64 << retries.min(4))).min(30_000)
1699 (3000u64.saturating_mul(1u64 << retries.min(4))).min(30_000)
1702+/// Insert a message into a conversation held in `ts` order.
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.
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);
1471/// Age out local history: keep only the most recent `MAX_MESSAGES_PER_PEER`
1716/// Age out local history: keep only the most recent `MAX_MESSAGES_PER_PEER`
1472/// messages of a conversation (PROTOCOL.md §9). Callers keep `convo` sorted by
1717/// messages of a conversation (PROTOCOL.md §9). Callers keep `convo` sorted by
1473/// `ts`, so the oldest messages are at the front. The subsequent kv write
1718/// `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> {
1500mod tests {
1745mod tests {
1501 use super::*;
1746 use super::*;
1502 use crate::crypto::Identity;
1747 use crate::crypto::Identity;
1748+ use crate::model::{MAX_OUTBOX_RETRIES, OUTBOX_TTL_MS};
1504 fn with_identity() -> Model {
1750 fn with_identity() -> Model {
1505 let id = Identity::from_seed(&[3u8; 32]);
1751 let id = Identity::from_seed(&[3u8; 32]);
@@ -1559,6 +1805,627 @@ mod tests {
1808+ fn queued(recipient: &str, body: &str) -> OutboxItem {
1810+ recipient.to_string(),
1811+ Arc::new(protocol::serialize_payload(&text(body, body), 1)),
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)]
1821+ messages: Vec<Envelope>,
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)
1828+ .map(|b| b.messages)
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)
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);
1848+ // -----------------------------------------------------------------------
1849+ // outbox: retry budget and TTL
1850+ // -----------------------------------------------------------------------
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.
1855+ fn an_undeliverable_item_is_dropped_once_it_runs_out_of_retries() {
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"));
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");
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);
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");
1879+ sent_batches(&mut cmd).len(),
1881+ "the message behind it is sent immediately, not left for the next flush"
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}");
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.
1891+ fn an_item_older_than_the_ttl_is_dropped_on_its_next_failure() {
1893+ let mut m = with_identity();
1894+ m.token = Some("t".into());
1895+ let peer = peer_hex(9);
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);
1902+ send_and_answer(&app, &mut m, 503);
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")));
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".
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);
1917+ !item.is_expired(now_ms()),
1918+ "an unsent item must not age out against the epoch"
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.
1925+ fn the_first_attempt_stamp_is_persisted_when_the_item_goes_out() {
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();
1932+ let mut cmd = app.update(Event::StartFlush, &mut m);
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))
1940+ assert!(wrote_outbox, "the stamped outbox is persisted");
1943+ /// A transient failure below the budget changes nothing but the counter — the
1944+ /// message stays queued and goes out on the next flush.
1946+ fn a_transient_failure_within_budget_keeps_the_message() {
1948+ let mut m = with_identity();
1949+ m.token = Some("t".into());
1950+ m.outbox.push_back(queued(&peer_hex(9), "hi"));
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");
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());
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.
1967+ fn a_401_does_not_spend_the_retry_budget() {
1969+ let mut m = with_identity();
1970+ m.token = Some("t".into());
1971+ m.outbox.push_back(queued(&peer_hex(9), "hi"));
1973+ send_and_answer(&app, &mut m, 401);
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");
1980+ // -----------------------------------------------------------------------
1981+ // outbox: batching
1982+ // -----------------------------------------------------------------------
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.
1987+ fn consecutive_items_for_one_recipient_go_out_in_a_single_request() {
1989+ let mut m = with_identity();
1990+ m.token = Some("t".into());
1991+ let peer = peer_hex(9);
1993+ m.outbox.push_back(queued(&peer, &format!("m{i}")));
1996+ let mut cmd = app.update(Event::StartFlush, &mut m);
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));
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());
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.
2012+ fn a_batch_stops_at_the_first_recipient_change() {
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"));
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));
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);
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.
2039+ fn a_batch_is_capped_at_the_relays_limit() {
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}")));
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);
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");
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.
2060+ fn a_batch_is_capped_by_body_size_as_well_as_count() {
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)
2073+ .take(protocol::MAX_BODY_LEN)
2075+ for i in 0..MAX_SEND_BATCH {
2076+ m.outbox.push_back(OutboxItem::new(
2078+ Arc::new(protocol::serialize_payload(&text(&format!("m{i}"), &big), 1)),
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();
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");
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.
2094+ fn an_unsendable_head_is_dropped_and_the_batch_behind_it_still_goes_out() {
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"));
2103+ let mut cmd = app.update(Event::StartFlush, &mut m);
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")));
2111+ // -----------------------------------------------------------------------
2112+ // outbox: profile supersede
2113+ // -----------------------------------------------------------------------
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.
2118+ fn repeated_profile_edits_supersede_rather_than_pile_up() {
2120+ let mut m = with_identity();
2121+ let a = peer_hex(9);
2122+ let b = peer_hex(11);
2123+ for peer in [&a, &b] {
2125+ .insert(peer.clone(), Contact::new(peer.clone(), String::new(), 0));
2128+ for name in ["One", "Two", "Three"] {
2129+ let _ = app.update(
2130+ Event::SaveProfile {
2131+ display_name: name.into(),
2132+ bio: String::new(),
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");
2147+ _ => panic!("expected a profile"),
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.
2156+ fn a_profile_already_in_flight_is_not_rewritten_under_the_send() {
2158+ let mut m = with_identity();
2159+ m.token = Some("t".into());
2160+ let peer = peer_hex(9);
2162+ .insert(peer.clone(), Contact::new(peer.clone(), String::new(), 0));
2164+ let _ = app.update(
2165+ Event::SaveProfile {
2166+ display_name: "Old".into(),
2167+ bio: String::new(),
2172+ // It goes out and is still in flight.
2173+ let _ = app.update(Event::StartFlush, &mut m);
2174+ assert!(m.flushing);
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(),
2186+ assert_eq!(m.outbox.len(), 2, "the in-flight item is left alone");
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"
2196+ _ => panic!("expected a profile"),
2200+ // -----------------------------------------------------------------------
2201+ // in-flight bookkeeping
2202+ // -----------------------------------------------------------------------
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.
2208+ fn blocking_the_peer_mid_send_does_not_pop_someone_elses_messages() {
2210+ let mut m = with_identity();
2211+ m.token = Some("t".into());
2212+ let enemy = peer_hex(9);
2213+ let friend = peer_hex(11);
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"));
2220+ let _ = app.update(Event::StartFlush, &mut m);
2221+ assert_eq!(m.in_flight.as_ref().unwrap().count, 2);
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(),
2231+ assert_eq!(m.outbox.len(), 1);
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);
2239+ // -----------------------------------------------------------------------
2241+ // -----------------------------------------------------------------------
2243+ /// Blocking only silences a peer; the entry stayed in the list forever.
2245+ fn deleting_a_contact_forgets_the_contact_the_history_and_the_queue() {
2247+ let mut m = with_identity();
2248+ let peer = peer_hex(9);
2249+ let other = peer_hex(11);
2250+ for p in [&peer, &other] {
2252+ .insert(p.clone(), Contact::new(p.clone(), String::new(), 0));
2255+ vec![StoredMessage {
2263+ m.outbox.push_back(queued(p, "queued"));
2265+ m.active_peer = Some(peer.clone());
2266+ m.page = Page::Chat;
2268+ let mut cmd = app.update(Event::DeleteContact { peer: peer.clone() }, &mut m);
2270+ assert!(!m.contacts.contains_key(&peer));
2271+ assert!(!m.messages.contains_key(&peer));
2273+ m.outbox.iter().all(|i| i.recipient != peer),
2274+ "nothing more goes to a peer we have forgotten"
2276+ assert_eq!(m.page, Page::Conversations, "the open chat is closed");
2277+ assert!(m.active_peer.is_none());
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);
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)))
2290+ assert!(deleted, "the messages:<peer> key is removed");
2293+ /// Deleting a peer we are not looking at must not throw the user out of the
2294+ /// chat they *are* looking at.
2296+ fn deleting_another_contact_leaves_the_open_chat_alone() {
2298+ let mut m = with_identity();
2299+ let open = peer_hex(9);
2300+ let doomed = peer_hex(11);
2301+ for p in [&open, &doomed] {
2303+ .insert(p.clone(), Contact::new(p.clone(), String::new(), 0));
2305+ m.active_peer = Some(open.clone());
2306+ m.page = Page::Chat;
2308+ let _ = app.update(Event::DeleteContact { peer: doomed }, &mut m);
2310+ assert_eq!(m.page, Page::Chat);
2311+ assert_eq!(m.active_peer.as_deref(), Some(open.as_str()));
2314+ // -----------------------------------------------------------------------
2315+ // watermark clamping
2316+ // -----------------------------------------------------------------------
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.
2323+ fn a_far_future_ack_cannot_suppress_the_acks_that_follow_it() {
2325+ let mut m = with_identity();
2326+ let peer = peer_hex(9);
2328+ .insert(peer.clone(), Contact::new(peer.clone(), String::new(), 0));
2331+ vec![StoredMessage {
2340+ // The poisoned ack: acks nothing we sent, but sets the watermark.
2341+ let ack = Payload::DeliveryAck {
2342+ ack_ids: vec!["nothing".into()],
2344+ let page = page_from(9, &m, &ack, i64::MAX);
2345+ let _ = app.ingest_poll(&mut m, page);
2347+ m.contacts[&peer].last_ack_ts <= now_ms(),
2348+ "the watermark cannot be pushed into the future"
2351+ // A perfectly ordinary ack, sent now. It must still land.
2352+ let ack = Payload::DeliveryAck {
2353+ ack_ids: vec!["m1".into()],
2355+ let page = page_from(9, &m, &ack, now_ms());
2356+ let _ = app.ingest_poll(&mut m, page);
2358+ m.messages[&peer][0].delivered,
2359+ "an honest ack after a far-future one must still be honoured"
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.
2367+ fn a_far_future_profile_cannot_freeze_the_peers_own_profile() {
2369+ let mut m = with_identity();
2370+ let peer = peer_hex(9);
2372+ .insert(peer.clone(), Contact::new(peer.clone(), String::new(), 0));
2374+ let profile = |name: &str| Payload::Profile {
2375+ display_name: name.into(),
2376+ bio: String::new(),
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());
2383+ let page = page_from(9, &m, &profile("Now"), now_ms());
2384+ let _ = app.ingest_poll(&mut m, page);
2386+ m.contacts[&peer].display_name, "Now",
2387+ "a later, honest profile must still apply"
2391+ // -----------------------------------------------------------------------
2392+ // conversation ordering
2393+ // -----------------------------------------------------------------------
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.
2399+ fn messages_are_inserted_in_timestamp_order() {
2401+ let mut m = with_identity();
2402+ let peer = peer_hex(9);
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);
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);
2414+ let convo = &m.messages[&peer];
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<_>>()
2420+ // Ties keep arrival order, exactly as the stable sort used to give.
2421+ let tied: Vec<&str> = convo
2423+ .filter(|s| s.ts == 200)
2424+ .map(|s| s.id.as_str())
2426+ assert_eq!(tied, vec!["m2", "m3"]);
1562 /// PROTOCOL.md §4: a blocked peer's messages are not shown — and acking them
2429 /// PROTOCOL.md §4: a blocked peer's messages are not shown — and acking them
1563 /// would tell the blocked peer we are online and reading.
2430 /// would tell the blocked peer we are online and reading.
@@ -1674,10 +2541,8 @@ mod tests {
1674 let mut m = with_identity();
2541 let mut m = with_identity();
1675 m.token = Some("t".into());
2542 m.token = Some("t".into());
1676 let bad = hex::encode([0xabu8; 32]); // 32 bytes, not a curve point
2543 let bad = hex::encode([0xabu8; 32]); // 32 bytes, not a curve point
1677− m.outbox.push_back(OutboxItem {
2545+ .push_back(OutboxItem::new(bad, Arc::new("{}".into())));
1679− envelope_json: Arc::new("{}".into()),
1682 let _ = app.update(Event::StartFlush, &mut m);
2547 let _ = app.update(Event::StartFlush, &mut m);
@@ -1855,15 +2720,15 @@ mod tests {
1856 // An ack to the peer we are about to block, plus unrelated mail.
2721 // An ack to the peer we are about to block, plus unrelated mail.
1857 for recipient in [&enemy, &friend] {
2722 for recipient in [&enemy, &friend] {
1858− m.outbox.push_back(OutboxItem {
2723+ m.outbox.push_back(OutboxItem::new(
1859− recipient: recipient.clone(),
2724+ recipient.clone(),
1860− envelope_json: Arc::new(protocol::serialize_payload(
2725+ Arc::new(protocol::serialize_payload(
1861 &Payload::DeliveryAck {
2726 &Payload::DeliveryAck {
1862 ack_ids: vec!["m1".into()],
2727 ack_ids: vec!["m1".into()],
1869 let _ = app.update(
2734 let _ = app.update(
@@ -1952,11 +2817,11 @@ mod tests {
1952 let app = Skrepka;
2817 let app = Skrepka;
1953 let mut m = with_identity();
2818 let mut m = with_identity();
1954 m.token = Some("t".into());
2819 m.token = Some("t".into());
1955− m.flushing = true;
2820+ m.outbox
1956− m.outbox.push_back(OutboxItem {
2821+ .push_back(OutboxItem::new(peer_hex(9), Arc::new("{}".into())));
1957− recipient: peer_hex(9),
2822+
1958− envelope_json: Arc::new("{}".into()),
2823+ let _ = app.update(Event::StartFlush, &mut m);
1959− });
2824+ assert!(m.flushing);
1961 let resp = crux_http::testing::ResponseBuilder::ok().body(Vec::new()).build();
2826 let resp = crux_http::testing::ResponseBuilder::ok().body(Vec::new()).build();
1962 let _ = app.update(Event::SendResult(Ok(resp)), &mut m);
2827 let _ = app.update(Event::SendResult(Ok(resp)), &mut m);
@@ -1970,11 +2835,10 @@ mod tests {
1970 let app = Skrepka;
2835 let app = Skrepka;
1971 let mut m = with_identity();
2836 let mut m = with_identity();
1972 m.token = Some("t".into());
2837 m.token = Some("t".into());
2838+ // Two different peers, or the batching would send both at once.
1973− for _ in 0..2 {
2839+ for seed in [9u8, 11] {
1974− m.outbox.push_back(OutboxItem {
2841+ .push_back(OutboxItem::new(peer_hex(seed), Arc::new("{}".into())));
1975− recipient: peer_hex(9),
1976− envelope_json: Arc::new("{}".into()),
1979 // First send goes out and marks the model as flushing.
2843 // First send goes out and marks the model as flushing.
1980 let _ = app.update(Event::StartFlush, &mut m);
2844 let _ = app.update(Event::StartFlush, &mut m);