fix(mesh): persist message history + send-seq counters across restarts
Both lived only in RAM: every archipelago restart/reboot wiped the whole chat history (user-reported), and — worse — reset the per-target outbound sequence counters, so peers' (sender_pubkey, sender_seq) dedup silently dropped the first messages sent after a reboot as replays. mesh-messages.json in the data dir (0600 — DM plaintext), restored in MeshService::new before the listener spawns; a 5s debounced persister task snapshots at one choke point instead of hooking every mutation path (store, delivered/transport/encrypted stamps, edits, deletes, prunes). Atomic write-then-rename; corrupt/missing file skips restore instead of blocking mesh startup. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
parent
27331d66e4
commit
2f26cb2bc4
@ -21,7 +21,7 @@ use std::collections::{HashMap, HashSet, VecDeque};
|
|||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
use tokio::sync::{broadcast, mpsc, RwLock};
|
use tokio::sync::{broadcast, mpsc, RwLock};
|
||||||
use tracing::{error, info};
|
use tracing::{error, info, warn};
|
||||||
|
|
||||||
/// How often to broadcast our identity advertisement (seconds).
|
/// How often to broadcast our identity advertisement (seconds).
|
||||||
const ADVERT_INTERVAL: Duration = Duration::from_secs(60);
|
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).
|
/// Maximum stored messages (circular buffer).
|
||||||
const MAX_MESSAGES: usize = 100;
|
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<MeshMessage>,
|
||||||
|
#[serde(default)]
|
||||||
|
send_seqs: HashMap<u32, u64>,
|
||||||
|
}
|
||||||
|
|
||||||
/// Check if two ISO8601 timestamps are within N seconds of each other.
|
/// Check if two ISO8601 timestamps are within N seconds of each other.
|
||||||
fn within_seconds_iso(ts1: &str, ts2: &str, secs: i64) -> bool {
|
fn within_seconds_iso(ts1: &str, ts2: &str, secs: i64) -> bool {
|
||||||
use chrono::DateTime;
|
use chrono::DateTime;
|
||||||
@ -402,6 +418,90 @@ impl MeshState {
|
|||||||
let count = self.peers.read().await.len();
|
let count = self.peers.read().await.len();
|
||||||
self.status.write().await.peer_count = count;
|
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<MeshState>) {
|
||||||
|
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<Vec<u8>> = 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.
|
/// Spawn the background mesh listener task.
|
||||||
|
|||||||
@ -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 {
|
Ok(Self {
|
||||||
state,
|
state,
|
||||||
config,
|
config,
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user