hyperhive/hive-matrix-mcp/src/sliding.rs

//! 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()
        // Typing notices are never read and fire constantly in a busy room.
        // After `with_all_extensions`: "without" only resets the setting to
        // unset, and that call would fill an unset one back in.
        .without_typing_extension()
        .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())));
    }
}