diff --git a/Cargo.lock b/Cargo.lock index 634cd3c22..f4c4d4c3f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1967,6 +1967,7 @@ dependencies = [ "futures-util", "hive-sock-client", "matrix-sdk", + "matrix-sdk-base", "mime", "mime_guess", "reqwest", @@ -1976,6 +1977,7 @@ dependencies = [ "serde_json", "swarm-matrix-client", "swarm-secret-client", + "tempfile", "tokio", "tracing", "tracing-subscriber", diff --git a/hive-matrix-mcp/Cargo.toml b/hive-matrix-mcp/Cargo.toml index 31f3a9b39..96bd9cb72 100644 --- a/hive-matrix-mcp/Cargo.toml +++ b/hive-matrix-mcp/Cargo.toml @@ -14,6 +14,10 @@ clap.workspace = true futures-util.workspace = true hive-sock-client.workspace = true matrix-sdk.workspace = true +# The crypto store trait and its `Changes`, for the to-device token reset in `sliding`. +matrix-sdk-base = { version = "0.18", default-features = false, features = [ + "e2e-encryption", +] } mime = "0.3" mime_guess = "2" reqwest.workspace = true @@ -35,4 +39,5 @@ name = "hive-matrix-daemon" path = "src/main.rs" [dev-dependencies] +tempfile = "3" tokio = { workspace = true, features = ["test-util"] } diff --git a/hive-matrix-mcp/src/client.rs b/hive-matrix-mcp/src/client.rs index f128c9b6c..7436181bb 100644 --- a/hive-matrix-mcp/src/client.rs +++ b/hive-matrix-mcp/src/client.rs @@ -137,6 +137,7 @@ pub async fn build_and_restore( fs::create_dir_all(state_dir) .await .with_context(|| format!("mkdir matrix state dir {}", state_dir.display()))?; + crate::sliding::reset_classic_to_device_token(state_dir).await; let client = Client::builder() .homeserver_url(homeserver) .sqlite_store(state_dir, None) diff --git a/hive-matrix-mcp/src/sliding.rs b/hive-matrix-mcp/src/sliding.rs index fd62a9371..a48407765 100644 --- a/hive-matrix-mcp/src/sliding.rs +++ b/hive-matrix-mcp/src/sliding.rs @@ -11,11 +11,14 @@ use std::future::Future; use std::time::Duration; -use anyhow::{Result, bail}; +use std::path::Path; + +use anyhow::{Result, anyhow, bail}; use futures_util::StreamExt as _; -use matrix_sdk::Client; 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. @@ -48,6 +51,53 @@ 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 { + 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)> { @@ -127,8 +177,47 @@ impl ErrorRun { #[cfg(test)] mod tests { - use super::{ErrorRun, MAX_CONSECUTIVE_ERRORS, decide, required_state}; + 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 { + 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() {