mollusk/peerspeak
Public Access
1
0
Fork 0
Code Issues Pull Requests 3 Actions Packages Projects Releases 11 Wiki Activity

chat: honest local send status + sender-side pacing (Phase 5)
CI / check (push) Successful in 4m17s
Details

Browse Source 
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<LocalSend>; 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 <noreply@anthropic.com>
This commit is contained in:
mollusk
2026-07-18 19:03:25 -04:00
co-authored by Claude Fable 5
parent 77d2bf2992
commit 5f4eba1815
6 changed files with 728 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/app/mod.rs
+410 -7
View File
@@ -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<std::ops::Range<usize>>,
/// 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<LocalSend>,
}
/// 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<Vec<u8>>,
},
}
/// 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<AttachmentKey>,
/// Clip ids waiting for the existing attachment fetch path to return bytes.
pending_plays: HashSet<crate::files::AttachmentId>,
/// 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<u64>,
/// 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<u64, PendingSend>,
/// Filename-hinted audio whose fetched bytes or decoder validation failed;
/// these entries fall back to the normal file chip.
invalid_audio: HashSet<crate::files::AttachmentId>,
@@ -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<String>) {
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<AppMessage> {
} 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<AppMessage> {
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<AppMessage> {
from: Some(from),
attachment,
links: Vec::new(),
local_send: None,
},
);
}
@@ -3085,7 +3248,12 @@ fn update(state: &mut AppState, message: AppMessage) -> Task<AppMessage> {
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<AppMessage> {
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<AppMessage> {
.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<AppMessage> {
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<Element<'_, AppMessage>> = 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<AppMessage> 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<SendStatus> {
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));
}
}