hive-matrix-mcp: sync with sliding sync when the server has it, classic /sync otherwise

hive-matrix-mcp/src/sliding.rs

@@ -0,0 +1,155 @@
+//! The sliding-sync (MSC4186) sync loop, the cheap alternative to classic
+//! `/sync`: one growing list over every joined room, the newest event per
+//! room and a minimal set of state, instead of up to 20 events and the state
+//! delta of every room that moved. It fills the same matrix-sdk store, so the
+//! unread check (the agent's own receipt plus a live `/messages`) is
+//! unchanged.
+//!
+//! [`wanted`] decides whether to use it; `main`'s loop falls back to `/sync`
+//! when it is not wanted or [`run`] gives up.
+
+use std::future::Future;
+use std::time::Duration;
+
+use anyhow::{Result, bail};
+use futures_util::StreamExt as _;
+use matrix_sdk::Client;
+use matrix_sdk::ruma::events::StateEventType;
+use matrix_sdk::sliding_sync::{SlidingSyncList, SlidingSyncMode, Version};
+
+/// `classic` forces `/sync`; anything else (or unset) lets the server's
+/// advertised support decide.
+const MODE_ENV: &str = "HIVE_MATRIX_SYNC";
+
+/// Rooms asked for per growing-list step. The list grows until it covers every
+/// joined room, so this only sets how many rooms the first responses carry.
+const BATCH: u32 = 50;
+
+/// Consecutive failed polls after which [`run`] gives up and the caller falls
+/// back to `/sync`.
+const MAX_CONSECUTIVE_ERRORS: u32 = 5;
+
+/// Pause after a failed poll, so a dead server is not hammered.
+const ERROR_PAUSE: Duration = Duration::from_secs(5);
+
+/// Whether this account should sync with sliding sync: the server advertises
+/// MSC4186 and the operator has not forced `/sync`.
+pub async fn wanted(client: &Client) -> bool {
+    let forced_classic = std::env::var(MODE_ENV).is_ok_and(|v| v.eq_ignore_ascii_case("classic"));
+    let advertised = client
+        .available_sliding_sync_versions()
+        .await
+        .iter()
+        .any(|v| matches!(v, Version::Native));
+    decide(forced_classic, advertised)
+}
+
+fn decide(forced_classic: bool, advertised: bool) -> bool {
+    advertised && !forced_classic
+}
+
+/// The state the daemon reads from a room: its name, whether it is
+/// encrypted, who is in it (lazily) and the power levels sends depend on.
+fn required_state() -> Vec<(StateEventType, String)> {
+    [
+        (StateEventType::RoomName, ""),
+        (StateEventType::RoomCanonicalAlias, ""),
+        (StateEventType::RoomEncryption, ""),
+        (StateEventType::RoomCreate, ""),
+        (StateEventType::RoomPowerLevels, ""),
+        (StateEventType::RoomMember, "$LAZY"),
+        (StateEventType::RoomMember, "$ME"),
+    ]
+    .into_iter()
+    .map(|(t, k)| (t, k.to_owned()))
+    .collect()
+}
+
+/// Poll sliding sync for `client`, calling `after_update` once per response.
+/// Returns `Err` once 5 polls in a row have failed, so the caller can fall
+/// back to `/sync`.
+pub async fn run<F, Fut>(client: &Client, after_update: F) -> Result<()>
+where
+    F: Fn() -> Fut,
+    Fut: Future<Output = ()>,
+{
+    let list = SlidingSyncList::builder("all-rooms")
+        .sync_mode(SlidingSyncMode::new_growing(BATCH))
+        .timeline_limit(1)
+        .required_state(required_state());
+    let sync = client
+        .sliding_sync("hive-matrix")?
+        .version(Version::Native)
+        .with_all_extensions()
+        .add_list(list)
+        .build()
+        .await?;
+    let stream = sync.sync();
+    futures_util::pin_mut!(stream);
+    let mut errors = ErrorRun::default();
+    while let Some(update) = stream.next().await {
+        match update {
+            Ok(_) => {
+                errors.ok();
+                after_update().await;
+            }
+            Err(e) => {
+                tracing::warn!(error = %e, "sliding sync poll failed");
+                if errors.failed() {
+                    bail!("sliding sync failed {MAX_CONSECUTIVE_ERRORS} polls in a row: {e}");
+                }
+                tokio::time::sleep(ERROR_PAUSE).await;
+            }
+        }
+    }
+    bail!("sliding sync stream ended")
+}
+
+/// Counts failed polls in a row; a good poll resets it.
+#[derive(Default)]
+struct ErrorRun(u32);
+
+impl ErrorRun {
+    fn ok(&mut self) {
+        self.0 = 0;
+    }
+
+    /// Records a failure; true once there are [`MAX_CONSECUTIVE_ERRORS`] in a row.
+    fn failed(&mut self) -> bool {
+        self.0 += 1;
+        self.0 >= MAX_CONSECUTIVE_ERRORS
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{ErrorRun, MAX_CONSECUTIVE_ERRORS, decide, required_state};
+    use matrix_sdk::ruma::events::StateEventType;
+
+    #[test]
+    fn sliding_sync_needs_server_support_and_no_opt_out() {
+        assert!(decide(false, true));
+        assert!(!decide(false, false), "a server without MSC4186 gets /sync");
+        assert!(!decide(true, true), "the operator can force /sync");
+    }
+
+    #[test]
+    fn gives_up_only_after_a_run_of_failures() {
+        let mut run = ErrorRun::default();
+        for _ in 1..MAX_CONSECUTIVE_ERRORS {
+            assert!(!run.failed());
+        }
+        run.ok();
+        for _ in 1..MAX_CONSECUTIVE_ERRORS {
+            assert!(!run.failed(), "a good poll resets the count");
+        }
+        assert!(run.failed());
+    }
+
+    #[test]
+    fn the_room_state_asks_for_encryption_and_lazy_members() {
+        let state = required_state();
+        assert!(state.contains(&(StateEventType::RoomEncryption, String::new())));
+        assert!(state.contains(&(StateEventType::RoomMember, "$LAZY".to_owned())));
+    }
+}