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 std::path::Path;

use anyhow::{Result, anyhow, bail};
use futures_util::StreamExt as _;
use matrix_sdk::ruma::OwnedRoomId;
use matrix_sdk::ruma::events::StateEventType;
use matrix_sdk::sliding_sync::{SlidingSyncList, SlidingSyncMode, Version};
use matrix_sdk::{Client, SqliteCryptoStore};
use matrix_sdk_base::crypto::store::{CryptoStore, types::Changes};

/// `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 to-device token a classic `/sync` leaves in the crypto store cannot start
/// a sliding sync: the sdk sends the store's token as `extensions.to_device.since`
/// and the homeserver refuses anything but an integer (400, "should look like an
/// int"), which sent every start back to `/sync`, whose own token is then in the
/// store again. This puts `0` in its place (the server redelivers only the
/// to-device events nobody acknowledged), so the first sliding poll is valid.
/// A token that already is an integer, and a store with none, are left alone.
/// Best effort: a failure is logged and the sliding sync tries as before.
///
/// Call it before the client opens the store.
pub async fn reset_classic_to_device_token(state_dir: &Path) {
    let result = async {
        let store = SqliteCryptoStore::open(state_dir, None).await?;
        reset_token(&store).await
    }
    .await;
    match result {
        Ok(true) => {
            tracing::info!(
                "reset the crypto store's /sync token so sliding sync can start from it"
            );
        }
        Ok(false) => {}
        Err(e) => tracing::warn!(error = %e, "could not check the crypto store's to-device token"),
    }
}

/// Whether `store` held a token that is not an integer, now replaced by `0`.
async fn reset_token(store: &impl CryptoStore) -> Result<bool> {
    let held = store
        .next_batch_token()
        .await
        .map_err(|e| anyhow!("{e:?}"))?;
    if held.is_none_or(|t| !t.is_empty() && t.bytes().all(|b| b.is_ascii_digit())) {
        return Ok(false);
    }
    let reset = Changes {
        next_batch_token: Some("0".to_owned()),
        ..Changes::default()
    };
    store
        .save_changes(reset)
        .await
        .map_err(|e| anyhow!("{e:?}"))?;
    Ok(true)
}

/// 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 with
/// the rooms that response touched.
/// 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(Vec<OwnedRoomId>) -> 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(summary) => {
                errors.ok();
                after_update(summary.rooms).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, reset_classic_to_device_token,
    };
    use matrix_sdk::SqliteCryptoStore;
    use matrix_sdk::ruma::events::StateEventType;
    use matrix_sdk_base::crypto::store::{CryptoStore, types::Changes};

    async fn token_after_reset(before: Option<&str>) -> Option<String> {
        let dir = tempfile::tempdir().expect("tempdir");
        let store = SqliteCryptoStore::open(dir.path(), None)
            .await
            .expect("open");
        if let Some(token) = before {
            let seeded = Changes {
                next_batch_token: Some(token.to_owned()),
                ..Changes::default()
            };
            store.save_changes(seeded).await.expect("seed");
        }
        drop(store);
        reset_classic_to_device_token(dir.path()).await;
        let store = SqliteCryptoStore::open(dir.path(), None)
            .await
            .expect("reopen");
        store.next_batch_token().await.expect("read")
    }

    #[tokio::test]
    async fn a_classic_sync_token_is_replaced_by_zero() {
        let classic = "s251551370_223970929_3253937_0_41771_2_1_1_1";
        assert_eq!(token_after_reset(Some(classic)).await.as_deref(), Some("0"));
    }

    #[tokio::test]
    async fn an_integer_token_and_an_empty_store_are_left_alone() {
        assert_eq!(
            token_after_reset(Some("41771")).await.as_deref(),
            Some("41771")
        );
        assert_eq!(token_after_reset(None).await, None);
    }

    #[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())));
    }
}