From 5f4eba1815770cd4282c81b39f08ec71984a874d Mon Sep 17 00:00:00 2001 From: Mollusk Date: Sat, 18 Jul 2026 19:03:25 -0400 Subject: [PATCH] chat: honest local send status + sender-side pacing (Phase 5) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Closes the final phase of docs/chat-hardening-plan.md. Two problems: a locally echoed message always looked sent even when the core had no active session or the gossip broadcast failed; and a fast burst could broadcast 'successfully' yet be silently dropped by every receiver's per-author rate bucket (8 burst, then 1/s) with no sender feedback. Send status: CoreCommand::SendChat/SendChatFile carry a local-only id (never on the wire); the core replies with UiEvent::ChatSendResult after the gossip broadcast succeeds or fails, and a no-active-session is now an explicit failure rather than a silent no-op. gossip send_chat, which previously returned Ok on a missing sender/topic or an encode failure, now returns Err. ChatEntry gains local_send: Option; failed sends render a red 'Not sent — {reason} [Retry]' line, Broadcast/Pending render nothing (there are no delivery receipts, so silence is the honest success state). Sender-side pacing (new src/app/sendqueue.rs): sends past the burst queue locally as 'queued…' and trickle out at the receivers' sustained rate, so nothing is lost and typing is never blocked (user chose queue-and-trickle over input throttling). The pacer reuses the gossip gate's own TokenBucket + per-author constants (now pub(crate)) so the two sides of the policy can't drift. A 250ms drain subscription runs only while the queue is non-empty. Retry re-dispatches the retained payload; re-serving the same attachment id replaces the ServeStore entry rather than double-counting bytes. The pacer and monotonic send-id counter survive a room reset (receivers' buckets persist; ids never alias a late result); queue and retry payloads are cleared. 582 lib tests (+11: 4 pacer/queue seam, 7 app-level transition/retry/reset); all-targets green, clippy -D warnings clean, fmt clean, smoke launch OK. No wire change (GOSSIP_PROTO stays 5). Tests-green-only — the two owed two-machine field-test items are logged in the plan. Co-Authored-By: Claude Fable 5 --- docs/chat-hardening-plan.md | 112 +++++++--- src/app/mod.rs | 417 +++++++++++++++++++++++++++++++++++- src/app/sendqueue.rs | 129 +++++++++++ src/core/messages.rs | 41 +++- src/core/mod.rs | 47 +++- src/network/gossip.rs | 56 ++--- 6 files changed, 728 insertions(+), 74 deletions(-) create mode 100644 src/app/sendqueue.rs diff --git a/docs/chat-hardening-plan.md b/docs/chat-hardening-plan.md index cd09693..f309e7d 100644 --- a/docs/chat-hardening-plan.md +++ b/docs/chat-hardening-plan.md @@ -1,18 +1,23 @@ # Chat hardening — ephemeral implementation plan -**Status (2026-07-18):** Phases 1–4 COMPLETE. Phase 1 = shared text policy in -`src/sanitize.rs`, ceilings enforced at UI input, sign point, and gossip ingress. -Phase 2 = roster-bound authorship (`src/core/chatroster.rs`), replay dedup + rate -limits (`ChatIngressGate` in `src/network/gossip.rs`). Phase 3 = attachment -cache/serve-store budgets, downscaled previews, auto-fetch byte/request budgets -(`src/core/fetchbudget.rs`), exact transfers, bounded local reads. Phase 4 = -parsed-URL link policy (`is_safe_web_url`/`link_ranges` in `src/sanitize.rs`, -`url` crate), 8-link cap, cached link ranges in `ChatEntry`, 512 KiB history -text budget, chat-body bidi-override strip (closes S14). All gates green each -phase. Phase 5 not started. This is a temporary scope contract for hardening -the existing room chat. Update the checkboxes and decision log as work lands, -then delete this file when the work is complete. Do not add link previews as -part of this effort. +**Status (2026-07-18):** Phases 1–5 COMPLETE (all plan phases done). Phase 1 = +shared text policy in `src/sanitize.rs`, ceilings enforced at UI input, sign +point, and gossip ingress. Phase 2 = roster-bound authorship +(`src/core/chatroster.rs`), replay dedup + rate limits (`ChatIngressGate` in +`src/network/gossip.rs`). Phase 3 = attachment cache/serve-store budgets, +downscaled previews, auto-fetch byte/request budgets (`src/core/fetchbudget.rs`), +exact transfers, bounded local reads. Phase 4 = parsed-URL link policy +(`is_safe_web_url`/`link_ranges` in `src/sanitize.rs`, `url` crate), 8-link cap, +cached link ranges in `ChatEntry`, 512 KiB history text budget, chat-body +bidi-override strip (closes S14). Phase 5 = honest local send status +(`CoreCommand::SendChat`/`SendChatFile` carry a local id, `UiEvent::ChatSendResult`, +`SendStatus` on own echoes) PLUS sender-side pacing (`src/app/sendqueue.rs` +mirrors the receivers' per-author budget so fast bursts trickle instead of being +silently dropped downstream). All gates green each phase. This is a temporary +scope contract for hardening the existing room chat; with every phase complete +and the two-machine field test done, delete this file (see the completion note +at the end). The two-machine field-test section below is still owed before that +deletion. Do not add link previews as part of this effort. ## Goal @@ -294,24 +299,37 @@ small spans an unnecessary UI/launcher surface. **Target:** never present a locally echoed message as successfully broadcast when the core rejected it or gossip broadcast failed. -- [ ] Add a local-only message id and `Pending`/`Broadcast`/`Failed` state to local - chat entries. Do not put this id or state on the wire. -- [ ] Carry the local id through `CoreCommand::SendChat`/`SendChatFile` and return a +- [x] Add a local-only message id and `Pending`/`Broadcast`/`Failed` state to local + chat entries. Do not put this id or state on the wire. (`ChatEntry.local_send: + Option`; `SendStatus` also has `Queued` for the paced-but-not-yet-sent + state — see the pacing decision-log entry.) +- [x] Carry the local id through `CoreCommand::SendChat`/`SendChatFile` and return a `UiEvent` result after the local gossip broadcast call succeeds or fails. -- [ ] If the core is not in an active session, return failure instead of silently - doing nothing. -- [ ] Show failure compactly with a retry action. A successful local broadcast must + (`SendChat`/`SendChatFile` gained `local_id`; new `UiEvent::ChatSendResult { local_id, + error }`.) +- [x] If the core is not in an active session, return failure instead of silently + doing nothing. (`send_chat` now `Err`s on missing sender/topic and on encode + failure; the core arm maps no-session to a `ChatSendResult` error.) +- [x] Show failure compactly with a retry action. A successful local broadcast must not be labeled “delivered” or “read”; PeerSpeak has no peer acknowledgements. -- [ ] Retry creates one new signed broadcast while retaining replay correctness and - attachment serving state. + (Failed → red "⚠ Not sent — {reason} [Retry]" line; Broadcast/Pending render + nothing — silence is the honest success state.) +- [x] Retry creates one new signed broadcast while retaining replay correctness and + attachment serving state. (`RetryChatSend(id)` re-dispatches the retained + `PendingSend`; re-serving the same attachment id REPLACES the `ServeStore` + entry, never double-counts — see `serve_store_replacement_accounting_and_remove_clear`.) ### Phase 5 tests -- [ ] Local echo starts pending, becomes broadcast on success, and becomes failed - on no-session/channel/gossip error. -- [ ] Results update only the matching local entry, including after history - eviction or room reset. -- [ ] Retry does not duplicate served bytes or mutate an unrelated entry. +- [x] Local echo starts pending, becomes broadcast on success, and becomes failed + on no-session/channel/gossip error. (`send_status_pending_then_broadcast_on_success`, + `send_status_failed_keeps_payload_for_retry`.) +- [x] Results update only the matching local entry, including after history + eviction or room reset. (`send_result_updates_only_the_matching_entry`, + `send_result_after_eviction_drops_orphan_payload`, `send_result_after_room_reset_is_a_noop`.) +- [x] Retry does not duplicate served bytes or mutate an unrelated entry. + (`retry_redispatches_only_the_targeted_send`; served-byte dedup = + `serve_store_replacement_accounting_and_remove_clear` in `files.rs`.) ## Compatibility and versioning @@ -359,6 +377,11 @@ it; do not make ordinary unit tests depend on external network access. rest as selectable plain text, with nothing dropped. - [ ] A message attempting bidi-override display spoofing renders in send order (the override characters are stripped, emoji/joining-script text intact). +- [ ] Send a fast burst (>8 messages in a second): all arrive at the peer in + order, none silently lost; the sender sees "queued…" on the overflow that + then clears as each goes out. +- [ ] Send with no active session (or a failing broadcast): the message shows + "⚠ Not sent" with a Retry, and Retry resends it once when connectivity is back. ## Completion criteria @@ -522,3 +545,40 @@ The plan is complete when: an open Save/Play on an evicted line keeps its bytes-in-hand (the save dialog falls back to the generic "download" name). The just-pushed entry is never evicted; a single message's 8 KiB ceiling cannot exceed the budget. +- **2026-07-18 (Phase 5):** Sender-side PACING was added to Phase 5's scope + (originally receiver-status only). The Phase 2 decision log deferred the + "apply the same local submit policy to accidental rapid Enter" item to pair + with Phase 5, and honest status alone would still let a fast burst broadcast + successfully yet be silently dropped by every receiver's per-author bucket + (8 burst, then 1/s) with no sender feedback. The user chose "queue and + trickle" over "throttle input": sends past the burst queue locally as + `SendStatus::Queued` ("queued…") and release at the receivers' sustained + rate, so nothing is lost and typing is never blocked. +- **2026-07-18 (Phase 5):** The pacer (`src/app/sendqueue.rs`) reuses the + gossip gate's OWN `TokenBucket` + `CHAT_AUTHOR_BURST`/`CHAT_AUTHOR_REFILL_PER_MS` + (made `pub(crate)`), so the two sides of the rate policy are one definition + and cannot drift. It mirrors only the PER-AUTHOR budget, not the room-wide + one — we cannot know other members' send rates, and the per-author bucket is + the one guaranteed to apply to us at every receiver. +- **2026-07-18 (Phase 5):** Send status renders as a line UNDER the message + (user pick over an inline suffix glyph); `Broadcast` and the transient + `Pending` show nothing because PeerSpeak has no delivery/read receipts, so an + unadorned message IS the honest "handed to the swarm" state. Only `Queued` + and `Failed` (with Retry) are surfaced. +- **2026-07-18 (Phase 5):** The pacer and the monotonic send-id counter + deliberately SURVIVE a room reset while the queue and retry payloads are + cleared: receivers' per-author buckets persist across our rejoin (so the + pacer should not refill to full), and never-reused ids keep a late + `ChatSendResult` from a pre-reset send from aliasing a new entry — verified by + `send_result_after_room_reset_is_a_noop`. +- **2026-07-18 (Phase 5):** The pacer clock is `Instant`-based + (`AppState.send_clock`), not wall-clock, so a system time jump can neither + rewind nor fast-forward the send budget. + +## Completion + +All five phases are implemented and every gate is green. Per the scope-contract +note at the top, this file should be DELETED once the owed two-machine field +test (the checklist below) has been run — that deletion is a separate, +user-gated step, not part of the Phase 5 commit. Until then the plan stays as +the record of what shipped and what remains to verify on real hardware. diff --git a/src/app/mod.rs b/src/app/mod.rs index b81edcf..81df11c 100644 --- a/src/app/mod.rs +++ b/src/app/mod.rs @@ -20,6 +20,9 @@ use crate::theme::{AppTheme, Palette}; use crate::widget::context_input::{context_input, locked_value}; use crate::widget::selectable_text::selectable_rich_text; +mod sendqueue; +use sendqueue::{LocalSend, SendPacer, SendStatus}; + use iced::widget::canvas::{Action, Frame, Geometry, Path, Program}; use iced::widget::text_input; use iced::widget::{ @@ -324,6 +327,22 @@ struct ChatEntry { /// cached ranges instead of rescanning and re-validating every message. /// At most [`crate::sanitize::CHAT_MSG_MAX_LINKS`] entries. links: Vec>, + /// Local-only send bookkeeping (chat-hardening Phase 5): `Some` exactly on + /// our own locally echoed messages, `None` on peers'. Never on the wire. + local_send: Option, +} + +/// Everything needed to (re)dispatch one local send (chat-hardening Phase 5). +/// Held in `AppState.send_payloads` from submit until the send is Broadcast, so +/// a queued release or a Retry rebuilds the exact same `CoreCommand`. +#[derive(Debug, Clone)] +enum PendingSend { + Text(String), + File { + text: String, + attachment: crate::files::ChatAttachment, + data: Arc>, + }, } /// Fetch state of a chat attachment's bytes (session-only). Absence from the @@ -824,6 +843,11 @@ pub enum AppMessage { AudioTick, /// Send the current chat input line (Enter or the Send button). ChatSubmit, + /// Drain cadence for the outbound chat queue while it is non-empty + /// (Phase 5 pacing): dispatches queued sends as pacer tokens refill. + ChatSendTick, + /// Re-dispatch a Failed local send (the "Retry" button), by local send id. + RetryChatSend(u64), /// Open a clicked chat link in the system browser (A13). OpenUrl(String), /// A room divider was dragged by the given pixel delta along its drag axis @@ -1045,6 +1069,22 @@ pub struct AppState { pending_saves: HashSet, /// Clip ids waiting for the existing attachment fetch path to return bytes. pending_plays: HashSet, + /// Monotonic id source for local chat sends (Phase 5). Never reset within a + /// run, so a stale result after a room reset can't alias a new entry. + next_send_id: u64, + /// Monotonic clock base for the send pacer (wall clocks can jump). + send_clock: std::time::Instant, + /// Sender-side pacer mirroring the receivers' per-author chat budget. + send_pacer: SendPacer, + /// Local send ids waiting for a pacer token, oldest first ("queued…"). + send_queue: VecDeque, + /// Dispatch payload for each in-flight local send, keyed by local id. Holds + /// everything needed to (re)dispatch: it backs both queued release and the + /// Retry button. For file sends the `Arc` is one clone of the same + /// allocation the cache/serve store share. Retained from submit until the + /// send is Broadcast (or the room resets); a Failed send keeps its payload + /// so Retry can re-dispatch without re-reading the file. + send_payloads: HashMap, /// Filename-hinted audio whose fetched bytes or decoder validation failed; /// these entries fall back to the normal file chip. invalid_audio: HashSet, @@ -1201,6 +1241,12 @@ impl AppState { self.conn_stats.clear(); self.locally_muted.clear(); self.chat_messages.clear(); + // Unsent queue + retry bytes die with the room's transcript. The pacer + // and id counter deliberately survive: receivers' per-author buckets + // persist across our rejoin, and monotonic ids keep any in-flight + // result from aliasing a post-reset entry. + self.send_queue.clear(); + self.send_payloads.clear(); self.chat_input.clear(); self.attachments.clear(); self.image_lightbox = None; @@ -1225,6 +1271,98 @@ impl AppState { self.clock_skew_warning = None; } + /// Monotonic milliseconds for the send pacer. `Instant`-based so a wall-clock + /// jump can never rewind or fast-forward the pacing budget. + fn send_now_ms(&self) -> u64 { + self.send_clock.elapsed().as_millis() as u64 + } + + /// Point the most recent local echo with this id at a new send status. Returns + /// whether such an entry still exists (it may have been evicted from history or + /// dropped by a room reset) so callers can release orphaned payloads. + fn set_send_status(&mut self, id: u64, status: SendStatus) -> bool { + if let Some(entry) = self + .chat_messages + .iter_mut() + .rev() + .find(|m| m.local_send.as_ref().is_some_and(|s| s.id == id)) + { + if let Some(ls) = entry.local_send.as_mut() { + ls.status = status; + } + true + } else { + false + } + } + + /// Hand one payload to the core, marking the entry Pending on success or + /// Failed if the core loop is gone (a dead reliable channel). The payload is + /// retained either way — a Retry can still re-dispatch after a dead-core + /// failure once the loop is back. + fn dispatch_send(&mut self, id: u64, payload: &PendingSend) { + let ok = match payload { + PendingSend::Text(text) => self.controller.send(CoreCommand::SendChat { + local_id: id, + text: text.clone(), + }), + PendingSend::File { + text, + attachment, + data, + } => self.controller.send(CoreCommand::SendChatFile { + local_id: id, + text: text.clone(), + attachment: attachment.clone(), + data: data.clone(), + }), + }; + self.set_send_status( + id, + if ok { + SendStatus::Pending + } else { + SendStatus::Failed("core unavailable".to_string()) + }, + ); + } + + /// Land a core `ChatSendResult` on the matching local echo (Phase 5). + /// Success (`error = None`) means our signed frame reached the swarm — NOT a + /// delivery receipt — and the payload is done. A failure keeps the payload + /// so Retry can re-dispatch, unless the entry is already gone (history + /// eviction / room reset), in which case the payload is dropped so its map + /// can't leak. Either way an id with no matching entry is a harmless no-op. + fn apply_send_result(&mut self, local_id: u64, error: Option) { + match error { + None => { + self.set_send_status(local_id, SendStatus::Broadcast); + self.send_payloads.remove(&local_id); + } + Some(e) => { + if !self.set_send_status(local_id, SendStatus::Failed(e)) { + self.send_payloads.remove(&local_id); + } + } + } + } + + /// Decide a fresh (or retried) send's fate: dispatch immediately only when + /// nothing is already queued AND the pacer — mirroring the receivers' + /// per-author budget — grants a token now; otherwise queue it as "queued…" + /// so it can never overtake an earlier message or exceed what receivers + /// admit. The payload is stored for release/retry. + fn submit_send(&mut self, id: u64, payload: PendingSend) { + self.send_payloads.insert(id, payload.clone()); + let now = self.send_now_ms(); + if self.send_queue.is_empty() && self.send_pacer.try_send(now) { + self.dispatch_send(id, &payload); + } else { + self.set_send_status(id, SendStatus::Queued); + self.send_queue.push_back(id); + } + } + fn custom_sound_path(&self, sound: Sound) -> &str { let opt = match sound { Sound::SelfJoin => &self.config.custom_sound_self_join, @@ -1376,6 +1514,11 @@ impl Default for AppState { image_lightbox: None, pending_saves: HashSet::new(), pending_plays: HashSet::new(), + next_send_id: 0, + send_clock: std::time::Instant::now(), + send_pacer: SendPacer::new(0), + send_queue: VecDeque::new(), + send_payloads: HashMap::new(), invalid_audio: HashSet::new(), clip_player, clip_status, @@ -1656,12 +1799,22 @@ fn subscription(state: &AppState) -> Subscription { } else { Subscription::none() }; + // Drain the outbound chat queue while it has waiting messages (Phase 5 + // pacing). A queued message is released only as the pacer refills (~1/s), so + // a 250ms poll is responsive without busy-looping; it stops entirely once + // the queue empties. + let send_queue_sub = if state.send_queue.is_empty() { + Subscription::none() + } else { + iced::time::every(std::time::Duration::from_millis(250)).map(|_| AppMessage::ChatSendTick) + }; Subscription::batch(vec![ core_sub, event_sub, audio_sub, clock_skew_sub, rescan_label_sub, + send_queue_sub, ]) } @@ -2141,6 +2294,15 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { state.recording_started = None; state.status_message = format!("Saved recording → {path}"); } + UiEvent::ChatSendResult { local_id, error } => { + // Land the honest outcome on the matching local echo (Phase + // 5). `error = None` = our signed frame reached the swarm, + // NOT a delivery receipt. On success the payload is done; on + // failure it's retained for Retry — unless the entry is gone + // (history eviction / room reset), in which case drop it so + // the payload map can't leak. + state.apply_send_result(local_id, error); + } UiEvent::ChatMessage { from, name, @@ -2169,6 +2331,7 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { from: Some(from), attachment, links: Vec::new(), + local_send: None, }, ); } @@ -3085,7 +3248,12 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { AppMessage::ChatSubmit => { let text = crate::sanitize::sanitize_chat(&state.chat_input); if !text.is_empty() { - // Local echo (gossip suppresses our own author, so it won't come back). + // Local echo (gossip suppresses our own author, so it won't come + // back). The send is dispatched-or-queued through the pacer so a + // fast burst is trickled at the rate receivers actually admit, + // and the entry carries honest send status (Phase 5). + let id = state.next_send_id; + state.next_send_id += 1; push_chat( &mut state.chat_messages, ChatEntry { @@ -3095,12 +3263,34 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { from: Some(state.self_id.clone()), attachment: None, links: Vec::new(), + local_send: Some(LocalSend { + id, + status: SendStatus::Pending, + }), }, ); - let _ = state.controller.send(CoreCommand::SendChat(text)); + state.submit_send(id, PendingSend::Text(text)); state.chat_input.clear(); } } + AppMessage::ChatSendTick => { + let now = state.send_now_ms(); + let ready = sendqueue::release_ready(&mut state.send_queue, &mut state.send_pacer, now); + for id in ready { + if let Some(payload) = state.send_payloads.get(&id).cloned() { + state.dispatch_send(id, &payload); + } + } + } + AppMessage::RetryChatSend(id) => { + // Re-dispatch a Failed send from its retained payload. Guard against a + // double click while it is already queued/in-flight. + if !state.send_queue.contains(&id) + && let Some(payload) = state.send_payloads.get(&id).cloned() + { + state.submit_send(id, payload); + } + } AppMessage::PickAttachmentFile => { // Native picker off the UI thread; returns (filename, bytes). The // read is BOUNDED (metadata precheck + cap+1 read, Phase 3C) — an @@ -3171,6 +3361,8 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { .attachments .insert(key, AttachmentState::Ready(bytes.clone()), handle); } + let send_id = state.next_send_id; + state.next_send_id += 1; push_chat( &mut state.chat_messages, ChatEntry { @@ -3180,13 +3372,20 @@ fn update(state: &mut AppState, message: AppMessage) -> Task { from: Some(state.self_id.clone()), attachment: Some(att.clone()), links: Vec::new(), + local_send: Some(LocalSend { + id: send_id, + status: SendStatus::Pending, + }), + }, + ); + state.submit_send( + send_id, + PendingSend::File { + text: String::new(), + attachment: att, + data: bytes, }, ); - let _ = state.controller.send(CoreCommand::SendChatFile { - text: String::new(), - attachment: att, - data: bytes, - }); } } AppMessage::SaveAttachment(key) => { @@ -7514,6 +7713,42 @@ fn view(state: &AppState) -> Element<'_, AppMessage> { .spacing(0), ); } + // Honest send status under our own messages (Phase 5). Success + // (Broadcast) and the transient Pending deliberately show + // nothing — there are no delivery receipts, so silence is the + // honest "it went out" state. Only a waiting queue slot or an + // outright failure is surfaced. + if let Some(ls) = &m.local_send { + let status_elem: Option> = match &ls.status { + SendStatus::Queued => { + Some(text("queued…").size(11).color(color_subtext).into()) + } + SendStatus::Failed(reason) => Some( + row![ + text(format!("⚠ Not sent — {reason}")) + .size(11) + .color(color_red), + button(text("Retry").size(11)) + .on_press(AppMessage::RetryChatSend(ls.id)) + .style(b_style(color_surface, color_red, color_text, 6.0)) + .padding(4), + ] + .spacing(6) + .align_y(iced::alignment::Vertical::Center) + .into(), + ), + SendStatus::Pending | SendStatus::Broadcast => None, + }; + if let Some(status_elem) = status_elem { + chat_col = chat_col.push( + row![ + iced::widget::Space::new().width(iced::Length::Fixed(30.0)), + status_elem, + ] + .spacing(0), + ); + } + } } } let chat_scroll = scrollable(chat_col) @@ -9274,6 +9509,8 @@ impl Program for Icon { #[cfg(test)] mod tests { + use super::PendingSend; + use super::sendqueue::{self, LocalSend, SendStatus}; use super::{ AppConfig, AppMessage, AppState, AttachmentCache, AttachmentState, CLOCK_SKEW_WARNING_VISIBLE_SECS, ChatEntry, ClockSkewBanner, GateMeter, METER_MAX, @@ -9284,6 +9521,7 @@ mod tests { update, }; use iroh::SecretKey; + use std::collections::VecDeque; /// Two distinct (peer, id) keys for cache tests; `id` may be shared to model /// a malicious peer reusing a victim's attachment id (Tier C F-12). @@ -9497,6 +9735,7 @@ mod tests { id: shared_id, }), links: Vec::new(), + local_send: None, }; // Attacker's line is FIRST in history, so a bare-id scan would pick it. let messages = vec![mk(attacker, "evil.sh"), mk(victim, "report.pdf")]; @@ -9544,6 +9783,7 @@ mod tests { from: Some(peer.to_string()), attachment: None, links: Vec::new(), + local_send: None, }); state.chat_input = "draft".to_string(); let att_key = (peer, attachment_id); @@ -10453,6 +10693,7 @@ mod tests { from: None, attachment: None, links: Vec::new(), + local_send: None, }; push_chat(&mut messages, entry); assert_eq!(messages.len(), 1); @@ -10475,6 +10716,7 @@ mod tests { from: None, attachment: None, links: Vec::new(), + local_send: None, }, ); } @@ -10506,6 +10748,7 @@ mod tests { from: None, attachment: None, links: Vec::new(), + local_send: None, }, ); } @@ -10541,6 +10784,7 @@ mod tests { from: None, attachment: None, links: Vec::new(), + local_send: None, }, ); let total: usize = messages.iter().map(|m| m.text.len()).sum(); @@ -10568,6 +10812,7 @@ mod tests { from: None, attachment: None, links: Vec::new(), + local_send: None, }, ); // Ranges are filled at push time (the construction-site value is @@ -10579,4 +10824,162 @@ mod tests { let rebuilt: String = pieces.iter().map(|(s, _)| *s).collect(); assert_eq!(rebuilt, m.text); } + + // ---- Phase 5: honest send status (chat-hardening plan) ---- + + /// Push a locally authored echo the way `ChatSubmit` does and return its id. + fn push_own(state: &mut AppState, text: &str) -> u64 { + let id = state.next_send_id; + state.next_send_id += 1; + state.chat_messages.push(ChatEntry { + name: "Me (You)".to_string(), + text: text.to_string(), + mine: true, + from: Some(state.self_id.clone()), + attachment: None, + links: Vec::new(), + local_send: Some(LocalSend { + id, + status: SendStatus::Pending, + }), + }); + id + } + + fn status_of(state: &AppState, id: u64) -> Option { + state + .chat_messages + .iter() + .find_map(|m| m.local_send.as_ref().filter(|s| s.id == id)) + .map(|s| s.status.clone()) + } + + #[test] + fn send_status_pending_then_broadcast_on_success() { + let mut state = AppState::default(); + let id = push_own(&mut state, "hi"); + state.submit_send(id, PendingSend::Text("hi".to_string())); + // Empty queue + a fresh full pacer → dispatched immediately. + assert_eq!(status_of(&state, id), Some(SendStatus::Pending)); + assert!(state.send_payloads.contains_key(&id)); + state.apply_send_result(id, None); + assert_eq!(status_of(&state, id), Some(SendStatus::Broadcast)); + // A completed send releases its retry payload. + assert!(!state.send_payloads.contains_key(&id)); + } + + #[test] + fn send_status_failed_keeps_payload_for_retry() { + let mut state = AppState::default(); + let id = push_own(&mut state, "yo"); + state.submit_send(id, PendingSend::Text("yo".to_string())); + state.apply_send_result(id, Some("not in a room".to_string())); + assert_eq!( + status_of(&state, id), + Some(SendStatus::Failed("not in a room".to_string())) + ); + // Retained so Retry can re-dispatch the exact payload. + assert!(state.send_payloads.contains_key(&id)); + } + + #[test] + fn send_result_updates_only_the_matching_entry() { + let mut state = AppState::default(); + let a = push_own(&mut state, "a"); + state.submit_send(a, PendingSend::Text("a".to_string())); + let b = push_own(&mut state, "b"); + state.submit_send(b, PendingSend::Text("b".to_string())); + state.apply_send_result(a, None); + assert_eq!(status_of(&state, a), Some(SendStatus::Broadcast)); + assert_eq!(status_of(&state, b), Some(SendStatus::Pending)); + // A result for an id with no matching entry is a harmless no-op. + state.apply_send_result(9999, None); + assert_eq!(status_of(&state, b), Some(SendStatus::Pending)); + } + + #[test] + fn send_result_after_eviction_drops_orphan_payload() { + let mut state = AppState::default(); + let id = push_own(&mut state, "gone"); + state.submit_send(id, PendingSend::Text("gone".to_string())); + // Simulate the entry leaving history (byte-budget eviction) with its + // send still in flight. + state + .chat_messages + .retain(|m| m.local_send.as_ref().map(|s| s.id) != Some(id)); + state.apply_send_result(id, Some("dead".to_string())); + // No entry to mark → the payload must not leak. + assert!(!state.send_payloads.contains_key(&id)); + } + + #[test] + fn send_result_after_room_reset_is_a_noop() { + let mut state = AppState::default(); + let id = push_own(&mut state, "before"); + state.submit_send(id, PendingSend::Text("before".to_string())); + state.reset_room_state(); + assert!(state.chat_messages.is_empty()); + assert!(state.send_queue.is_empty()); + assert!(state.send_payloads.is_empty()); + // A late result for the pre-reset send touches nothing and adds no entry. + state.apply_send_result(id, None); + assert!(state.chat_messages.is_empty()); + assert!(state.send_payloads.is_empty()); + } + + #[test] + fn fast_burst_dispatches_the_budget_then_queues() { + let mut state = AppState::default(); + let mut ids = Vec::new(); + // Nine near-instant sends: the pacer grants exactly the 8-message burst + // (sub-millisecond elapsed refills nothing), so the 9th must queue. + for _ in 0..9 { + let id = push_own(&mut state, "m"); + state.submit_send(id, PendingSend::Text("m".to_string())); + ids.push(id); + } + for &id in &ids[..8] { + assert_eq!(status_of(&state, id), Some(SendStatus::Pending)); + } + assert_eq!(status_of(&state, ids[8]), Some(SendStatus::Queued)); + assert_eq!(state.send_queue, VecDeque::from([ids[8]])); + + // Draining once the pacer has refilled (the ChatSendTick path) releases + // the queued message and dispatches it. + let now = state.send_now_ms() + 2000; + let ready = sendqueue::release_ready(&mut state.send_queue, &mut state.send_pacer, now); + for id in ready { + let payload = state.send_payloads.get(&id).cloned().unwrap(); + state.dispatch_send(id, &payload); + } + assert!(state.send_queue.is_empty()); + assert_eq!(status_of(&state, ids[8]), Some(SendStatus::Pending)); + } + + #[test] + fn retry_redispatches_only_the_targeted_send() { + let mut state = AppState::default(); + let a = push_own(&mut state, "a"); + state.submit_send(a, PendingSend::Text("a".to_string())); + let b = push_own(&mut state, "b"); + state.submit_send(b, PendingSend::Text("b".to_string())); + // Both fail. + state.apply_send_result(a, Some("x".to_string())); + state.apply_send_result(b, Some("x".to_string())); + // Retry a: dispatched or requeued; b is untouched and still Failed. + if !state.send_queue.contains(&a) + && let Some(payload) = state.send_payloads.get(&a).cloned() + { + state.submit_send(a, payload); + } + assert!(matches!( + status_of(&state, a), + Some(SendStatus::Pending) | Some(SendStatus::Queued) + )); + assert_eq!( + status_of(&state, b), + Some(SendStatus::Failed("x".to_string())) + ); + assert!(state.send_payloads.contains_key(&b)); + } } diff --git a/src/app/sendqueue.rs b/src/app/sendqueue.rs new file mode 100644 index 0000000..f479188 --- /dev/null +++ b/src/app/sendqueue.rs @@ -0,0 +1,129 @@ +//! Sender-side chat send status and pacing (chat-hardening Phase 5). +//! +//! Every RECEIVER admits our chat through a per-author token bucket +//! ([`CHAT_AUTHOR_BURST`] then 1/s) and silently drops what exceeds it, with no +//! acknowledgement wire. The only way the sender can be honest about fast +//! bursts is to never exceed that budget in the first place: sends past the +//! burst are queued locally (shown as "queued…") and trickled out at the +//! receivers' sustained rate. The pacer deliberately reuses the receiver +//! gate's own [`TokenBucket`] and constants so the two sides of the policy +//! cannot drift apart. +//! +//! Everything here is pure — `now_ms` is passed in, never read from a clock — +//! so every boundary is unit-testable. + +use std::collections::VecDeque; + +use crate::network::gossip::{CHAT_AUTHOR_BURST, CHAT_AUTHOR_REFILL_PER_MS, TokenBucket}; + +/// Send lifecycle of one locally authored chat message. Success is +/// [`SendStatus::Broadcast`] — "our signed frame was handed to the gossip +/// swarm" — deliberately NOT "delivered": PeerSpeak has no peer +/// acknowledgements, so the honest success presentation is no label at all. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum SendStatus { + /// Waiting in the local outbound queue for a pacer token. + Queued, + /// Handed to the core; the broadcast result has not come back yet. + Pending, + /// The signed broadcast reached the gossip swarm. + Broadcast, + /// The send failed; carries a short reason. The entry offers a Retry. + Failed(String), +} + +/// Local-only send bookkeeping attached to our own chat entries. The id never +/// goes on the wire; it ties a `ChatSendResult` back to the matching echo. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct LocalSend { + pub id: u64, + pub status: SendStatus, +} + +/// Sender-side pacer mirroring the receiver's per-author admission budget. +#[derive(Debug, Clone, Copy)] +pub struct SendPacer { + bucket: TokenBucket, +} + +impl SendPacer { + pub fn new(now_ms: u64) -> Self { + Self { + bucket: TokenBucket::full(CHAT_AUTHOR_BURST, now_ms), + } + } + + /// Take one send token if the mirrored per-author budget allows it now. + pub fn try_send(&mut self, now_ms: u64) -> bool { + self.bucket + .try_take(CHAT_AUTHOR_BURST, CHAT_AUTHOR_REFILL_PER_MS, now_ms) + } +} + +/// Pop the queued ids that may be dispatched now: strict front-of-queue order, +/// one pacer token each, stopping at the first refusal so a message can never +/// overtake an earlier one. +pub fn release_ready(queue: &mut VecDeque, pacer: &mut SendPacer, now_ms: u64) -> Vec { + let mut ready = Vec::new(); + while !queue.is_empty() && pacer.try_send(now_ms) { + // The unwrap is safe: the loop condition just checked non-empty. + ready.push(queue.pop_front().unwrap()); + } + ready +} + +#[cfg(test)] +mod tests { + use super::*; + + const T0: u64 = 1_000_000; + + #[test] + fn pacer_allows_the_full_burst_then_refuses() { + let mut pacer = SendPacer::new(T0); + for _ in 0..CHAT_AUTHOR_BURST as usize { + assert!(pacer.try_send(T0)); + } + assert!(!pacer.try_send(T0)); + } + + #[test] + fn pacer_refills_at_one_per_second() { + let mut pacer = SendPacer::new(T0); + for _ in 0..CHAT_AUTHOR_BURST as usize { + assert!(pacer.try_send(T0)); + } + // 999ms is just under one token; 1000ms grants exactly one. + assert!(!pacer.try_send(T0 + 999)); + assert!(pacer.try_send(T0 + 1000)); + assert!(!pacer.try_send(T0 + 1000)); + } + + #[test] + fn release_ready_preserves_order_and_stops_at_refusal() { + let mut pacer = SendPacer::new(T0); + // Drain the burst so only refill tokens remain. + for _ in 0..CHAT_AUTHOR_BURST as usize { + assert!(pacer.try_send(T0)); + } + let mut queue: VecDeque = [10, 11, 12].into_iter().collect(); + // 2 seconds of refill = 2 tokens: exactly the first two, in order. + let ready = release_ready(&mut queue, &mut pacer, T0 + 2000); + assert_eq!(ready, vec![10, 11]); + assert_eq!(queue, VecDeque::from([12])); + // No tokens left at the same instant. + assert!(release_ready(&mut queue, &mut pacer, T0 + 2000).is_empty()); + assert_eq!(queue, VecDeque::from([12])); + } + + #[test] + fn release_ready_empty_queue_consumes_no_tokens() { + let mut pacer = SendPacer::new(T0); + let mut queue = VecDeque::new(); + assert!(release_ready(&mut queue, &mut pacer, T0).is_empty()); + // The full burst must still be available. + for _ in 0..CHAT_AUTHOR_BURST as usize { + assert!(pacer.try_send(T0)); + } + } +} diff --git a/src/core/messages.rs b/src/core/messages.rs index 2470da3..db7bd4f 100644 --- a/src/core/messages.rs +++ b/src/core/messages.rs @@ -66,12 +66,21 @@ pub enum CoreCommand { /// Set what a recording captures (mixed / per-peer stems / both). Takes /// effect on the next recording start. Sent at startup from config. SetRecordingMode(RecordingMode), - /// Broadcast a room text-chat message. No-op when not in a call. - SendChat(String), + /// Broadcast a room text-chat message. `local_id` is the app's local-only + /// handle for this send — it never goes on the wire; the core echoes it back + /// in [`UiEvent::ChatSendResult`] so the UI can mark the matching local echo + /// honestly (chat-hardening Phase 5). Not being in a call is a FAILURE + /// result, not a silent no-op. + SendChat { + local_id: u64, + text: String, + }, /// Send a chat message carrying a file attachment. The app has already read + /// capped the file and built the descriptor; core makes the bytes available - /// on the file plane and broadcasts the descriptor. + /// on the file plane and broadcasts the descriptor. `local_id` as in + /// [`CoreCommand::SendChat`]. SendChatFile { + local_id: u64, text: String, attachment: crate::files::ChatAttachment, /// Shared, not owned: the same allocation is retained by the UI cache @@ -227,8 +236,12 @@ pub fn delivery_class(cmd: &CoreCommand) -> DeliveryClass { | CoreCommand::SetAudioProfile(_) | CoreCommand::SetRecording(_) | CoreCommand::SetRecordingMode(_) - | CoreCommand::SendChat(_) + | CoreCommand::SendChat { + local_id: _, + text: _, + } | CoreCommand::SendChatFile { + local_id: _, text: _, attachment: _, data: _, @@ -315,8 +328,12 @@ pub fn coalesce_key(cmd: &CoreCommand) -> Option { | CoreCommand::SetAudioProfile(_) | CoreCommand::SetRecording(_) | CoreCommand::SetRecordingMode(_) - | CoreCommand::SendChat(_) + | CoreCommand::SendChat { + local_id: _, + text: _, + } | CoreCommand::SendChatFile { + local_id: _, text: _, attachment: _, data: _, @@ -420,6 +437,15 @@ pub enum UiEvent { RecordingStopped { path: String, }, + /// The outcome of one locally initiated chat send (chat-hardening Phase 5). + /// `error = None` means our signed broadcast was handed to the gossip swarm + /// — deliberately NOT a delivery/read receipt; PeerSpeak has no peer + /// acknowledgements. `local_id` is the app's own handle from the + /// `SendChat`/`SendChatFile` command and never appears on the wire. + ChatSendResult { + local_id: u64, + error: Option, + }, /// A room text-chat message arrived from a peer (never our own — local /// messages are echoed by the UI on send). `from` is the sender's node id /// string, used to key their avatar (W4). @@ -616,7 +642,10 @@ mod tests { CoreCommand::SetPeerMuted(peer, true), CoreCommand::SetPresenceMode(PresenceMode::Normal), CoreCommand::SetAudioProfile(crate::config::AudioProfile::BadNetwork), - CoreCommand::SendChat("hello".to_string()), + CoreCommand::SendChat { + local_id: 1, + text: "hello".to_string(), + }, ]; for cmd in commands { diff --git a/src/core/mod.rs b/src/core/mod.rs index a2cd555..b0bc4e7 100644 --- a/src/core/mod.rs +++ b/src/core/mod.rs @@ -3259,27 +3259,54 @@ async fn run_core_loop( } } - CoreCommand::SendChat(text) => { - if let Some(session) = &active_session - && let Err(e) = session.room_state.send_chat(text, None).await - { + CoreCommand::SendChat { local_id, text } => { + // Honest result either way (Phase 5): no active session is a + // FAILURE the sender must see, not a silent drop. + let error = match &active_session { + Some(session) => session + .room_state + .send_chat(text, None) + .await + .err() + .map(|e| e.to_string()), + None => Some("not in a room".to_string()), + }; + if let Some(e) = &error { crate::log_msg(&format!("Failed to send chat: {e}")); } + let _ = ui_tx + .send(UiEvent::ChatSendResult { local_id, error }) + .await; } CoreCommand::SendChatFile { + local_id, text, attachment, data, } => { - if let Some(session) = &active_session { - // Make the bytes fetchable by room members, then broadcast the - // descriptor alongside the (possibly empty) caption text. - session.transport.serve_attachment(attachment.id, data); - if let Err(e) = session.room_state.send_chat(text, Some(attachment)).await { - crate::log_msg(&format!("Failed to send chat file: {e}")); + let error = match &active_session { + Some(session) => { + // Make the bytes fetchable by room members, then broadcast + // the descriptor alongside the (possibly empty) caption + // text. Re-serving the same id on a retry REPLACES the + // store entry (same Arc), never double-counts it. + session.transport.serve_attachment(attachment.id, data); + session + .room_state + .send_chat(text, Some(attachment)) + .await + .err() + .map(|e| e.to_string()) } + None => Some("not in a room".to_string()), + }; + if let Some(e) = &error { + crate::log_msg(&format!("Failed to send chat file: {e}")); } + let _ = ui_tx + .send(UiEvent::ChatSendResult { local_id, error }) + .await; } CoreCommand::FetchAttachment { from, attachment } => { diff --git a/src/network/gossip.rs b/src/network/gossip.rs index 2a7b295..4bd15f5 100644 --- a/src/network/gossip.rs +++ b/src/network/gossip.rs @@ -271,8 +271,10 @@ const CHAT_REPLAY_CACHE_CAP: usize = 1024; /// Per-author chat budget: a burst of 8 absorbs a fast typist; 1 msg/s sustained /// is well above real human chat rate while bounding a flooder to a trickle. -const CHAT_AUTHOR_BURST: f64 = 8.0; -const CHAT_AUTHOR_REFILL_PER_MS: f64 = 1.0 / 1000.0; +/// `pub(crate)` because the sender-side pacer (chat-hardening Phase 5) mirrors +/// this exact policy — one definition, so the two sides can never drift apart. +pub(crate) const CHAT_AUTHOR_BURST: f64 = 8.0; +pub(crate) const CHAT_AUTHOR_REFILL_PER_MS: f64 = 1.0 / 1000.0; /// Room-wide chat budget across ALL authors, so a set of sock-puppet identities /// can't multiply the per-author budget into unbounded event-channel pressure. @@ -290,15 +292,16 @@ const CHAT_AUTHOR_BUCKETS_CAP: usize = 64; const CHAT_REJECT_LOG_COOLDOWN_MS: u64 = 10_000; /// A minimal deterministic token bucket: time is passed in, never read from a -/// clock, so every boundary is unit-testable. +/// clock, so every boundary is unit-testable. Shared with the sender-side chat +/// pacer (`app::sendqueue`) so both sides of the rate policy use one mechanism. #[derive(Debug, Clone, Copy)] -struct TokenBucket { +pub(crate) struct TokenBucket { tokens: f64, last_ms: u64, } impl TokenBucket { - fn full(burst: f64, now_ms: u64) -> Self { + pub(crate) fn full(burst: f64, now_ms: u64) -> Self { Self { tokens: burst, last_ms: now_ms, @@ -307,7 +310,7 @@ impl TokenBucket { /// Refill for elapsed time (capped at `burst`), then take one token if /// available. Returns whether a token was consumed. - fn try_take(&mut self, burst: f64, refill_per_ms: f64, now_ms: u64) -> bool { + pub(crate) fn try_take(&mut self, burst: f64, refill_per_ms: f64, now_ms: u64) -> bool { let elapsed = now_ms.saturating_sub(self.last_ms) as f64; self.tokens = (self.tokens + elapsed * refill_per_ms).min(burst); self.last_ms = now_ms; @@ -1276,26 +1279,29 @@ impl RoomState for IrohGossipState { let sender_opt = self.active_sender.lock().unwrap().clone(); let topic_opt = *self.active_topic_bytes.lock().unwrap(); - if let (Some(sender), Some(topic)) = (sender_opt, topic_opt) { - let payload = sign_gossip( - &self.secret_key, - &topic, + // A missing sender/topic or an encode failure is a real send failure the + // caller must see (chat-hardening Phase 5) — silently returning Ok here + // would let the UI present an unsent message as broadcast. + let (Some(sender), Some(topic)) = (sender_opt, topic_opt) else { + return Err(NetError::Other("Not in a room".to_string())); + }; + let payload = sign_gossip( + &self.secret_key, + &topic, + ts, + GossipMessage::Chat { + name, + text, ts, - GossipMessage::Chat { - name, - text, - ts, - attachment, - }, - ); - if let Ok(bytes) = serde_json::to_vec(&payload) { - sender - .broadcast(bytes.into()) - .await - .map_err(|e| NetError::Gossip(e.to_string()))?; - } - } - Ok(()) + attachment, + }, + ); + let bytes = serde_json::to_vec(&payload) + .map_err(|e| NetError::Other(format!("Failed to encode chat: {e}")))?; + sender + .broadcast(bytes.into()) + .await + .map_err(|e| NetError::Gossip(e.to_string())) } async fn leave(&self) -> Result<(), NetError> {