matrix-rust-sdk

public

Star0

Fork 0

Watch0

8 commits 4 branches 0 tags 0 releases16.6 MiB

Commit `7aace883fbe0`

tarball

atagen <boss@atagen.co>2026-10-24 23:45 UTC

11 files changed+483-4

sliding-sync: fence response checkpoints on durable commits

Event handlers discard returned errors and detached response processing survives caller cancellation. Durable consumers need a fallible acknowledgement before response publication and checkpoint advancement.

Modified file: crates/matrix-sdk-ui/src/room_list_service/mod.rs+27 additions−1 deletions@@ -63,7 +63,9 @@ use eyeball::Subscriber;
 use futures_util::{Stream, StreamExt, pin_mut};
 use matrix_sdk::{
     Client, Error as SlidingSyncError, Room, SlidingSync, SlidingSyncList, SlidingSyncMode,
-    event_cache::EventCacheError, sliding_sync::PollTimeout, timeout::timeout,
+    event_cache::EventCacheError,
+    sliding_sync::{PollTimeout, ResponseCommitHook},
+    timeout::timeout,
 };
 pub use room_list::*;
 use ruma::{
@@ -176,6 +178,26 @@ impl RoomListService {
         timeline_limit: u32,
         extra_required_state: Vec<(StateEventType, String)>,
     ) -> Result<Self, Error> {
+        Self::new_with_commit_hook(
+            client,
+            share_pos,
+            connection_id,
+            timeline_limit,
+            extra_required_state,
+            None,
+        )
+        .await
+    }
+
+    #[doc = "The hook applies only to this room list connection, before checkpoint advancement."]
+    pub async fn new_with_commit_hook(
+        client: Client,
+        share_pos: bool,
+        connection_id: &str,
+        timeline_limit: u32,
+        extra_required_state: Vec<(StateEventType, String)>,
+        response_commit_hook: Option<ResponseCommitHook>,
+    ) -> Result<Self, Error> {
         let mut builder = client
             .sliding_sync(connection_id)
             .map_err(Error::SlidingSync)?
@@ -190,6 +212,10 @@ impl RoomListService {
                 enabled: Some(true),
             }));
 
+        if let Some(hook) = response_commit_hook {
+            builder = builder.with_response_commit_hook(hook);
+        }
+
         match client.enabled_thread_subscriptions().await {
             Ok(true) => {
                 debug!("Client requested thread subscriptions extension");

Modified file: crates/matrix-sdk-ui/src/sync_service.rs+12 additions−1 deletions@@ -38,6 +38,7 @@ use matrix_sdk::{
     executor::{JoinHandle, spawn},
     ruma::events::StateEventType,
     sleep::sleep,
+    sliding_sync::ResponseCommitHook,
 };
 use thiserror::Error;
 use tokio::sync::{
@@ -763,6 +764,7 @@ impl SyncService {
 
 #[derive(Clone)]
 pub struct SyncServiceBuilder {
+    room_list_response_commit_hook: Option<ResponseCommitHook>,
     /// SDK client.
     client: Client,
 
@@ -800,6 +802,7 @@ pub struct SyncServiceBuilder {
 impl SyncServiceBuilder {
     fn new(client: Client) -> Self {
         Self {
+            room_list_response_commit_hook: None,
             client,
             with_offline_mode: false,
             with_share_pos: true,
@@ -839,6 +842,12 @@ impl SyncServiceBuilder {
         self
     }
 
+    #[doc = "Notification lookups and encryption sync do not inherit this room list commit hook."]
+    pub fn with_room_list_response_commit_hook(mut self, hook: ResponseCommitHook) -> Self {
+        self.room_list_response_commit_hook = Some(hook);
+        self
+    }
+
     /// Request these state events for every room the room list service
     /// syncs, on top of its defaults. Use it for state the sdk doesn't know
     /// about, so it lands in the state store like the rest.
@@ -864,6 +873,7 @@ impl SyncServiceBuilder {
     /// as the sliding syncs are supposed to run.
     pub async fn build(self) -> Result<SyncService, Error> {
         let Self {
+            room_list_response_commit_hook,
             client,
             with_offline_mode,
             with_share_pos,
@@ -875,12 +885,13 @@ impl SyncServiceBuilder {
 
         let encryption_sync_permit = Arc::new(AsyncMutex::new(EncryptionSyncPermit::new()));
 
-        let room_list = RoomListService::new_with_required_state(
+        let room_list = RoomListService::new_with_commit_hook(
             client.clone(),
             with_share_pos,
             &room_list_conn_id,
             room_list_timeline_limit,
             room_list_extra_required_state,
+            room_list_response_commit_hook,
         )
         .await?;
 

Added file: crates/matrix-sdk-ui/tests/response_commit.rs+64 additions@@ -0,0 +1,64 @@
+#![cfg(not(target_family = "wasm"))]
+
+use std::{
+    sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    },
+    time::Duration,
+};
+
+use futures_util::{StreamExt, pin_mut};
+use matrix_sdk::{
+    ruma::{api::client::sync::sync_events::v5, room_id},
+    sliding_sync::ResponseCommitHook,
+    test_utils::mocks::MatrixMockServer,
+};
+use matrix_sdk_ui::sync_service::SyncService;
+use serde_json::json;
+
+#[tokio::test]
+async fn sync_service_passes_commit_hook_only_to_its_room_list_connection() {
+    let server = MatrixMockServer::new().await;
+    let client = server.client_builder().build().await;
+    let commits = Arc::new(AtomicUsize::new(0));
+    let hook = ResponseCommitHook::new({
+        let commits = commits.clone();
+        move |batch| {
+            let commits = commits.clone();
+            async move {
+                assert_eq!(batch.connection_id, "mentions-list");
+                assert!(batch.raw_rooms.contains_key(room_id!("!room:localhost")));
+                commits.fetch_add(1, Ordering::SeqCst);
+                Err("store unavailable".to_owned())
+            }
+        }
+    });
+    let service = SyncService::builder(client.clone())
+        .with_room_list_conn_id("mentions-list".to_owned())
+        .with_room_list_response_commit_hook(hook)
+        .build()
+        .await
+        .unwrap();
+    let mut response = v5::Response::new("position-1".to_owned());
+    response.rooms.insert(
+        room_id!("!room:localhost").to_owned(),
+        serde_json::from_value(json!({ "initial": true, "required_state": [], "timeline": [] }))
+            .unwrap(),
+    );
+    server.mock_sliding_sync().ok(response).mount().await;
+    let rooms = service.room_list_service();
+    let updates = rooms.sync();
+    pin_mut!(updates);
+    let error = tokio::time::timeout(Duration::from_secs(5), updates.next())
+        .await
+        .unwrap()
+        .unwrap()
+        .unwrap_err();
+    assert!(error.to_string().contains("store unavailable"));
+    assert_eq!(commits.load(Ordering::SeqCst), 1);
+
+    let independent = client.sliding_sync("independent").unwrap().build().await.unwrap();
+    tokio::time::timeout(Duration::from_secs(5), independent.sync_once()).await.unwrap().unwrap();
+    assert_eq!(commits.load(Ordering::SeqCst), 1);
+}

Modified file: crates/matrix-sdk/Cargo.toml+4 additions@@ -205,5 +205,9 @@ webpki-roots = "1.0.6"
 name = "integration"
 required-features = ["testing"]
 
+[[test]]
+name = "response_commit"
+required-features = ["testing", "e2e-encryption"]
+
 [lints]
 workspace = true

Modified file: crates/matrix-sdk/src/sliding_sync/builder.rs+11 additions−2 deletions@@ -11,8 +11,8 @@ use ruma::{OwnedRoomId, api::client::sync::sync_events::v5 as http};
 use tokio::sync::{Mutex as AsyncMutex, RwLock as AsyncRwLock, broadcast::channel};
 
 use super::{
-    Error, SlidingSync, SlidingSyncInner, SlidingSyncListBuilder, SlidingSyncPositionMarkers,
-    Version, cache::format_storage_key_prefix,
+    Error, ResponseCommitHook, SlidingSync, SlidingSyncInner, SlidingSyncListBuilder,
+    SlidingSyncPositionMarkers, Version, cache::format_storage_key_prefix,
 };
 use crate::{Client, Result};
 
@@ -22,6 +22,7 @@ use crate::{Client, Result};
 /// [`crate::SlidingSync::builder`].
 #[derive(Debug, Clone)]
 pub struct SlidingSyncBuilder {
+    response_commit_hook: Option<ResponseCommitHook>,
     id: String,
     storage_key: String,
     version: Option<Version>,
@@ -36,6 +37,12 @@ pub struct SlidingSyncBuilder {
 }
 
 impl SlidingSyncBuilder {
+    #[doc = "Fences this connection's checkpoint until durable ingestion acknowledges the response."]
+    pub fn with_response_commit_hook(mut self, hook: ResponseCommitHook) -> Self {
+        self.response_commit_hook = Some(hook);
+        self
+    }
+
     pub(super) fn new(id: String, client: Client) -> Result<Self, Error> {
         if id.len() > 16 {
             Err(Error::InvalidSlidingSyncIdentifier)
@@ -44,6 +51,7 @@ impl SlidingSyncBuilder {
                 format_storage_key_prefix(&id, client.user_id().ok_or(Error::UnauthenticatedUser)?);
 
             Ok(Self {
+                response_commit_hook: None,
                 id,
                 storage_key,
                 version: None,
@@ -278,6 +286,7 @@ impl SlidingSyncBuilder {
         let lists = AsyncRwLock::new(lists);
 
         Ok(SlidingSync::new(SlidingSyncInner {
+            response_commit_hook: self.response_commit_hook,
             id: self.id,
 
             client,

Modified file: crates/matrix-sdk/src/sliding_sync/client.rs+6 additions@@ -257,6 +257,12 @@ impl SlidingSyncResponseProcessor {
         Ok(())
     }
 
+    pub fn response(&self) -> SyncResponse {
+        let mut response = self.response.clone().unwrap_or_default();
+        response.to_device.extend(self.to_device_events.clone());
+        response
+    }
+
     pub async fn process_and_take_response(mut self) -> Result<SyncResponse> {
         let mut response = self.response.take().unwrap_or_default();
 

Added file: crates/matrix-sdk/src/sliding_sync/commit.md+20 additions@@ -0,0 +1,20 @@
+A durable consumer can fence a Sliding Sync checkpoint with this hook.
+
+The SDK calls the hook after processing room state and releasing the state
+store lock, before response handlers and observers receive the response.
+Room state may already be persisted. The hook must tolerate replay of the
+same events after an unsuccessful commit and use stable event keys.
+
+Return success only after the consumer's durable transaction commits.
+Return an error when storage fails or a store acknowledgement is dropped.
+An error prevents position advancement and response handler publication.
+Other room-state notifications may already have been emitted during processing.
+
+Response processing survives cancellation of the calling sync task. The
+consumer must arrange for shutdown to drop or reject outstanding store
+acknowledgements so the hook can finish with an error. Cancelling the caller
+alone does not cancel the hook.
+
+The hook belongs to one Sliding Sync instance. Separate notification and
+encryption connections do not inherit it. Native callbacks and futures must
+be sendable. Browser callbacks and futures may use local state.

Added file: crates/matrix-sdk/src/sliding_sync/commit.rs+62 additions@@ -0,0 +1,62 @@
+use std::{collections::BTreeMap, fmt, future::Future, sync::Arc};
+
+use matrix_sdk_base::sync::SyncResponse;
+use matrix_sdk_common::{BoxFuture, SendOutsideWasm, SyncOutsideWasm};
+use ruma::{OwnedRoomId, api::client::sync::sync_events::v5 as http};
+
+#[doc = include_str!("commit.md")]
+#[derive(Clone)]
+pub struct ResponseCommitHook(Arc<CommitFn>);
+
+#[cfg(not(target_family = "wasm"))]
+type CommitFn = dyn Fn(ResponseCommitBatch) -> BoxFuture<'static, Result<(), String>> + Send + Sync;
+#[cfg(target_family = "wasm")]
+type CommitFn = dyn Fn(ResponseCommitBatch) -> BoxFuture<'static, Result<(), String>>;
+
+#[doc = "The processed response and original room payloads share one checkpoint boundary."]
+#[derive(Clone)]
+pub struct ResponseCommitBatch {
+    #[doc = "Identifies the Sliding Sync connection whose checkpoint is fenced."]
+    pub connection_id: String,
+    #[doc = "The checkpoint remains here if the commit fails."]
+    pub previous_pos: Option<String>,
+    #[doc = "The checkpoint becomes eligible for persistence after acknowledgement."]
+    pub next_pos: String,
+    #[doc = "Contains processed events before response handlers and observers receive them."]
+    pub response: SyncResponse,
+    #[doc = "Preserves original encrypted candidates, initial flags, and gap metadata for replay."]
+    pub raw_rooms: BTreeMap<OwnedRoomId, http::response::Room>,
+}
+
+impl fmt::Debug for ResponseCommitBatch {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter
+            .debug_struct("ResponseCommitBatch")
+            .field("connection_id", &self.connection_id)
+            .field("previous_pos", &self.previous_pos)
+            .field("next_pos", &self.next_pos)
+            .field("room_count", &self.raw_rooms.len())
+            .finish_non_exhaustive()
+    }
+}
+
+impl ResponseCommitHook {
+    #[doc = "A failed or dropped store acknowledgement must produce an error from this future."]
+    pub fn new<F, Fut>(commit: F) -> Self
+    where
+        F: Fn(ResponseCommitBatch) -> Fut + SendOutsideWasm + SyncOutsideWasm + 'static,
+        Fut: Future<Output = Result<(), String>> + SendOutsideWasm + 'static,
+    {
+        Self(Arc::new(move |batch| Box::pin(commit(batch))))
+    }
+
+    pub(super) async fn commit(&self, batch: ResponseCommitBatch) -> crate::Result<()> {
+        (self.0)(batch).await.map_err(|error| super::Error::ResponseCommitFailed(error).into())
+    }
+}
+
+impl fmt::Debug for ResponseCommitHook {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        formatter.debug_struct("ResponseCommitHook").finish_non_exhaustive()
+    }
+}

Modified file: crates/matrix-sdk/src/sliding_sync/error.rs+4 additions@@ -7,6 +7,10 @@ use thiserror::Error;
 #[derive(Error, Debug)]
 #[non_exhaustive]
 pub enum Error {
+    #[doc = "Durable ingestion failed before checkpoint advancement."]
+    #[error("Sliding sync response commit failed, {0}")]
+    ResponseCommitFailed(String),
+
     /// Sliding sync has been configured with a missing version. See
     /// [`crate::sliding_sync::Version`].
     #[error("Sliding sync version is missing")]

Modified file: crates/matrix-sdk/src/sliding_sync/mod.rs+13 additions@@ -18,6 +18,7 @@
 mod builder;
 mod cache;
 mod client;
+mod commit;
 mod error;
 mod list;
 
@@ -31,6 +32,7 @@ use std::{
 
 use async_stream::stream;
 pub use client::{Version, VersionBuilder};
+pub use commit::{ResponseCommitBatch, ResponseCommitHook};
 use futures_core::stream::Stream;
 use matrix_sdk_base::RequestedRequiredStates;
 #[cfg(feature = "e2e-encryption")]
@@ -62,6 +64,7 @@ pub struct SlidingSync {
 
 #[derive(Debug)]
 pub(super) struct SlidingSyncInner {
+    response_commit_hook: Option<ResponseCommitHook>,
     /// A unique identifier for this instance of sliding sync.
     ///
     /// Used to distinguish different connections to sliding sync.
@@ -331,6 +334,16 @@ impl SlidingSync {
             };
 
             // Release the lock before calling event handlers
+            if let Some(hook) = &self.inner.response_commit_hook {
+                hook.commit(ResponseCommitBatch {
+                    connection_id: self.inner.id.clone(),
+                    previous_pos: position.pos.clone(),
+                    next_pos: sliding_sync_response.pos.clone(),
+                    response: response_processor.response(),
+                    raw_rooms: sliding_sync_response.rooms.clone(),
+                })
+                .await?;
+            }
             response_processor.process_and_take_response().await?
         };
 

Added file: crates/matrix-sdk/tests/response_commit.rs+260 additions@@ -0,0 +1,260 @@
+#![cfg(not(target_family = "wasm"))]
+
+use std::{
+    collections::HashSet,
+    sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    },
+    time::Duration,
+};
+
+use matrix_sdk::{
+    Client, SlidingSync, SlidingSyncList, SlidingSyncMode,
+    ruma::{api::client::sync::sync_events::v5, events::AnySyncTimelineEvent, room_id, serde::Raw},
+    sliding_sync::{ResponseCommitBatch, ResponseCommitHook},
+    test_utils::mocks::MatrixMockServer,
+};
+use serde_json::{Value, json};
+use tokio::sync::{broadcast::error::TryRecvError, mpsc, oneshot};
+
+type PendingCommit = (ResponseCommitBatch, oneshot::Sender<Result<(), String>>);
+
+fn queued_hook() -> (ResponseCommitHook, mpsc::UnboundedReceiver<PendingCommit>) {
+    let (sender, receiver) = mpsc::unbounded_channel();
+    let hook = ResponseCommitHook::new(move |batch| {
+        let sender = sender.clone();
+        async move {
+            let (acknowledge, acknowledgement) = oneshot::channel();
+            sender.send((batch, acknowledge)).map_err(|_| "store stopped".to_owned())?;
+            acknowledgement.await.map_err(|_| "store acknowledgement dropped".to_owned())?
+        }
+    });
+    (hook, receiver)
+}
+
+async fn next_commit(receiver: &mut mpsc::UnboundedReceiver<PendingCommit>) -> PendingCommit {
+    tokio::time::timeout(Duration::from_secs(5), receiver.recv()).await.unwrap().unwrap()
+}
+
+async fn fixture(hook: Option<ResponseCommitHook>) -> (MatrixMockServer, Client, SlidingSync) {
+    let server = MatrixMockServer::new().await;
+    let client = server.client_builder().build().await;
+    let mut response = v5::Response::new("position-1".to_owned());
+    response.rooms.insert(
+        room_id!("!commit:localhost").to_owned(),
+        serde_json::from_value(json!({
+            "initial": true,
+            "limited": true,
+            "prev_batch": "history-token",
+            "required_state": [],
+            "timeline": [{
+                "type": "m.room.message",
+                "event_id": "$mention",
+                "sender": "@sender:localhost",
+                "origin_server_ts": 1,
+                "content": {
+                    "msgtype": "m.text",
+                    "body": "durable mention",
+                    "m.mentions": { "user_ids": [client.user_id().unwrap()] }
+                }
+            }, {
+                "type": "m.room.encrypted",
+                "event_id": "$encrypted-candidate",
+                "sender": "@sender:localhost",
+                "origin_server_ts": 2,
+                "content": {
+                    "algorithm": "m.megolm.v1.aes-sha2",
+                    "ciphertext": "unavailable",
+                    "sender_key": "unavailable",
+                    "session_id": "missing-session",
+                    "device_id": "SENDER"
+                }
+            }]
+        }))
+        .unwrap(),
+    );
+    server.mock_sliding_sync().ok(response).mount().await;
+    let mut builder = client.sliding_sync("commit-test").unwrap().share_pos().add_list(
+        SlidingSyncList::builder("rooms")
+            .sync_mode(SlidingSyncMode::new_selective().add_range(0..=0)),
+    );
+    if let Some(hook) = hook {
+        builder = builder.with_response_commit_hook(hook);
+    }
+    let sync = builder.build().await.unwrap();
+    (server, client, sync)
+}
+
+async fn saved_position(client: &Client) -> Option<String> {
+    let machine = client.olm_machine_for_testing().await;
+    let key = format!("sliding_sync_store::commit-test::{}::instance", client.user_id().unwrap());
+    let saved = machine.as_ref().unwrap().store().get_custom_value(&key).await.unwrap()?;
+    serde_json::from_slice::<Value>(&saved).unwrap()["pos"].as_str().map(ToOwned::to_owned)
+}
+
+#[tokio::test]
+async fn acknowledgement_fences_handlers_observers_and_checkpoint() {
+    let (hook, mut receiver) = queued_hook();
+    let (_server, client, sync) = fixture(Some(hook)).await;
+    let handled = Arc::new(AtomicUsize::new(0));
+    client.add_event_handler({
+        let handled = handled.clone();
+        move |_event: Raw<AnySyncTimelineEvent>| {
+            let handled = handled.clone();
+            async move {
+                handled.fetch_add(1, Ordering::SeqCst);
+            }
+        }
+    });
+    let mut updates = client.subscribe_to_all_room_updates();
+    let processing = tokio::spawn(async move { sync.sync_once().await });
+    let (batch, acknowledgement) = next_commit(&mut receiver).await;
+    assert_eq!(batch.connection_id, "commit-test");
+    assert!(batch.previous_pos.is_none());
+    assert_eq!(batch.next_pos, "position-1");
+    let room_id = room_id!("!commit:localhost");
+    assert_eq!(batch.response.rooms.joined[room_id].timeline.events.len(), 2);
+    let raw = serde_json::to_value(&batch.raw_rooms[room_id]).unwrap();
+    assert_eq!(raw["initial"], true);
+    assert_eq!(raw["limited"], true);
+    assert_eq!(raw["prev_batch"], "history-token");
+    assert_eq!(raw["timeline"][1]["type"], "m.room.encrypted");
+    assert!(saved_position(&client).await.is_none());
+    assert_eq!(handled.load(Ordering::SeqCst), 0);
+    assert!(matches!(updates.try_recv(), Err(TryRecvError::Empty)));
+    acknowledgement.send(Ok(())).unwrap();
+    processing.await.unwrap().unwrap();
+    assert_eq!(saved_position(&client).await.as_deref(), Some("position-1"));
+    assert_eq!(handled.load(Ordering::SeqCst), 2);
+    assert!(updates.try_recv().is_ok());
+}
+
+#[tokio::test]
+async fn failed_commit_replays_events_after_room_state_was_saved() {
+    let (hook, mut receiver) = queued_hook();
+    let (_server, client, sync) = fixture(Some(hook)).await;
+    let first_sync = sync.clone();
+    let processing = tokio::spawn(async move { first_sync.sync_once().await });
+    let (first, acknowledgement) = next_commit(&mut receiver).await;
+    assert!(client.get_room(room_id!("!commit:localhost")).is_some());
+    acknowledgement.send(Err("disk full".to_owned())).unwrap();
+    let error = processing.await.unwrap().unwrap_err();
+    assert!(error.to_string().contains("disk full"));
+    assert!(saved_position(&client).await.is_none());
+
+    let processing = tokio::spawn(async move { sync.sync_once().await });
+    let (replay, acknowledgement) = next_commit(&mut receiver).await;
+    assert_eq!(first.previous_pos, replay.previous_pos);
+    let room_id = room_id!("!commit:localhost");
+    let keys = |batch: &ResponseCommitBatch| {
+        batch.response.rooms.joined[room_id]
+            .timeline
+            .events
+            .iter()
+            .map(|event| (room_id.to_owned(), event.event_id().unwrap()))
+            .collect::<HashSet<_>>()
+    };
+    assert_eq!(keys(&first).len(), 2);
+    assert_eq!(keys(&first), keys(&replay));
+    acknowledgement.send(Ok(())).unwrap();
+    processing.await.unwrap().unwrap();
+    assert_eq!(saved_position(&client).await.as_deref(), Some("position-1"));
+}
+
+#[tokio::test]
+async fn dropped_acknowledgement_fails_without_publishing_or_saving_position() {
+    let (hook, mut receiver) = queued_hook();
+    let (_server, client, sync) = fixture(Some(hook)).await;
+    let mut updates = client.subscribe_to_all_room_updates();
+    let processing = tokio::spawn(async move { sync.sync_once().await });
+    let (_, acknowledgement) = next_commit(&mut receiver).await;
+    drop(acknowledgement);
+    assert!(processing.await.unwrap().unwrap_err().to_string().contains("acknowledgement dropped"));
+    assert!(saved_position(&client).await.is_none());
+    assert!(matches!(updates.try_recv(), Err(TryRecvError::Empty)));
+}
+
+#[tokio::test]
+async fn cancelled_caller_and_dropped_acknowledgement_preserve_checkpoint_for_retry() {
+    let (hook, mut receiver) = queued_hook();
+    let (_server, client, sync) = fixture(Some(hook)).await;
+    let first_sync = sync.clone();
+    let processing = tokio::spawn(async move { first_sync.sync_once().await });
+    let (_, acknowledgement) = next_commit(&mut receiver).await;
+    processing.abort();
+    assert!(processing.await.unwrap_err().is_cancelled());
+    drop(acknowledgement);
+
+    let processing = tokio::spawn(async move { sync.sync_once().await });
+    let (batch, acknowledgement) = next_commit(&mut receiver).await;
+    assert!(batch.previous_pos.is_none());
+    assert!(saved_position(&client).await.is_none());
+    acknowledgement.send(Ok(())).unwrap();
+    processing.await.unwrap().unwrap();
+}
+
+#[tokio::test]
+async fn successful_acknowledgement_remains_valid_after_caller_cancellation() {
+    let (hook, mut receiver) = queued_hook();
+    let (_server, client, sync) = fixture(Some(hook)).await;
+    let first_sync = sync.clone();
+    let processing = tokio::spawn(async move { first_sync.sync_once().await });
+    let (_, acknowledgement) = next_commit(&mut receiver).await;
+    processing.abort();
+    assert!(processing.await.unwrap_err().is_cancelled());
+    acknowledgement.send(Ok(())).unwrap();
+
+    let processing = tokio::spawn(async move { sync.sync_once().await });
+    let (batch, acknowledgement) = next_commit(&mut receiver).await;
+    assert_eq!(batch.previous_pos.as_deref(), Some("position-1"));
+    assert_eq!(saved_position(&client).await.as_deref(), Some("position-1"));
+    acknowledgement.send(Ok(())).unwrap();
+    processing.await.unwrap().unwrap();
+}
+
+#[tokio::test]
+async fn failed_commit_preserves_a_previous_checkpoint_across_instance_restore() {
+    let (hook, mut receiver) = queued_hook();
+    let (server, client, sync) = fixture(Some(hook.clone())).await;
+    let first_sync = sync.clone();
+    let processing = tokio::spawn(async move { first_sync.sync_once().await });
+    let (_, acknowledgement) = next_commit(&mut receiver).await;
+    acknowledgement.send(Ok(())).unwrap();
+    processing.await.unwrap().unwrap();
+
+    server.server().reset().await;
+    server.mock_sliding_sync().ok(v5::Response::new("position-2".to_owned())).mount().await;
+    let processing = tokio::spawn(async move { sync.sync_once().await });
+    let (batch, acknowledgement) = next_commit(&mut receiver).await;
+    assert_eq!(batch.previous_pos.as_deref(), Some("position-1"));
+    assert_eq!(batch.next_pos, "position-2");
+    acknowledgement.send(Err("write failed".to_owned())).unwrap();
+    assert!(processing.await.unwrap().is_err());
+    assert_eq!(saved_position(&client).await.as_deref(), Some("position-1"));
+
+    let restored = client
+        .sliding_sync("commit-test")
+        .unwrap()
+        .share_pos()
+        .with_response_commit_hook(hook)
+        .build()
+        .await
+        .unwrap();
+    let processing = tokio::spawn(async move { restored.sync_once().await });
+    let (batch, acknowledgement) = next_commit(&mut receiver).await;
+    assert_eq!(batch.previous_pos.as_deref(), Some("position-1"));
+    assert_eq!(batch.next_pos, "position-2");
+    acknowledgement.send(Ok(())).unwrap();
+    processing.await.unwrap().unwrap();
+    assert_eq!(saved_position(&client).await.as_deref(), Some("position-2"));
+}
+
+#[tokio::test]
+async fn absent_hook_preserves_checkpoint_and_observer_behavior() {
+    let (_server, client, sync) = fixture(None).await;
+    let mut updates = client.subscribe_to_all_room_updates();
+    tokio::time::timeout(Duration::from_secs(5), sync.sync_once()).await.unwrap().unwrap();
+    assert_eq!(saved_position(&client).await.as_deref(), Some("position-1"));
+    assert!(updates.try_recv().is_ok());
+}

matrix-rust-sdk

Commit 7aace883fbe0

Commit `7aace883fbe0`