diff --git a/core/archipelago/src/mesh/listener/mod.rs b/core/archipelago/src/mesh/listener/mod.rs index 4cd8baf3..4f2b9117 100644 --- a/core/archipelago/src/mesh/listener/mod.rs +++ b/core/archipelago/src/mesh/listener/mod.rs @@ -21,7 +21,7 @@ use std::collections::{HashMap, HashSet, VecDeque}; use std::sync::Arc; use std::time::Duration; use tokio::sync::{broadcast, mpsc, RwLock}; -use tracing::{error, info}; +use tracing::{error, info, warn}; /// How often to broadcast our identity advertisement (seconds). const ADVERT_INTERVAL: Duration = Duration::from_secs(60); @@ -52,6 +52,22 @@ const RX_STALL_TIMEOUT: Duration = Duration::from_secs(1800); /// Maximum stored messages (circular buffer). const MAX_MESSAGES: usize = 100; +/// On-disk message-history file under `data_dir/` (written by +/// `spawn_message_persister`, restored by `load_persisted_messages`). +const MESSAGES_FILE: &str = "mesh-messages.json"; + +/// Serialized form of the persisted history. `send_seqs` rides along because +/// the per-target outbound sequence counters must survive restarts too: +/// receivers dedup on (sender_pubkey, sender_seq), so a node that reboots and +/// starts counting from 1 again has its first messages silently dropped by +/// every peer that already saw those sequence numbers. +#[derive(serde::Serialize, serde::Deserialize)] +struct PersistedMessages { + messages: Vec, + #[serde(default)] + send_seqs: HashMap, +} + /// Check if two ISO8601 timestamps are within N seconds of each other. fn within_seconds_iso(ts1: &str, ts2: &str, secs: i64) -> bool { use chrono::DateTime; @@ -402,6 +418,90 @@ impl MeshState { let count = self.peers.read().await.len(); self.status.write().await.peer_count = count; } + + /// Restore the persisted message history + outbound sequence counters. + /// Called once at service startup, before the listener spawns. Missing + /// file (fresh node) is normal; a corrupt file is logged and skipped + /// rather than blocking mesh startup. + pub async fn load_persisted_messages(&self) { + let path = self.data_dir.join(MESSAGES_FILE); + let bytes = match tokio::fs::read(&path).await { + Ok(b) => b, + Err(e) if e.kind() == std::io::ErrorKind::NotFound => return, + Err(e) => { + warn!("mesh: reading {} failed: {e}", path.display()); + return; + } + }; + let persisted: PersistedMessages = match serde_json::from_slice(&bytes) { + Ok(p) => p, + Err(e) => { + warn!("mesh: parsing {} failed (skipping restore): {e}", path.display()); + return; + } + }; + let count = persisted.messages.len(); + let max_id = persisted.messages.iter().map(|m| m.id).max().unwrap_or(0); + *self.messages.write().await = persisted.messages.into(); + if !persisted.send_seqs.is_empty() { + *self.next_send_seq.write().await = persisted.send_seqs; + } + { + let mut id = self.next_message_id.write().await; + if *id <= max_id { + *id = max_id + 1; + } + } + info!("mesh: restored {count} persisted messages (next id {})", max_id + 1); + } +} + +/// Persist the message history on a low-rate timer instead of hooking every +/// mutation path (store, delivered/transport/encrypted stamps, edits, deletes, +/// read-receipt prunes — and whatever gets added next). Serializing ≤100 +/// messages every few seconds is trivial; the write is skipped when nothing +/// changed, and at most the last few seconds of history are lost on a hard +/// kill — versus the entire history, which is what a restart cost before this +/// existed. Atomic write-then-rename, 0600 (DM plaintext lives in this file). +pub fn spawn_message_persister(state: Arc) { + tokio::spawn(async move { + use std::os::unix::fs::PermissionsExt; + let path = state.data_dir.join(MESSAGES_FILE); + let tmp = path.with_extension("json.tmp"); + let mut last_written: Option> = None; + let mut tick = tokio::time::interval(Duration::from_secs(5)); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + loop { + tick.tick().await; + let snapshot = PersistedMessages { + messages: state.messages.read().await.iter().cloned().collect(), + send_seqs: state.next_send_seq.read().await.clone(), + }; + let json = match serde_json::to_vec_pretty(&snapshot) { + Ok(j) => j, + Err(e) => { + warn!("mesh: serializing message history failed: {e}"); + continue; + } + }; + if last_written.as_deref() == Some(json.as_slice()) { + continue; + } + if let Err(e) = tokio::fs::write(&tmp, &json).await { + warn!("mesh: writing {} failed: {e}", tmp.display()); + continue; + } + let perms = std::fs::Permissions::from_mode(0o600); + if let Err(e) = tokio::fs::set_permissions(&tmp, perms).await { + warn!("mesh: chmod {} failed: {e}", tmp.display()); + } + if let Err(e) = tokio::fs::rename(&tmp, &path).await { + warn!("mesh: renaming {} -> {} failed: {e}", tmp.display(), path.display()); + continue; + } + last_written = Some(json); + } + }); } /// Spawn the background mesh listener task. diff --git a/core/archipelago/src/mesh/mod.rs b/core/archipelago/src/mesh/mod.rs index 8dcb3101..62681b16 100644 --- a/core/archipelago/src/mesh/mod.rs +++ b/core/archipelago/src/mesh/mod.rs @@ -691,6 +691,13 @@ impl MeshService { } } + // Restore the message history + outbound send-sequence counters, then + // keep them persisted. Until 2026-07-22 both were RAM-only: every + // restart wiped the chat history, and the reset sequence counters made + // peers drop the first post-reboot messages as (pubkey, seq) replays. + state.load_persisted_messages().await; + listener::spawn_message_persister(Arc::clone(&state)); + Ok(Self { state, config,