@@ -22,7 +22,7 @@ use serde::{Deserialize, Serialize};
2222use crate::crypto::Identity;
2323use crate::model::{
2424 hex_to_ob, trunc_ob, Contact, ContactVM, MessageVM, OutboxItem, OwnProfile, ProfileVM,
25− Settings, StoredMessage, ViewModel,
25+ Settings, StoredMessage, ViewModel, MAX_MESSAGES_PER_PEER,
2626};
2727use crate::phonemic;
2828use crate::protocol::{self, Envelope, Payload};
@@ -33,6 +33,11 @@ type Time = crux_time::Time<Effect, Event>;
3333type HttpResult = crux_http::Result<crux_http::Response<Vec<u8>>>;
3434type KvData = Result<Option<Vec<u8>>, crux_kv::error::KeyValueError>;
3535
36+/// How far into the future an incoming, sender-chosen `ts` may sit before it is
37+/// clamped to "now" — enough slack for honest clock skew, not enough to let a
38+/// peer park itself at the top of the conversation list.
39+const MAX_FUTURE_SKEW_MS: i64 = 60_000;
40+
3641// kv keys
3742const K_SETTINGS: &str = "settings";
3843const K_PROFILE: &str = "profile";
@@ -383,7 +388,15 @@ impl App for Skrepka {
383388 }
384389 render()
385390 }
386− Event::Saved(_) => Command::done(),
391+ Event::Saved(res) => {
392+ // A dropped write silently loses data (a message, a contact, the
393+ // cursor), so surface it rather than swallowing the result.
394+ if let Err(e) = res {
395+ model.error = Some(format!("could not save data: {e}"));
396+ return render();
397+ }
398+ Command::done()
399+ }
387400
388401 // ---------------- navigation ----------------
389402 Event::ShowConversations => {
@@ -491,6 +504,12 @@ impl App for Skrepka {
491504 photo: model.profile.photo.clone(),
492505 };
493506 // Broadcast the new profile to every contact.
507+ //
508+ // Each item carries its own copy of the serialized payload, so a
509+ // profile with a photo is duplicated (base64 and all) once per
510+ // contact, in memory and in the `outbox` kv blob. Acceptable at
511+ // today's contact counts; the fix is to intern the payload and
512+ // have `OutboxItem` reference it by id.
494513 for peer in model.contacts.keys().cloned().collect::<Vec<_>>() {
495514 model.outbox.push_back(OutboxItem {
496515 recipient: peer,
@@ -795,6 +814,7 @@ impl Skrepka {
795814 outgoing: true,
796815 delivered: false,
797816 });
817+ trim_history(convo);
798818
799819 // Auto-create a bare contact if messaging a brand-new key.
800820 model
@@ -839,14 +859,24 @@ impl Skrepka {
839859 let (Some(id), Some(token)) = (model.identity(), model.token.clone()) else {
840860 return Command::done();
841861 };
842− let Ok(recipient) = hex::decode(&item.recipient) else {
843− // Bad recipient — drop it.
862+ // An unusable recipient key can never be encrypted to, so the item has to
863+ // be dropped or it blocks the head of the outbox forever — but say so,
864+ // rather than letting the message vanish.
865+ let drop_item = |model: &mut Model| {
866+ model.error = Some(format!(
867+ "cannot send to {}: invalid key",
868+ trunc_ob(&item.recipient)
869+ ));
844870 model.outbox.pop_front();
871+ };
872+
873+ let Ok(recipient) = hex::decode(&item.recipient) else {
874+ drop_item(model);
845875 return self.persist_outbox(model).and(self.flush_next(model));
846876 };
847877 let mut recip = [0u8; 32];
848878 if recipient.len() != 32 {
849− model.outbox.pop_front();
879+ drop_item(model);
850880 return self.persist_outbox(model).and(self.flush_next(model));
851881 }
852882 recip.copy_from_slice(&recipient);
@@ -854,9 +884,11 @@ impl Skrepka {
854884 let mut rng = rand_core::OsRng;
855885 let blob = match crate::crypto::encrypt(&mut rng, &id, &recip, item.envelope_json.as_bytes())
856886 {
887+ // Reachable for a 32-byte key that isn't a curve point: `AddContact`
888+ // rejects those now, but an older contact list may still hold one.
857889 Ok(b) => b,
858890 Err(_) => {
859− model.outbox.pop_front();
891+ drop_item(model);
860892 return self.persist_outbox(model).and(self.flush_next(model));
861893 }
862894 };
@@ -893,6 +925,7 @@ impl Skrepka {
893925 return Command::done();
894926 };
895927
928+ let now = now_ms();
896929 let mut touched_convos: Vec<String> = Vec::new();
897930 let mut contacts_dirty = false;
898931 // sender hex -> message ids needing acks
@@ -909,13 +942,25 @@ impl Skrepka {
909942 continue;
910943 };
911944 let sender = dec.sender_hex;
945+ // `ts` is chosen by the sender and drives conversation ordering,
946+ // in-chat ordering, and the displayed time. Bound it to the near
947+ // future so a peer cannot pin itself to the top of the list forever
948+ // with `ts = i64::MAX`.
949+ let ts = parsed.ts.min(now.saturating_add(MAX_FUTURE_SKEW_MS));
950+ let blocked = model.contacts.get(&sender).is_some_and(|c| c.blocked);
912951 match parsed.payload {
913952 Payload::Text { id: msg_id, body } => {
953+ // A blocked peer's messages are not displayed (PROTOCOL.md
954+ // §4), so don't store them — and don't ack them either: an
955+ // ack would tell them we are online and reading.
956+ if blocked {
957+ continue;
958+ }
914959 // auto-create contact
915960 if !model.contacts.contains_key(&sender) {
916961 model
917962 .contacts
918− .insert(sender.clone(), Contact::new(sender.clone(), String::new(), parsed.ts));
963+ .insert(sender.clone(), Contact::new(sender.clone(), String::new(), ts));
919964 contacts_dirty = true;
920965 }
921966 let convo = model.messages.entry(sender.clone()).or_default();
@@ -924,11 +969,12 @@ impl Skrepka {
924969 convo.push(StoredMessage {
925970 id: msg_id.clone(),
926971 body,
927− ts: parsed.ts,
972+ ts,
928973 outgoing: false,
929974 delivered: false,
930975 });
931976 convo.sort_by_key(|m| m.ts);
977+ trim_history(convo);
932978 if !touched_convos.contains(&sender) {
933979 touched_convos.push(sender.clone());
934980 }
@@ -957,13 +1003,15 @@ impl Skrepka {
9571003 let c = model
9581004 .contacts
9591005 .entry(sender.clone())
960− .or_insert_with(|| Contact::new(sender.clone(), String::new(), parsed.ts));
961− // Ignore stale profile replays (PROTOCOL.md §4).
962− if parsed.ts >= c.last_profile_ts {
1006+ .or_insert_with(|| Contact::new(sender.clone(), String::new(), ts));
1007+ // Ignore stale profile replays (PROTOCOL.md §4). The clamped
1008+ // `ts` is what gets recorded, so a peer cannot park
1009+ // `last_profile_ts` at `i64::MAX` and freeze its own profile.
1010+ if ts >= c.last_profile_ts {
9631011 c.display_name = display_name;
9641012 c.bio = bio;
9651013 c.photo = photo;
966− c.last_profile_ts = parsed.ts;
1014+ c.last_profile_ts = ts;
9671015 contacts_dirty = true;
9681016 }
9691017 }
@@ -998,6 +1046,16 @@ fn backoff_ms(retries: u32) -> u64 {
9981046 (3000u64.saturating_mul(1u64 << retries.min(4))).min(30_000)
9991047}
10001048
1049+/// Age out local history: keep only the most recent `MAX_MESSAGES_PER_PEER`
1050+/// messages of a conversation (PROTOCOL.md §9). Callers keep `convo` sorted by
1051+/// `ts`, so the oldest messages are at the front. The subsequent kv write
1052+/// persists the trimmed list, bounding the `messages:<peer>` blob too.
1053+fn trim_history(convo: &mut Vec<StoredMessage>) {
1054+ if let Some(excess) = convo.len().checked_sub(MAX_MESSAGES_PER_PEER) {
1055+ convo.drain(..excess);
1056+ }
1057+}
1058+
10011059/// Decode a kv `get` result into a typed value, ignoring errors / absence.
10021060fn parse_kv<T: for<'de> Deserialize<'de>>(res: KvData) -> Option<T> {
10031061 let bytes = res.ok().flatten()?;
@@ -1022,6 +1080,199 @@ mod tests {
10221080 Identity::from_seed(&[seed; 32]).public_key_hex()
10231081 }
10241082
1083+ /// A poll page carrying one payload, encrypted from `sender_seed` to `me`.
1084+ fn page_from(sender_seed: u8, me: &Model, payload: &Payload, ts: i64) -> PollResp {
1085+ let sender = Identity::from_seed(&[sender_seed; 32]);
1086+ let recipient = Identity::from_secret_bytes(me.secret_key.as_ref().unwrap()).unwrap();
1087+ let json = protocol::serialize_payload(payload, ts);
1088+ let blob = crate::crypto::encrypt(
1089+ &mut rand_core::OsRng,
1090+ &sender,
1091+ &recipient.public_key(),
1092+ json.as_bytes(),
1093+ )
1094+ .unwrap();
1095+ PollResp {
1096+ events: vec![PollEvent {
1097+ encrypted_blob: hex::encode(blob),
1098+ }],
1099+ cursor: 1,
1100+ }
1101+ }
1102+
1103+ fn text(id: &str, body: &str) -> Payload {
1104+ Payload::Text {
1105+ id: id.to_string(),
1106+ body: body.to_string(),
1107+ }
1108+ }
1109+
1110+ /// PROTOCOL.md §4: a blocked peer's messages are not shown — and acking them
1111+ /// would tell the blocked peer we are online and reading.
1112+ #[test]
1113+ fn a_blocked_senders_message_is_neither_stored_nor_acked() {
1114+ let app = Skrepka;
1115+ let mut m = with_identity();
1116+ let peer = peer_hex(9);
1117+ let mut contact = Contact::new(peer.clone(), "Bob".into(), 0);
1118+ contact.blocked = true;
1119+ m.contacts.insert(peer.clone(), contact);
1120+
1121+ let page = page_from(9, &m, &text("m1", "hi"), now_ms());
1122+ let _ = app.ingest_poll(&mut m, page);
1123+
1124+ assert!(
1125+ m.messages.get(&peer).is_none_or(Vec::is_empty),
1126+ "a blocked peer's message must not be stored"
1127+ );
1128+ assert!(m.outbox.is_empty(), "no delivery.ack goes back to a blocked peer");
1129+ // ...and nothing to render in the chat, even if it is opened.
1130+ m.active_peer = Some(peer);
1131+ m.page = Page::Chat;
1132+ assert!(app.view(&m).messages.is_empty());
1133+ }
1134+
1135+ /// The same page, unblocked: the control that proves the gate is the block
1136+ /// flag and not something else in the ingest path.
1137+ fn ingest_one_text_from_an_unblocked_peer() -> (Model, String) {
1138+ let app = Skrepka;
1139+ let mut m = with_identity();
1140+ let peer = peer_hex(9);
1141+ let page = page_from(9, &m, &text("m1", "hi"), now_ms());
1142+ let _ = app.ingest_poll(&mut m, page);
1143+ (m, peer)
1144+ }
1145+
1146+ #[test]
1147+ fn an_unblocked_senders_message_is_stored_and_acked() {
1148+ let (m, peer) = ingest_one_text_from_an_unblocked_peer();
1149+ assert_eq!(m.messages[&peer].len(), 1);
1150+ assert_eq!(m.messages[&peer][0].body, "hi");
1151+ assert_eq!(m.outbox.len(), 1, "one delivery.ack is queued");
1152+ assert_eq!(m.outbox[0].recipient, peer);
1153+ }
1154+
1155+ /// A peer sending `ts = i64::MAX` used to pin itself to the top of the
1156+ /// conversation list forever, and render with an absurd timestamp.
1157+ #[test]
1158+ fn a_far_future_timestamp_is_clamped_to_now() {
1159+ let app = Skrepka;
1160+ let mut m = with_identity();
1161+ let peer = peer_hex(9);
1162+
1163+ let page = page_from(9, &m, &text("m1", "hi"), i64::MAX);
1164+ let before = now_ms();
1165+ let _ = app.ingest_poll(&mut m, page);
1166+
1167+ let ts = m.messages[&peer][0].ts;
1168+ assert!(
1169+ ts <= now_ms() + MAX_FUTURE_SKEW_MS,
1170+ "ts {ts} must be clamped into the skew window"
1171+ );
1172+ assert!(ts >= before, "and clamped to now, not to some past value");
1173+ // The clamped value is what the shell sorts and renders by.
1174+ assert_eq!(app.view(&m).contacts[0].last_ts, ts);
1175+ }
1176+
1177+ /// Honest clock skew inside the tolerance is preserved as-is.
1178+ #[test]
1179+ fn a_slightly_future_timestamp_is_kept() {
1180+ let app = Skrepka;
1181+ let mut m = with_identity();
1182+ let peer = peer_hex(9);
1183+ let ts = now_ms() + 5_000;
1184+
1185+ let page = page_from(9, &m, &text("m1", "hi"), ts);
1186+ let _ = app.ingest_poll(&mut m, page);
1187+
1188+ assert_eq!(m.messages[&peer][0].ts, ts);
1189+ }
1190+
1191+ #[test]
1192+ fn message_history_is_capped_at_the_cutoff() {
1193+ let app = Skrepka;
1194+ let mut m = with_identity();
1195+ let peer = peer_hex(9);
1196+
1197+ let convo = m.messages.entry(peer.clone()).or_default();
1198+ for i in 0..MAX_MESSAGES_PER_PEER {
1199+ convo.push(StoredMessage {
1200+ id: format!("old-{i}"),
1201+ body: "x".into(),
1202+ ts: i as i64,
1203+ outgoing: false,
1204+ delivered: false,
1205+ });
1206+ }
1207+
1208+ let page = page_from(9, &m, &text("new", "newest"), now_ms());
1209+ let _ = app.ingest_poll(&mut m, page);
1210+
1211+ let convo = &m.messages[&peer];
1212+ assert_eq!(convo.len(), MAX_MESSAGES_PER_PEER, "the cap holds");
1213+ assert_eq!(convo.last().unwrap().id, "new", "the newest is kept");
1214+ assert_eq!(convo.first().unwrap().id, "old-1", "the oldest is aged out");
1215+ }
1216+
1217+ /// An off-curve key can't be encrypted to, so the send is dropped — but the
1218+ /// user has to be told, rather than watching the message vanish.
1219+ #[test]
1220+ fn a_send_to_an_unusable_key_reports_an_error() {
1221+ let app = Skrepka;
1222+ let mut m = with_identity();
1223+ m.token = Some("t".into());
1224+ let bad = hex::encode([0xabu8; 32]); // 32 bytes, not a curve point
1225+ m.outbox.push_back(OutboxItem {
1226+ recipient: bad,
1227+ envelope_json: "{}".into(),
1228+ });
1229+
1230+ let _ = app.update(Event::StartFlush, &mut m);
1231+
1232+ assert!(m.outbox.is_empty(), "the unsendable item does not wedge the outbox");
1233+ assert!(!m.flushing);
1234+ let err = m.error.expect("the failure is surfaced");
1235+ assert!(err.contains("invalid key"), "got: {err}");
1236+ }
1237+
1238+ #[test]
1239+ fn add_contact_rejects_a_key_that_is_not_a_curve_point() {
1240+ let app = Skrepka;
1241+ let mut m = with_identity();
1242+ let bad = hex::encode([0xabu8; 32]);
1243+
1244+ let _ = app.update(
1245+ Event::AddContact {
1246+ input: bad.clone(),
1247+ nickname: "Mallory".into(),
1248+ },
1249+ &mut m,
1250+ );
1251+
1252+ assert!(!m.contacts.contains_key(&bad), "an unsendable contact is not stored");
1253+ assert!(m.error.is_some());
1254+ }
1255+
1256+ /// A dropped kv write silently loses data; it must not be swallowed.
1257+ #[test]
1258+ fn a_failed_save_surfaces_an_error() {
1259+ let app = Skrepka;
1260+ let mut m = with_identity();
1261+
1262+ let _ = app.update(
1263+ Event::Saved(Err(crux_kv::error::KeyValueError::Io {
1264+ message: "no space left on device".into(),
1265+ })),
1266+ &mut m,
1267+ );
1268+ assert!(m.error.is_some(), "a failed write is reported");
1269+
1270+ // A successful write stays quiet.
1271+ m.error = None;
1272+ let _ = app.update(Event::Saved(Ok(None)), &mut m);
1273+ assert!(m.error.is_none());
1274+ }
1275+
10251276 #[test]
10261277 fn server_host_strips_scheme_port_and_case() {
10271278 assert_eq!(server_host("https://Relay.Example.com:8443/x"), "relay.example.com");