From a6fc7145bc0c73841dfac1ce98fbbd20981baefc Mon Sep 17 00:00:00 2001 From: l5y <220195275+l5yth@users.noreply.github.com> Date: Mon, 5 Jan 2026 17:34:54 +0100 Subject: [PATCH] matrix: cache seen messages by rx_time not id (#594) * matrix: cache seen messages by rx_time not id * matrix: fix review comments * matrix: fix review comments * matrix: cover missing unit test vectors * matrix: fix tests --- matrix/src/main.rs | 290 +++++++++++++++++++++++++++++---------------- 1 file changed, 187 insertions(+), 103 deletions(-) diff --git a/matrix/src/main.rs b/matrix/src/main.rs index 0a9a397..9c17419 100644 --- a/matrix/src/main.rs +++ b/matrix/src/main.rs @@ -28,7 +28,16 @@ use crate::potatomesh::{FetchParams, PotatoClient, PotatoMessage}; #[derive(Debug, serde::Serialize, serde::Deserialize, Default)] pub struct BridgeState { + /// Highest message id processed by the bridge. last_message_id: Option, + /// Highest rx_time observed; used to build incremental fetch queries. + #[serde(default)] + last_rx_time: Option, + /// Message ids seen at the current last_rx_time for de-duplication. + #[serde(default)] + last_rx_time_ids: Vec, + /// Legacy checkpoint timestamp used before last_rx_time was added. + #[serde(default, skip_serializing)] last_checked_at: Option, } @@ -42,7 +51,10 @@ impl BridgeState { if data.trim().is_empty() { return Ok(Self::default()); } - let s: Self = serde_json::from_str(&data)?; + let mut s: Self = serde_json::from_str(&data)?; + if s.last_rx_time.is_none() { + s.last_rx_time = s.last_checked_at; + } Ok(s) } @@ -53,17 +65,32 @@ impl BridgeState { } fn should_forward(&self, msg: &PotatoMessage) -> bool { - match self.last_message_id { - None => true, - Some(last) => msg.id > last, + match self.last_rx_time { + None => match self.last_message_id { + None => true, + Some(last_id) => msg.id > last_id, + }, + Some(last_ts) => { + if msg.rx_time > last_ts { + true + } else if msg.rx_time < last_ts { + false + } else { + !self.last_rx_time_ids.contains(&msg.id) + } + } } } fn update_with(&mut self, msg: &PotatoMessage) { - self.last_message_id = Some(match self.last_message_id { - None => msg.id, - Some(last) => last.max(msg.id), - }); + self.last_message_id = Some(msg.id); + if self.last_rx_time.is_none() || Some(msg.rx_time) > self.last_rx_time { + self.last_rx_time = Some(msg.rx_time); + self.last_rx_time_ids = vec![msg.id]; + } else if Some(msg.rx_time) == self.last_rx_time && !self.last_rx_time_ids.contains(&msg.id) + { + self.last_rx_time_ids.push(msg.id); + } } } @@ -73,7 +100,7 @@ fn build_fetch_params(state: &BridgeState) -> FetchParams { limit: None, since: None, } - } else if let Some(ts) = state.last_checked_at { + } else if let Some(ts) = state.last_rx_time { FetchParams { limit: None, since: Some(ts), @@ -86,34 +113,18 @@ fn build_fetch_params(state: &BridgeState) -> FetchParams { } } -fn update_checkpoint(state: &mut BridgeState, delivered_all: bool, now_secs: u64) -> bool { - if !delivered_all { - return false; - } - - if state.last_message_id.is_some() { - state.last_checked_at = Some(now_secs); - true - } else { - false - } -} - async fn poll_once( potato: &PotatoClient, matrix: &MatrixAppserviceClient, state: &mut BridgeState, state_path: &str, - now_secs: u64, ) { let params = build_fetch_params(state); match potato.fetch_messages(params).await { Ok(mut msgs) => { - // sort by id ascending so we process in order - msgs.sort_by_key(|m| m.id); - - let mut delivered_all = true; + // sort by rx_time so we process by actual receipt time + msgs.sort_by_key(|m| m.rx_time); for msg in &msgs { if !state.should_forward(msg) { @@ -124,13 +135,15 @@ async fn poll_once( if let Some(port) = &msg.portnum { if port != "TEXT_MESSAGE_APP" { state.update_with(msg); + if let Err(e) = state.save(state_path) { + error!("Error saving state: {:?}", e); + } continue; } } if let Err(e) = handle_message(potato, matrix, state, msg).await { error!("Error handling message {}: {:?}", msg.id, e); - delivered_all = false; continue; } @@ -139,13 +152,6 @@ async fn poll_once( error!("Error saving state: {:?}", e); } } - - // Only advance checkpoint after successful delivery and a known last_message_id. - if update_checkpoint(state, delivered_all, now_secs) { - if let Err(e) = state.save(state_path) { - error!("Error saving state: {:?}", e); - } - } } Err(e) => { error!("Error fetching PotatoMesh messages: {:?}", e); @@ -180,12 +186,7 @@ async fn main() -> Result<()> { let poll_interval = Duration::from_secs(cfg.potatomesh.poll_interval_secs); loop { - let now_secs = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs(); - - poll_once(&potato, &matrix, &mut state, state_path, now_secs).await; + poll_once(&potato, &matrix, &mut state, state_path).await; sleep(poll_interval).await; } @@ -212,22 +213,14 @@ async fn handle_message( .clone() .unwrap_or_else(|| node.long_name.clone()); + let preset_short = modem_preset_short(&msg.modem_preset); let body = format!( - "[{short}] {text}\n({from_id} → {to_id}, {rssi}, {snr}, {chan}/{preset})", + "[{freq}][{preset_short}][{channel}][{short}] {text}", + freq = msg.lora_freq, + preset_short = preset_short, + channel = msg.channel_name, short = short, text = msg.text, - from_id = msg.from_id, - to_id = msg.to_id, - rssi = msg - .rssi - .map(|v| format!("RSSI {v} dB")) - .unwrap_or_else(|| "RSSI n/a".to_string()), - snr = msg - .snr - .map(|v| format!("SNR {v} dB")) - .unwrap_or_else(|| "SNR n/a".to_string()), - chan = msg.channel_name, - preset = msg.modem_preset, ); matrix.send_text_message_as(&user_id, &body).await?; @@ -236,6 +229,19 @@ async fn handle_message( Ok(()) } +/// Build a compact modem preset label like "LF" for "LongFast". +fn modem_preset_short(preset: &str) -> String { + let letters: String = preset + .chars() + .filter(|ch| ch.is_ascii_uppercase()) + .collect(); + if letters.is_empty() { + preset.chars().take(2).collect() + } else { + letters + } +} + #[cfg(test)] mod tests { use super::*; @@ -264,6 +270,12 @@ mod tests { } } + #[test] + fn modem_preset_short_handles_camelcase() { + assert_eq!(modem_preset_short("LongFast"), "LF"); + assert_eq!(modem_preset_short("MediumFast"), "MF"); + } + #[test] fn bridge_state_initially_forwards_all() { let state = BridgeState::default(); @@ -273,39 +285,72 @@ mod tests { } #[test] - fn bridge_state_tracks_highest_id_and_skips_older() { + fn bridge_state_tracks_latest_rx_time_and_skips_older() { let mut state = BridgeState::default(); let m1 = sample_msg(10); let m2 = sample_msg(20); let m3 = sample_msg(15); + let m1 = PotatoMessage { rx_time: 10, ..m1 }; + let m2 = PotatoMessage { rx_time: 20, ..m2 }; + let m3 = PotatoMessage { rx_time: 15, ..m3 }; // First message, should forward assert!(state.should_forward(&m1)); state.update_with(&m1); assert_eq!(state.last_message_id, Some(10)); + assert_eq!(state.last_rx_time, Some(10)); // Second message, higher id, should forward assert!(state.should_forward(&m2)); state.update_with(&m2); assert_eq!(state.last_message_id, Some(20)); + assert_eq!(state.last_rx_time, Some(20)); // Third message, lower than last, should NOT forward assert!(!state.should_forward(&m3)); // state remains unchanged assert_eq!(state.last_message_id, Some(20)); + assert_eq!(state.last_rx_time, Some(20)); } #[test] - fn bridge_state_update_is_monotonic() { - let mut state = BridgeState { - last_message_id: Some(50), + fn bridge_state_uses_legacy_id_filter_when_rx_time_missing() { + let state = BridgeState { + last_message_id: Some(10), + last_rx_time: None, + last_rx_time_ids: vec![], last_checked_at: None, }; - let m = sample_msg(40); + let older = sample_msg(9); + let newer = sample_msg(11); - state.update_with(&m); // id is lower than current - // last_message_id must stay at 50 - assert_eq!(state.last_message_id, Some(50)); + assert!(!state.should_forward(&older)); + assert!(state.should_forward(&newer)); + } + + #[test] + fn bridge_state_dedupes_same_timestamp() { + let mut state = BridgeState::default(); + let m1 = PotatoMessage { + rx_time: 100, + ..sample_msg(10) + }; + let m2 = PotatoMessage { + rx_time: 100, + ..sample_msg(9) + }; + let dup = PotatoMessage { + rx_time: 100, + ..sample_msg(10) + }; + + assert!(state.should_forward(&m1)); + state.update_with(&m1); + assert!(state.should_forward(&m2)); + state.update_with(&m2); + assert!(!state.should_forward(&dup)); + assert_eq!(state.last_rx_time, Some(100)); + assert_eq!(state.last_rx_time_ids, vec![10, 9]); } #[test] @@ -316,13 +361,17 @@ mod tests { let state = BridgeState { last_message_id: Some(12345), - last_checked_at: Some(99), + last_rx_time: Some(99), + last_rx_time_ids: vec![123], + last_checked_at: Some(77), }; state.save(path_str).unwrap(); let loaded_state = BridgeState::load(path_str).unwrap(); assert_eq!(loaded_state.last_message_id, Some(12345)); - assert_eq!(loaded_state.last_checked_at, Some(99)); + assert_eq!(loaded_state.last_rx_time, Some(99)); + assert_eq!(loaded_state.last_rx_time_ids, vec![123]); + assert_eq!(loaded_state.last_checked_at, None); } #[test] @@ -333,7 +382,8 @@ mod tests { let state = BridgeState::load(path_str).unwrap(); assert_eq!(state.last_message_id, None); - assert_eq!(state.last_checked_at, None); + assert_eq!(state.last_rx_time, None); + assert!(state.last_rx_time_ids.is_empty()); } #[test] @@ -346,50 +396,36 @@ mod tests { let state = BridgeState::load(path_str).unwrap(); assert_eq!(state.last_message_id, None); + assert_eq!(state.last_rx_time, None); + assert!(state.last_rx_time_ids.is_empty()); assert_eq!(state.last_checked_at, None); } #[test] - fn update_checkpoint_requires_last_message_id() { - let mut state = BridgeState { - last_message_id: None, - last_checked_at: Some(10), - }; + fn bridge_state_migrates_legacy_checkpoint() { + let tmp_dir = tempfile::tempdir().unwrap(); + let file_path = tmp_dir.path().join("legacy_state.json"); + let path_str = file_path.to_str().unwrap(); - let saved = update_checkpoint(&mut state, true, 123); - assert!(!saved); - assert_eq!(state.last_checked_at, Some(10)); - } + fs::write( + path_str, + r#"{"last_message_id":42,"last_checked_at":1710000000}"#, + ) + .unwrap(); - #[test] - fn update_checkpoint_skips_when_not_delivered() { - let mut state = BridgeState { - last_message_id: Some(5), - last_checked_at: Some(10), - }; - - let saved = update_checkpoint(&mut state, false, 123); - assert!(!saved); - assert_eq!(state.last_checked_at, Some(10)); - } - - #[test] - fn update_checkpoint_sets_when_safe() { - let mut state = BridgeState { - last_message_id: Some(5), - last_checked_at: None, - }; - - let saved = update_checkpoint(&mut state, true, 123); - assert!(saved); - assert_eq!(state.last_checked_at, Some(123)); + let state = BridgeState::load(path_str).unwrap(); + assert_eq!(state.last_message_id, Some(42)); + assert_eq!(state.last_rx_time, Some(1_710_000_000)); + assert!(state.last_rx_time_ids.is_empty()); } #[test] fn fetch_params_respects_missing_last_message_id() { let state = BridgeState { last_message_id: None, - last_checked_at: Some(123), + last_rx_time: Some(123), + last_rx_time_ids: vec![], + last_checked_at: None, }; let params = build_fetch_params(&state); @@ -401,7 +437,9 @@ mod tests { fn fetch_params_uses_since_when_safe() { let state = BridgeState { last_message_id: Some(1), - last_checked_at: Some(123), + last_rx_time: Some(123), + last_rx_time_ids: vec![], + last_checked_at: None, }; let params = build_fetch_params(&state); @@ -413,6 +451,8 @@ mod tests { fn fetch_params_defaults_to_small_window() { let state = BridgeState { last_message_id: Some(1), + last_rx_time: None, + last_rx_time_ids: vec![], last_checked_at: None, }; @@ -422,7 +462,7 @@ mod tests { } #[tokio::test] - async fn poll_once_persists_checkpoint_without_messages() { + async fn poll_once_leaves_state_unchanged_without_messages() { let tmp_dir = tempfile::tempdir().unwrap(); let state_path = tmp_dir.path().join("state.json"); let state_str = state_path.to_str().unwrap(); @@ -453,18 +493,62 @@ mod tests { let mut state = BridgeState { last_message_id: Some(1), + last_rx_time: Some(100), + last_rx_time_ids: vec![1], last_checked_at: None, }; - poll_once(&potato, &matrix, &mut state, state_str, 123).await; + poll_once(&potato, &matrix, &mut state, state_str).await; mock_msgs.assert(); - // Should have advanced checkpoint and saved it. - assert_eq!(state.last_checked_at, Some(123)); + // No new data means state remains unchanged and is not persisted. + assert_eq!(state.last_rx_time, Some(100)); + assert_eq!(state.last_rx_time_ids, vec![1]); + assert!(!state_path.exists()); + } + + #[tokio::test] + async fn poll_once_persists_state_for_non_text_messages() { + let tmp_dir = tempfile::tempdir().unwrap(); + let state_path = tmp_dir.path().join("state.json"); + let state_str = state_path.to_str().unwrap(); + + let mut server = mockito::Server::new_async().await; + let mock_msgs = server + .mock("GET", "/api/messages") + .match_query(mockito::Matcher::Any) + .with_status(200) + .with_header("content-type", "application/json") + .with_body( + r#"[{"id":1,"rx_time":100,"rx_iso":"2025-11-27T00:00:00Z","from_id":"!abcd1234","to_id":"^all","channel":1,"portnum":"POSITION_APP","text":"","rssi":-100,"hop_limit":1,"lora_freq":868,"modem_preset":"MediumFast","channel_name":"TEST","snr":0.0,"node_id":"!abcd1234"}]"#, + ) + .create(); + + let http_client = reqwest::Client::new(); + let potatomesh_cfg = PotatomeshConfig { + base_url: server.url(), + poll_interval_secs: 1, + }; + let matrix_cfg = MatrixConfig { + homeserver: server.url(), + as_token: "AS_TOKEN".to_string(), + server_name: "example.org".to_string(), + room_id: "!roomid:example.org".to_string(), + }; + + let potato = PotatoClient::new(http_client.clone(), potatomesh_cfg); + let matrix = MatrixAppserviceClient::new(http_client, matrix_cfg); + let mut state = BridgeState::default(); + + poll_once(&potato, &matrix, &mut state, state_str).await; + + mock_msgs.assert(); + assert!(state_path.exists()); let loaded = BridgeState::load(state_str).unwrap(); - assert_eq!(loaded.last_checked_at, Some(123)); assert_eq!(loaded.last_message_id, Some(1)); + assert_eq!(loaded.last_rx_time, Some(100)); + assert_eq!(loaded.last_rx_time_ids, vec![1]); } #[tokio::test]