@@ -91,6 +91,15 @@ const MAX_SEND_BATCH: usize = 100;
9191/// relay's job to reject it, and its rejection is per-message and permanent.
9292const MAX_SEND_BATCH_BYTES: usize = 4 * 1024 * 1024;
9393
94+/// Ceiling on a `Retry-After` we will honour from a relay.
95+///
96+/// The header is a number the relay chooses and we obey, so without a cap a
97+/// single `Retry-After: 999999999` — from a hostile relay, or a merely broken
98+/// one — parks the outbox for thirty years with nothing on screen to say why.
99+/// Five minutes is longer than any real rate-limit window and still recovers on
100+/// its own.
101+const MAX_RETRY_AFTER_MS: u64 = 5 * 60 * 1000;
102+
94103// kv keys
95104const K_SETTINGS: &str = "settings";
96105const K_PROFILE: &str = "profile";
@@ -245,13 +254,16 @@ const STARTUP_LOADS: u8 = 5;
245254/// The batch a live `/messages` request is carrying: the first `count` items of
246255/// the outbox, all bound for `recipient`.
247256///
248−/// `SendResult` needs both. The count is what it pops on success and what it
257+/// `SendResult` needs all three. The count is what it pops on success and what it
249258/// charges a retry against on a transient failure; the recipient is what tells
250259/// `SetBlocked` and `DeleteContact` whether the items they are about to remove
251−/// are the very ones this send is going to pop.
260+/// are the very ones this send is going to pop; and `bytes` — the size of the
261+/// body we actually built — is what a `413` halves to find a batch this relay
262+/// will accept (see `Model::send_batch_budget`).
252263struct InFlight {
253264 recipient: String,
254265 count: usize,
266+ bytes: usize,
255267}
256268
257269pub struct Model {
@@ -305,6 +317,20 @@ pub struct Model {
305317 /// it (`SetBlocked`, `DeleteContact`): there is then nothing left to pop, and
306318 /// popping by count anyway would eat whichever messages slid into their place.
307319 in_flight: Option<InFlight>,
320+ /// What this relay will actually accept as a request body, in hex characters,
321+ /// once it has told us. `None` until then — meaning "assume
322+ /// `MAX_SEND_BATCH_BYTES`", which is a guess at a cap the protocol never
323+ /// publishes.
324+ ///
325+ /// A relay configured below our guess answers `413`, and a `413` read as a
326+ /// transient failure is a batch that is rebuilt identically, rejected
327+ /// identically, and retried until every item in it has burned its retry budget
328+ /// and been dropped — the outbox drains into the void without a single message
329+ /// leaving. So a `413` *measures* the cap instead: halve what we just tried and
330+ /// rebuild smaller. It is deliberately not reset on success — the batch that
331+ /// worked is evidence about the relay, and the relay's cap does not grow — only
332+ /// on a server change, where the evidence no longer applies.
333+ send_batch_budget: Option<usize>,
308334 /// Startup kv loads still outstanding. `Connect` only fires at 0, so a poll
309335 /// can't land and be persisted against a half-loaded model — which would
310336 /// then be clobbered by the loads still in flight.
@@ -335,6 +361,7 @@ impl Default for Model {
335361 auth_gen: 0,
336362 flushing: false,
337363 in_flight: None,
364+ send_batch_budget: None,
338365 loads_pending: 0,
339366 page: Page::default(),
340367 active_peer: None,
@@ -560,52 +587,76 @@ impl App for Skrepka {
560587 }
561588
562589 Event::LoadedSettings(res) => {
563− if let Some(s) = parse_kv::<Settings>(res) {
564− // Re-validate on the way in. The URL was normalized when the
565− // user typed it, but the blob on disk is not trustworthy: an
566− // older build wrote URLs this parser rejects, and a corrupted
567− // or hand-edited file would reach `Http::post`, which unwraps
568− // the parse. That is a panic on every launch — a boot loop
569− // with no way out from inside the app.
570− model.settings.server_url = normalize_server_url(&s.server_url)
571− .unwrap_or_else(|| DEFAULT_SERVER_URL.to_string());
590+ match parse_kv::<Settings>(res) {
591+ Ok(Some(s)) => {
592+ // Re-validate on the way in. The URL was normalized when the
593+ // user typed it, but the blob on disk is not trustworthy: an
594+ // older build wrote URLs this parser rejects, and a corrupted
595+ // or hand-edited file would reach `Http::post`, which unwraps
596+ // the parse. That is a panic on every launch — a boot loop
597+ // with no way out from inside the app.
598+ model.settings.server_url = normalize_server_url(&s.server_url)
599+ .unwrap_or_else(|| DEFAULT_SERVER_URL.to_string());
600+ }
601+ Ok(None) => {}
602+ Err(()) => {
603+ model.error = Some("could not read settings from storage".into());
604+ }
572605 }
573606 self.startup_done(model).and(render())
574607 }
575608 Event::LoadedProfile(res) => {
576− if let Some(p) = parse_kv::<OwnProfile>(res) {
577− model.profile = p;
609+ match parse_kv::<OwnProfile>(res) {
610+ Ok(Some(p)) => model.profile = p,
611+ Ok(None) => {}
612+ Err(()) => model.error = Some("could not read profile from storage".into()),
578613 }
579614 self.startup_done(model).and(render())
580615 }
581616 Event::LoadedContacts(res) => {
582617 let mut cmds = vec![self.startup_done(model)];
583− if let Some(list) = parse_kv::<Vec<Contact>>(res) {
584− model.contacts = list.into_iter().map(|c| (c.pubkey.clone(), c)).collect();
585− // Lazily load each conversation's messages.
586− cmds.extend(model.contacts.keys().map(|peer| {
587− let peer = peer.clone();
588− KeyValue::get(k_messages(&peer))
589− .then_send(move |r| Event::LoadedMessages(peer.clone(), r))
590− }));
618+ match parse_kv::<Vec<Contact>>(res) {
619+ Ok(Some(list)) => {
620+ model.contacts = list.into_iter().map(|c| (c.pubkey.clone(), c)).collect();
621+ // Lazily load each conversation's messages.
622+ cmds.extend(model.contacts.keys().map(|peer| {
623+ let peer = peer.clone();
624+ KeyValue::get(k_messages(&peer))
625+ .then_send(move |r| Event::LoadedMessages(peer.clone(), r))
626+ }));
627+ }
628+ Ok(None) => {}
629+ Err(()) => model.error = Some("could not read contacts from storage".into()),
591630 }
592631 Command::all(cmds).and(render())
593632 }
594633 Event::LoadedCursor(res) => {
595− if let Some(c) = parse_kv::<i64>(res) {
596− model.cursor = c;
634+ match parse_kv::<i64>(res) {
635+ Ok(Some(c)) => model.cursor = c,
636+ Ok(None) => {}
637+ Err(()) => model.error = Some("could not read cursor from storage".into()),
597638 }
598639 self.startup_done(model).and(render())
599640 }
600641 Event::LoadedOutbox(res) => {
601− if let Some(list) = parse_kv::<Vec<OutboxItem>>(res) {
602− model.outbox = list.into();
642+ match parse_kv::<Vec<OutboxItem>>(res) {
643+ Ok(Some(list)) => model.outbox = list.into(),
644+ Ok(None) => {}
645+ Err(()) => model.error = Some("could not read outbox from storage".into()),
603646 }
604647 self.startup_done(model).and(render())
605648 }
606649 Event::LoadedMessages(peer, res) => {
607− if let Some(list) = parse_kv::<Vec<StoredMessage>>(res) {
608− model.messages.insert(peer, list);
650+ match parse_kv::<Vec<StoredMessage>>(res) {
651+ Ok(Some(list)) => {
652+ model.messages.insert(peer, list);
653+ }
654+ Ok(None) => {}
655+ Err(()) => {
656+ // Don't overwrite a potentially-loaded conversation with
657+ // an empty default — the read failed, the data is still
658+ // on disk.
659+ }
609660 }
610661 render()
611662 }
@@ -683,6 +734,12 @@ impl App for Skrepka {
683734 model.settings.server_url = url;
684735 model.token = None;
685736 model.conn = ConnStatus::Offline;
737+ // Whatever we learned about the old relay's body cap was about *it*.
738+ // A new relay is entitled to a bigger one, and carrying a shrunken
739+ // budget over would cap our batches against a limit that no longer
740+ // exists — with no `413` left to correct it, since the new relay has
741+ // no reason to send one.
742+ model.send_batch_budget = None;
686743 // The cursor is *per relay*: it is that server's own sequence
687744 // number, and the next poll acks everything up to it. Carrying
688745 // relay A's cursor (roughly `now_ms()`) over to relay B tells B
@@ -1069,6 +1126,56 @@ impl App for Skrepka {
10691126 // The batch stays exactly as it is — re-auth, then re-send it.
10701127 return Command::event(Event::Authenticate);
10711128 }
1129+ if status == 429 {
1130+ // Rate-limited. Not a delivery failure and not the items'
1131+ // fault, so they are not charged a retry — a client that spent
1132+ // its delivery budget on the relay's load would drop perfectly
1133+ // good messages for being sent too enthusiastically. They stay
1134+ // queued untouched; we just wait.
1135+ //
1136+ // `Retry-After` is the relay saying how long. Only the
1137+ // delta-seconds form is read — the HTTP-date form falls back to
1138+ // our own backoff — and it is clamped either way, because
1139+ // obeying an unbounded number from a relay is how the outbox
1140+ // gets parked forever.
1141+ let delay = resp
1142+ .header("retry-after")
1143+ .map(|v| v.last().as_str())
1144+ .and_then(|v| v.trim().parse::<u64>().ok())
1145+ .map_or_else(
1146+ || Duration::from_millis(backoff_ms(model.poll_retries)),
1147+ |secs| {
1148+ Duration::from_millis(
1149+ secs.saturating_mul(1000).min(MAX_RETRY_AFTER_MS),
1150+ )
1151+ },
1152+ );
1153+ return Time::notify_after(delay)
1154+ .0
1155+ .then_send(|_| Event::StartFlush)
1156+ .and(render());
1157+ }
1158+ if status == 413 {
1159+ // The body outgrew what this relay accepts. `MAX_SEND_BATCH_BYTES`
1160+ // is only a guess at that cap, and this relay's is smaller — so
1161+ // treating the `413` as transient rebuilds the very same batch,
1162+ // which is rejected the very same way, forever.
1163+ //
1164+ // Measure the cap instead: halve what this batch actually
1165+ // weighed and rebuild under that. The items are untouched (we
1166+ // only pop on 200/400), so there is nothing to re-queue.
1167+ //
1168+ // Only a *splittable* batch takes this path. A lone item over
1169+ // the cap cannot be made any smaller, and halving the budget
1170+ // around it would spin — `flush_next` always takes the first
1171+ // item whatever it weighs. It falls through to `retry_batch`,
1172+ // whose budget is what eventually drops it, with an error the
1173+ // user can see.
1174+ if let Some(b) = batch.as_ref().filter(|b| b.count > 1) {
1175+ model.send_batch_budget = Some(b.bytes / 2);
1176+ return self.flush_next(model);
1177+ }
1178+ }
10721179 if (200..300).contains(&status) || status == 400 {
10731180 // The batch was accepted, or rejected outright (`self_send` /
10741181 // `invalid_message` — both permanent, and the relay takes a
@@ -1182,7 +1289,6 @@ impl App for Skrepka {
11821289 active_peer_ob: hex_to_ob(&active_peer),
11831290 active_peer_blocked: active_contact.is_some_and(|c| c.blocked),
11841291 messages,
1185− compose: model.compose.clone(),
11861292 error: model.error.clone().unwrap_or_default(),
11871293 }
11881294 }
@@ -1344,6 +1450,9 @@ impl Skrepka {
13441450 let mut recipient: Option<String> = None;
13451451 let mut messages: Vec<Envelope> = Vec::new();
13461452 let mut bytes = 0usize;
1453+ // What we believe this relay's body cap to be — `MAX_SEND_BATCH_BYTES`
1454+ // until a `413` tells us otherwise.
1455+ let batch_budget = model.send_batch_budget.unwrap_or(MAX_SEND_BATCH_BYTES);
13471456
13481457 while messages.len() < MAX_SEND_BATCH {
13491458 let Some(item) = model.outbox.get(messages.len()) else {
@@ -1371,7 +1480,7 @@ impl Skrepka {
13711480 // payload must still reach the relay, which rejects it permanently
13721481 // (`400`) and so gets it out of the queue for good. Past the first,
13731482 // stop before the body outgrows what a relay will accept.
1374− if !messages.is_empty() && bytes + hex_blob.len() > MAX_SEND_BATCH_BYTES {
1483+ if !messages.is_empty() && bytes + hex_blob.len() > batch_budget {
13751484 break;
13761485 }
13771486 bytes += hex_blob.len();
@@ -1412,6 +1521,7 @@ impl Skrepka {
14121521 model.in_flight = Some(InFlight {
14131522 recipient,
14141523 count: messages.len(),
1524+ bytes,
14151525 });
14161526 let url = format!("{}/messages", model.settings.server_url);
14171527 match Http::post(url)
@@ -1524,6 +1634,17 @@ impl Skrepka {
15241634 };
15251635 let sender = dec.sender_hex;
15261636
1637+ // Mail from ourselves. A relay cannot forge this — the sender identity
1638+ // is signed inside the AEAD — but it can *echo back* a blob we sent
1639+ // through it, and one addressed to ourselves decrypts perfectly well.
1640+ // Nothing downstream would notice: it would auto-create a contact for
1641+ // our own key (which `AddContact` explicitly refuses to do), file our
1642+ // own words as an incoming message, and queue a delivery ack addressed
1643+ // to us — which flushes, comes back, and is acked again.
1644+ if sender == model.my_pubkey {
1645+ continue;
1646+ }
1647+
15271648 // Blocking cuts the peer off entirely (PROTOCOL.md §4) — not just
15281649 // their texts. A blocked peer must not be able to rewrite their
15291650 // display name and avatar in our contact list, mark our messages
@@ -1581,8 +1702,10 @@ impl Skrepka {
15811702 id_set.insert(msg_id.clone());
15821703 trim_history(convo);
15831704 touched_convos.insert(sender.clone());
1705+ // Only ack a message we actually stored — a duplicate id
1706+ // was already acked when its first copy was ingested.
1707+ ack_targets.entry(sender.clone()).or_default().push(msg_id);
15841708 }
1585− ack_targets.entry(sender).or_default().push(msg_id);
15861709 }
15871710 Payload::DeliveryAck { ack_ids } => {
15881711 // Staleness check: drop a replayed ack whose ts predates the
@@ -1636,7 +1759,15 @@ impl Skrepka {
16361759 // peer that parked it in the future would freeze its own
16371760 // profile, since every honest update it sent afterwards would
16381761 // read as a stale replay of itself.
1639− if ts >= c.last_profile_ts {
1762+ //
1763+ // Strict `>` (not `>=`): a replayed profile at the *same* `ts`
1764+ // as the one we already stored is a re-delivery, not an update.
1765+ // `>=` would let a captured blob overwrite the cached profile
1766+ // with an identical one — harmless in the normal case, but it
1767+ // also means a peer who sends `ts = 0` (the default when the
1768+ // field is missing) can overwrite another `ts = 0` profile
1769+ // indefinitely.
1770+ if ts > c.last_profile_ts {
16401771 c.display_name = display_name;
16411772 c.bio = bio;
16421773 c.photo = photo;
@@ -1723,22 +1854,26 @@ fn trim_history(convo: &mut Vec<StoredMessage>) {
17231854 }
17241855}
17251856
1726−/// Decode a kv `get` result into a typed value, ignoring errors / absence.
1857+/// Decode a kv `get` result into a typed value.
17271858///
1728−/// `None` conflates three different things: the key is absent (a fresh install),
1729−/// the read failed, and the bytes are present but don't deserialize. The caller
1730−/// keeps its default in all three, and the next mutation writes that default back
1731−/// — so a blob we merely *failed to read* is destroyed rather than left alone.
1859+/// Returns `Ok(Some(value))` when the key exists and deserializes, `Ok(None)`
1860+/// when the key is absent (a fresh install), and `Err` when the read itself
1861+/// failed (e.g. the file is locked or corrupt). The caller MUST NOT overwrite
1862+/// a key with a default when the read failed — the data is still on disk, and
1863+/// writing the default back destroys it. An absent key is safe to default.
17321864///
1733−/// The mitigation is upstream: every persisted type is `#[serde(default)]` (see
1734−/// `model.rs`), so adding a field can't make an old blob unreadable, which is by
1735−/// far the likeliest way this fires in practice. A genuinely corrupt blob is
1736−/// still lost. Fixing that properly means distinguishing the three cases here and
1737−/// refusing to overwrite on the third — a change to every caller and to the error
1738−/// surface, not a drive-by.
1739−fn parse_kv<T: for<'de> Deserialize<'de>>(res: KvData) -> Option<T> {
1740− let bytes = res.ok().flatten()?;
1741− serde_json::from_slice(&bytes).ok()
1865+/// The old `parse_kv` conflated read errors with absence, which meant a
1866+/// background wake while the device was locked (where `completeUnlessOpen`
1867+/// refuses to open existing files) would load an empty model, and the next
1868+/// `persist_*` would write that emptiness back over the real data.
1869+fn parse_kv<T: for<'de> Deserialize<'de>>(res: KvData) -> Result<Option<T>, ()> {
1870+ match res {
1871+ Ok(Some(bytes)) => serde_json::from_slice(&bytes)
1872+ .map(Some)
1873+ .map_err(|_| ()),
1874+ Ok(None) => Ok(None),
1875+ Err(_) => Err(()),
1876+ }
17421877}
17431878
17441879#[cfg(test)]
@@ -1839,6 +1974,33 @@ mod tests {
18391974 .build()
18401975 }
18411976
1977+ /// The same, plus one header — the only thing a relay can say to us beyond a
1978+ /// status code, and `Retry-After` is the one we act on.
1979+ fn response_with_header(status: u16, name: &str, value: &str) -> crux_http::Response<Vec<u8>> {
1980+ let status = crux_http::http::StatusCode::try_from(status).unwrap();
1981+ crux_http::testing::ResponseBuilder::with_status(status)
1982+ .header(name, value)
1983+ .body(Vec::new())
1984+ .build()
1985+ }
1986+
1987+ /// The delays a command actually asks the shell to wait, in milliseconds.
1988+ /// Asserting on the model alone cannot see these — a backoff lives entirely in
1989+ /// the effect.
1990+ fn scheduled_delays(cmd: &mut Command<Effect, Event>) -> Vec<u64> {
1991+ cmd.effects()
1992+ .filter_map(|e| match e {
1993+ Effect::Time(req) => match req.operation {
1994+ crux_time::TimeRequest::NotifyAfter { duration, .. } => {
1995+ Some(Duration::from(duration).as_millis() as u64)
1996+ }
1997+ _ => None,
1998+ },
1999+ _ => None,
2000+ })
2001+ .collect()
2002+ }
2003+
18422004 /// One full send round-trip: flush the outbox, then answer with `status`.
18432005 fn send_and_answer(app: &Skrepka, m: &mut Model, status: u16) {
18442006 let _ = app.update(Event::StartFlush, m);
@@ -1977,6 +2139,71 @@ mod tests {
19772139 assert!(m.token.is_none(), "but the token is thrown away");
19782140 }
19792141
2142+ /// A 429 is the relay's load, not the message's fault. Charging the batch a
2143+ /// retry for it would let a busy relay drive perfectly good messages out of the
2144+ /// outbox — and the relay usually says exactly how long to wait, so wait that
2145+ /// long rather than guessing.
2146+ #[test]
2147+ fn a_429_waits_out_retry_after_without_spending_the_retry_budget() {
2148+ let app = Skrepka;
2149+ let mut m = with_identity();
2150+ m.token = Some("t".into());
2151+ m.outbox.push_back(queued(&peer_hex(9), "hi"));
2152+
2153+ let _ = app.update(Event::StartFlush, &mut m);
2154+ let mut cmd = app.update(
2155+ Event::SendResult(Ok(response_with_header(429, "retry-after", "2"))),
2156+ &mut m,
2157+ );
2158+
2159+ assert_eq!(m.outbox.len(), 1, "the message is still queued");
2160+ assert_eq!(m.outbox[0].retries, 0, "and unpunished — it did nothing wrong");
2161+ assert!(!m.flushing, "the guard is released so the retry can run");
2162+ assert_eq!(
2163+ scheduled_delays(&mut cmd),
2164+ vec![2_000],
2165+ "and we wait exactly as long as the relay asked"
2166+ );
2167+ }
2168+
2169+ /// Without the header there is nothing to honour, so fall back to our own
2170+ /// backoff rather than hammering the relay that just asked us to slow down.
2171+ #[test]
2172+ fn a_429_without_retry_after_falls_back_to_the_usual_backoff() {
2173+ let app = Skrepka;
2174+ let mut m = with_identity();
2175+ m.token = Some("t".into());
2176+ m.outbox.push_back(queued(&peer_hex(9), "hi"));
2177+
2178+ let _ = app.update(Event::StartFlush, &mut m);
2179+ let mut cmd = app.update(Event::SendResult(Ok(ok_response(429))), &mut m);
2180+
2181+ assert_eq!(m.outbox[0].retries, 0, "still not the message's fault");
2182+ assert_eq!(scheduled_delays(&mut cmd), vec![backoff_ms(m.poll_retries)]);
2183+ }
2184+
2185+ /// `Retry-After` is a number a relay chooses and we obey. Unbounded, one of
2186+ /// them parks the outbox for thirty years — with nothing on screen to say why.
2187+ #[test]
2188+ fn an_absurd_retry_after_is_clamped() {
2189+ let app = Skrepka;
2190+ let mut m = with_identity();
2191+ m.token = Some("t".into());
2192+ m.outbox.push_back(queued(&peer_hex(9), "hi"));
2193+
2194+ let _ = app.update(Event::StartFlush, &mut m);
2195+ let mut cmd = app.update(
2196+ Event::SendResult(Ok(response_with_header(429, "retry-after", "999999999"))),
2197+ &mut m,
2198+ );
2199+
2200+ assert_eq!(
2201+ scheduled_delays(&mut cmd),
2202+ vec![MAX_RETRY_AFTER_MS],
2203+ "we wait, but only as long as we are willing to"
2204+ );
2205+ }
2206+
19802207 // -----------------------------------------------------------------------
19812208 // outbox: batching
19822209 // -----------------------------------------------------------------------
@@ -2088,6 +2315,101 @@ mod tests {
20882315 assert!(body <= MAX_SEND_BATCH_BYTES, "the body stays inside the budget");
20892316 }
20902317
2318+ /// `MAX_SEND_BATCH_BYTES` is a *guess* at a cap the protocol never publishes, and
2319+ /// a relay configured below it answers `413`. Read as a transient failure, that
2320+ /// `413` rebuilds the identical batch, is rejected identically, and repeats until
2321+ /// every item has burned its retry budget — the outbox drains without a single
2322+ /// message being delivered. So a `413` measures the cap instead of fighting it.
2323+ #[test]
2324+ fn a_413_halves_the_batch_budget_and_retries_smaller() {
2325+ let app = Skrepka;
2326+ let mut m = with_identity();
2327+ m.token = Some("t".into());
2328+ let peer = peer_hex(9);
2329+ for i in 0..5 {
2330+ m.outbox.push_back(queued(&peer, &format!("m{i}")));
2331+ }
2332+
2333+ // Nothing is known about this relay yet, so all five go out at once.
2334+ let mut cmd = app.update(Event::StartFlush, &mut m);
2335+ let first = sent_batches(&mut cmd);
2336+ assert_eq!(first[0].len(), 5, "the whole run goes out on the first attempt");
2337+ assert!(m.send_batch_budget.is_none(), "and we have assumed nothing");
2338+ let sent_bytes: usize = first[0].iter().map(|e| e.encrypted_blob.len()).sum();
2339+
2340+ // ...and the relay says that was too big.
2341+ let mut cmd = app.update(Event::SendResult(Ok(ok_response(413))), &mut m);
2342+
2343+ assert_eq!(
2344+ m.send_batch_budget,
2345+ Some(sent_bytes / 2),
2346+ "the budget is halved against what actually failed, not against the constant"
2347+ );
2348+ assert_eq!(m.outbox.len(), 5, "a 413 delivered nothing, so nothing is popped");
2349+ assert!(
2350+ m.outbox.iter().all(|i| i.retries == 0),
2351+ "and the items are not charged for the relay being smaller than we guessed"
2352+ );
2353+
2354+ // The retry goes out at once, and it is smaller.
2355+ let second = sent_batches(&mut cmd);
2356+ assert_eq!(second.len(), 1, "retried immediately, not left for the next flush");
2357+ let shrunk = second[0].len();
2358+ assert!(shrunk < 5, "with a smaller batch, got {shrunk} items");
2359+ assert!(shrunk > 0, "but not an empty one");
2360+
2361+ // The relay takes the smaller batch, and exactly it is popped.
2362+ let _ = app.update(Event::SendResult(Ok(ok_response(200))), &mut m);
2363+ assert_eq!(m.outbox.len(), 5 - shrunk, "exactly the batch that landed is gone");
2364+ assert_eq!(
2365+ m.send_batch_budget,
2366+ Some(sent_bytes / 2),
2367+ "and what we learned about the relay survives the success — its cap did not grow"
2368+ );
2369+ }
2370+
2371+ /// A batch of one cannot be split, and `flush_next` takes the first item whatever
2372+ /// it weighs — so halving the budget around a lone oversized item would rebuild
2373+ /// the same one-item batch forever, at full speed, with no retry ever charged.
2374+ /// It has to fall through to the retry budget, which is what eventually drops it.
2375+ #[test]
2376+ fn a_413_on_a_single_item_ages_out_instead_of_spinning() {
2377+ let app = Skrepka;
2378+ let mut m = with_identity();
2379+ m.token = Some("t".into());
2380+ m.outbox.push_back(queued(&peer_hex(9), "bigger than this relay allows"));
2381+
2382+ for i in 1..MAX_OUTBOX_RETRIES {
2383+ send_and_answer(&app, &mut m, 413);
2384+ assert_eq!(m.outbox.len(), 1, "still queued after {i} rejection(s)");
2385+ assert_eq!(m.outbox[0].retries, i, "and charged for each of them");
2386+ }
2387+ assert!(
2388+ m.send_batch_budget.is_none(),
2389+ "an unsplittable batch teaches us nothing about the cap"
2390+ );
2391+
2392+ send_and_answer(&app, &mut m, 413);
2393+ assert!(m.outbox.is_empty(), "the item that can never be sent is dropped");
2394+ assert!(
2395+ m.error.is_some_and(|e| e.contains("gave up sending")),
2396+ "and the user is told, rather than the message quietly vanishing"
2397+ );
2398+ }
2399+
2400+ /// What we learned is about *that relay*. A new one is entitled to a bigger cap,
2401+ /// and it will never send the `413` that would correct a budget carried over.
2402+ #[test]
2403+ fn changing_relay_forgets_the_learned_batch_budget() {
2404+ let app = Skrepka;
2405+ let mut m = with_identity();
2406+ m.send_batch_budget = Some(1024);
2407+
2408+ let _ = app.update(Event::SetServerUrl("https://other.example".into()), &mut m);
2409+
2410+ assert!(m.send_batch_budget.is_none());
2411+ }
2412+
20912413 /// An unusable key at the head still has to be discarded, batching or not —
20922414 /// and the discard must not take the rest of the queue with it.
20932415 #[test]
@@ -2378,9 +2700,12 @@ mod tests {
23782700 };
23792701 let page = page_from(9, &m, &profile("Far"), i64::MAX);
23802702 let _ = app.ingest_poll(&mut m, page);
2381− assert!(m.contacts[&peer].last_profile_ts <= now_ms());
2703+ let first_ts = m.contacts[&peer].last_profile_ts;
2704+ assert!(first_ts <= now_ms());
23822705
2383− let page = page_from(9, &m, &profile("Now"), now_ms());
2706+ // A later profile at a strictly greater ts must still apply (the
2707+ // staleness check is strict `>`, not `>=`).
2708+ let page = page_from(9, &m, &profile("Now"), first_ts + 1);
23842709 let _ = app.ingest_poll(&mut m, page);
23852710 assert_eq!(
23862711 m.contacts[&peer].display_name, "Now",
@@ -3209,6 +3534,43 @@ mod tests {
32093534 assert_eq!(m.contacts[&peer].display_name, "Mallory");
32103535 }
32113536
3537+ /// A relay cannot forge mail from us — the sender identity is signed inside the
3538+ /// AEAD — but it can *echo back* a blob we sent through it, and one addressed to
3539+ /// ourselves decrypts perfectly well. Nothing downstream would notice: it would
3540+ /// file our own words as incoming, put our own key in our contact list (which
3541+ /// `AddContact` explicitly refuses), and queue a delivery ack addressed to us —
3542+ /// which flushes, comes back, and is acked again.
3543+ #[test]
3544+ fn a_message_from_ourselves_is_dropped() {
3545+ let app = Skrepka;
3546+ let mut m = with_identity();
3547+ let me = Identity::from_secret_bytes(m.secret_key.as_ref().unwrap()).unwrap();
3548+
3549+ let json = protocol::serialize_payload(&text("m1", "echo"), now_ms());
3550+ let blob = crate::crypto::encrypt(
3551+ &mut rand_core::OsRng,
3552+ &me,
3553+ &me.public_key(),
3554+ json.as_bytes(),
3555+ )
3556+ .unwrap();
3557+ let page = PollResp {
3558+ events: vec![PollEvent {
3559+ encrypted_blob: hex::encode(blob),
3560+ }],
3561+ cursor: 1,
3562+ };
3563+
3564+ let _ = app.ingest_poll(&mut m, page);
3565+
3566+ assert!(
3567+ !m.contacts.contains_key(&m.my_pubkey),
3568+ "we are not our own contact"
3569+ );
3570+ assert!(m.messages.is_empty(), "and our own words are not incoming mail");
3571+ assert!(m.outbox.is_empty(), "and we do not ack ourselves");
3572+ }
3573+
32123574 /// `polling` gates the whole long-poll loop and is cleared only by a
32133575 /// `PollResult`. If the shell ever drops the HTTP effect, the client goes
32143576 /// silent forever with no visible error.