matrix_sdk/event_cache/caches/thread/

mod.rs

// Copyright 2026 The Matrix.org Foundation C.I.C.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! Threads-related data structures.

pub mod pagination;
mod state;
mod updates;

use std::{fmt, sync::Arc};

use eyeball::AsyncLock;
use matrix_sdk_base::{
    deserialized_responses::ThreadSummary,
    event_cache::{Event, thread::ThreadInfo},
    read_receipts::ReadReceipts,
    sync::Timeline,
};
use ruma::{
    EventId, OwnedEventId, OwnedRoomId, OwnedUserId, RoomId, events::relation::RelationType,
    room_version_rules::RoomVersionRules,
};
use tokio::sync::{Notify, broadcast::Sender, mpsc};
use tracing::{instrument, trace};

pub(in super::super) use self::state::ThreadEventCacheState;
pub(super) use self::updates::ThreadEventCacheUpdateSender;
pub use self::{pagination::ThreadPagination, updates::ThreadEventCacheUpdate};
#[cfg(feature = "e2e-encryption")]
use super::super::redecryptor::MaybeResolvedEvent;
use super::{
    super::{
        Result,
        states::{CacheStateLock, StateLock, selectors::ThreadStateSelector},
    },
    EventsOrigin, TimelineVectorDiffs,
    read_receipts::MaybeReceiptEventContent,
    room::{RoomEventCacheGenericUpdate, RoomEventCacheLinkedChunkUpdate},
    subscriber::{AutoShrinkMessage, Subscriber},
};
use crate::room::WeakRoom;

/// All the information related to a single thread.
///
/// Cloning is shallow, and thus is cheap to do.
#[derive(Clone)]
pub struct ThreadEventCache {
    inner: Arc<ThreadEventCacheInner>,
}

/// The (non-cloneable) details of the `ThreadEventCache`.
struct ThreadEventCacheInner {
    /// The room ID.
    room_id: OwnedRoomId,

    /// The thread root ID.
    thread_id: OwnedEventId,

    /// The room where this thread belongs to.
    weak_room: WeakRoom,

    /// State for this thread's event cache.
    state: CacheStateLock<ThreadStateSelector>,

    /// A notifier that we received a new pagination token.
    pagination_batch_token_notifier: Notify,

    /// Sender to the auto-shrink channel.
    ///
    /// See doc comment around [`EventCache::auto_shrink_linked_chunk_task`] for
    /// more details.
    auto_shrink_sender: mpsc::Sender<AutoShrinkMessage>,

    /// Update sender for this thread.
    update_sender: ThreadEventCacheUpdateSender,
}

impl fmt::Debug for ThreadEventCache {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        f.debug_struct("ThreadEventCache").finish_non_exhaustive()
    }
}

impl ThreadEventCache {
    /// Create a new empty thread event cache.
    #[allow(clippy::too_many_arguments)]
    pub(super) async fn new(
        room_id: OwnedRoomId,
        thread_id: OwnedEventId,
        own_user_id: OwnedUserId,
        room_version_rules: RoomVersionRules,
        weak_room: WeakRoom,
        state: &StateLock,
        auto_shrink_sender: mpsc::Sender<AutoShrinkMessage>,
        generic_update_sender: Sender<RoomEventCacheGenericUpdate>,
        linked_chunk_update_sender: Sender<RoomEventCacheLinkedChunkUpdate>,
    ) -> Result<Self> {
        let update_sender = ThreadEventCacheUpdateSender::new(generic_update_sender.clone());

        // Computed while building the state: nothing may await once the state
        // is inserted, or a cancellation or error would leave it
        // inserted without its cache, making every later attempt fail.
        let mut timeline_is_not_empty = false;

        let cache_state = state
            .try_insert_once_with(
                ThreadStateSelector::new(room_id.clone(), thread_id.clone()),
                async |store_guard| {
                    let cache_state = ThreadEventCacheState::new(
                        room_id.clone(),
                        thread_id.clone(),
                        weak_room.clone(),
                        own_user_id,
                        room_version_rules,
                        store_guard,
                        update_sender.clone(),
                        linked_chunk_update_sender,
                    )
                    .await?;

                    timeline_is_not_empty =
                        cache_state.thread_linked_chunk.revents().next().is_some();

                    Ok(cache_state)
                },
            )
            .await?;

        let cache = Self {
            inner: Arc::new(ThreadEventCacheInner {
                room_id: room_id.clone(),
                thread_id,
                weak_room,
                state: cache_state,
                pagination_batch_token_notifier: Notify::new(),
                auto_shrink_sender,
                update_sender,
            }),
        };

        // If at least one event has been loaded, it means there is a timeline.
        // Let's emit a generic update.
        if timeline_is_not_empty {
            let _ = generic_update_sender
                .send(RoomEventCacheGenericUpdate { room_id: room_id.to_owned() });
        }

        Ok(cache)
    }

    /// Get the room ID for this room.
    pub fn room_id(&self) -> &RoomId {
        &self.inner.room_id
    }

    /// Get the thread ID for this thread.
    pub fn thread_id(&self) -> &EventId {
        &self.inner.thread_id
    }

    /// Get the number of unread messages.
    ///
    /// To get multiple information about the read receipts, use
    /// [`Self::read_receipts`] as it involves a single lock.
    pub async fn num_unread_messages(&self) -> Result<u64> {
        Ok(self.inner.state.read().await?.thread_info.read().await.read_receipts.num_unread)
    }

    /// Get the number of unread notifications.
    ///
    /// To get multiple information about the read receipts, use
    /// [`Self::read_receipts`] as it involves a single lock.
    pub async fn num_unread_notifications(&self) -> Result<u64> {
        Ok(self.inner.state.read().await?.thread_info.read().await.read_receipts.num_notifications)
    }

    /// Get the number of unread mentions, that is, messages causing a highlight
    /// in a room.
    ///
    /// To get multiple information about the read receipts, use
    /// [`Self::read_receipts`] as it involves a single lock.
    pub async fn num_unread_mentions(&self) -> Result<u64> {
        Ok(self.inner.state.read().await?.thread_info.read().await.read_receipts.num_mentions)
    }

    /// Get the detailed information about read receipts for this thread.
    pub async fn read_receipts(&self) -> Result<ReadReceipts> {
        Ok(self.inner.state.read().await?.thread_info.read().await.read_receipts.clone())
    }

    /// Subscribe to update of the [`ThreadInfo`].
    pub async fn subscribe_to_thread_info(
        &self,
    ) -> Result<eyeball::Subscriber<ThreadInfo, AsyncLock>> {
        Ok(self.inner.state.read().await?.thread_info.subscribe().await)
    }

    /// Subscribe to this thread updates, after getting the initial list of
    /// events.
    ///
    /// Creating, and especially dropping, a [`Subscriber`] isn't free, as it
    /// triggers side-effects.
    pub async fn subscribe(&self) -> Result<(Vec<Event>, Subscriber<ThreadEventCacheUpdate>)> {
        let state = self.inner.state.read().await?;
        let events =
            state.thread_linked_chunk().events().map(|(_position, item)| item.clone()).collect();

        let subscribers_handle = state.subscribers_handle();

        let subscriber = Subscriber::new(
            self.inner.update_sender.new_thread_receiver(),
            AutoShrinkMessage::Thread {
                room_id: self.inner.room_id.clone(),
                thread_id: self.inner.thread_id.clone(),
            },
            self.inner.auto_shrink_sender.clone(),
            subscribers_handle,
        );

        trace!("added a thread event cache subscriber; new count: {}", subscribers_handle.count());

        Ok((events, subscriber))
    }

    /// Return a [`ThreadPagination`] useful for running back-pagination queries
    /// in this thread.
    pub fn pagination(&self) -> ThreadPagination {
        ThreadPagination::new(self.inner.clone())
    }

    /// Return a reference to the state.
    pub(in super::super) fn state(&self) -> &CacheStateLock<ThreadStateSelector> {
        &self.inner.state
    }

    /// Handle an update from a joined room.
    #[instrument(skip_all, fields(room_id = %self.inner.room_id, thread_root = %self.inner.thread_id))]
    pub(super) async fn handle_joined_room_update(
        &self,
        timeline: Timeline,
        read_receipts: MaybeReceiptEventContent,
    ) -> Result<()> {
        self.handle_timeline(timeline, read_receipts).await
    }

    /// Handle an update from a left room.
    #[instrument(skip_all, fields(room_id = %self.inner.room_id, thread_root = %self.inner.thread_id))]
    pub(super) async fn handle_left_room_update(&self, timeline: Timeline) -> Result<()> {
        self.handle_timeline(timeline, MaybeReceiptEventContent::none()).await
    }

    /// Handle a [`Timeline`], i.e. new events received by a sync for this
    /// thread.
    async fn handle_timeline(
        &self,
        timeline: Timeline,
        read_receipts: MaybeReceiptEventContent,
    ) -> Result<()> {
        if timeline.events.is_empty() && timeline.prev_batch.is_none() && read_receipts.is_none() {
            return Ok(());
        }

        trace!("adding new events");

        let mut state = self.inner.state.write().await?;

        let (stored_prev_batch_token, timeline_event_diffs) =
            state.handle_sync(timeline, &read_receipts).await?;

        // Now that all events have been added, we can trigger the
        // `pagination_token_notifier`.
        if stored_prev_batch_token {
            self.inner.pagination_batch_token_notifier.notify_one();
        }

        if !timeline_event_diffs.is_empty() {
            state.update_sender.send(
                ThreadEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
                    diffs: timeline_event_diffs,
                    origin: EventsOrigin::Sync,
                }),
                // This function is part of the `RoomEventCache` flow. The
                // generic update is handled by it.
                None,
            );
        }

        if let Some(read_receipts) = read_receipts.into_inner() {
            state
                .update_sender
                .send(ThreadEventCacheUpdate::AddReadReceiptEvent { event: read_receipts }, None);
        }

        Ok(())
    }

    /// Update the [`ThreadSummary`] for this thread, and return a copy of it.
    pub(in super::super) async fn update_thread_summary(&self) -> Result<ThreadSummary> {
        let mut state = self.inner.state.write().await?;

        let thread_summary = state.update_thread_summary().await?;

        state
            .update_sender
            .send(ThreadEventCacheUpdate::UpdateSummary(thread_summary.clone()), None);

        Ok(thread_summary)
    }

    /// Find a single event in this thread.
    ///
    /// It starts by looking into loaded events in `EventLinkedChunk` before
    /// looking inside the storage.
    #[cfg(test)]
    async fn find_event(
        &self,
        event_id: &EventId,
    ) -> Result<Option<(super::EventLocation, Event)>> {
        self.inner.state.read().await?.find_event(event_id).await
    }

    /// Try to find an event by ID in this thread, along with its related
    /// events.
    ///
    /// You can filter which types of related events to retrieve using `filter`.
    /// `None` will retrieve related events of any type.
    ///
    /// The related events are sorted like this:
    ///
    /// - events saved out-of-band (with `RoomEventCache::save_events`) will be
    ///   located at the beginning of the array.
    /// - events present in the linked chunk (be it in memory or in the storage)
    ///   will be sorted according to their ordering in the linked chunk.
    pub async fn find_event_with_relations(
        &self,
        event_id: &EventId,
        filter: Option<Vec<RelationType>>,
    ) -> Result<Option<(Event, Vec<Event>)>> {
        // Search in all loaded or stored events.
        Ok(self
            .inner
            .state
            .read()
            .await?
            .find_event_with_relations(event_id, filter)
            .await
            .ok()
            .flatten())
    }

    /// Try to locate the events in the linked chunk corresponding to the given
    /// list of resolved events, and replace them, while alerting observers
    /// about the update.
    ///
    /// Return `true` if at least one event has been updated.
    #[cfg(feature = "e2e-encryption")]
    pub(in super::super) async fn replace_in_memory_utds(
        &self,
        resolved_events: &[MaybeResolvedEvent],
    ) -> Result<bool> {
        let mut state = self.inner.state.write().await?;
        let timeline_event_diffs = state.replace_in_memory_utds(resolved_events)?;

        // Drain the updates to the store, events have already been updated
        // before calling this method.
        let _ = state.thread_linked_chunk_mut().store_updates().take();

        state
            .post_process_upserted_events(
                resolved_events.iter().filter_map(|resolved_event| resolved_event.as_resolved()),
                // Read receipt events aren't encrypted, so we can't have
                // decrypted a new one here. As a result, we don't have any new
                // receipt events to post-process, so we can just pass `None`
                // here.
                //
                // Note: read receipts may be updated anyhow in the
                // post-processing step, as the redecryption may have decrypted
                // some events that don't count as unreads.
                None,
            )
            .await?;

        let timeline_event_diffs = timeline_event_diffs
            .into_iter()
            .flatten()
            .chain(state.thread_linked_chunk_mut().updates_as_vector_diffs())
            .collect::<Vec<_>>();

        Ok(if !timeline_event_diffs.is_empty() {
            state.update_sender.send(
                ThreadEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs {
                    diffs: timeline_event_diffs,
                    origin: EventsOrigin::Cache,
                }),
                Some(RoomEventCacheGenericUpdate { room_id: self.inner.room_id.clone() }),
            );

            true
        } else {
            false
        })
    }
}

#[cfg(all(test, not(target_family = "wasm")))] // This uses the cross-process lock, so needs time support.
mod timed_tests {
    use std::sync::Arc;

    use assert_matches::assert_matches;
    use eyeball_im::VectorDiff;
    use futures_util::FutureExt as _;
    use matrix_sdk_base::{
        RoomState, ThreadingSupport,
        cross_process_lock::CrossProcessLockConfig,
        event_cache::{
            Gap,
            store::{EventCacheStore as _, MemoryStore},
        },
        linked_chunk::{
            ChunkContent, ChunkIdentifier, LinkedChunkId, Position, Update,
            lazy_loader::from_all_chunks,
        },
        store::StoreConfig,
        sync::Timeline,
    };
    use matrix_sdk_test::{ALICE, async_test, event_factory::EventFactory};
    use ruma::{
        event_id,
        events::{AnySyncMessageLikeEvent, AnySyncTimelineEvent},
        room_id, user_id,
    };
    use strass::assert_let;
    use tokio::task::yield_now;

    use super::{
        super::{super::RoomEventCacheGenericUpdate, TimelineVectorDiffs},
        MaybeReceiptEventContent, ThreadEventCacheUpdate,
    };
    use crate::{assert_let_timeout, test_utils::client::MockClientBuilder};

    #[async_test]
    async fn test_write_to_storage() {
        let room_id = room_id!("!r0");
        let thread_root = event_id!("$t0_ev0");
        let thread_event_id_0 = event_id!("$t0_ev1");

        let f = EventFactory::new().room(room_id).sender(user_id!("@mnt_io:matrix.org"));

        let event_cache_store = Arc::new(MemoryStore::new());

        let client = MockClientBuilder::new(None)
            .on_builder(|builder| {
                builder
                    .store_config(
                        StoreConfig::new(CrossProcessLockConfig::multi_process("hodor"))
                            .event_cache_store(event_cache_store.clone()),
                    )
                    .with_threading_support(ThreadingSupport::Enabled { with_subscriptions: true })
            })
            .build()
            .await;

        let event_cache = client.event_cache().unwrap();

        client.base_client().get_or_create_room(room_id, RoomState::Joined);

        let (thread_event_cache, _drop_handles) =
            event_cache.thread(room_id, thread_root).await.unwrap();
        let (thread_events, mut thread_stream) = thread_event_cache.subscribe().await.unwrap();

        assert!(thread_events.is_empty());

        // Propagate an update for a message and a prev-batch token.
        let timeline = Timeline {
            limited: true,
            prev_batch: Some("raclette".to_owned()),
            events: vec![
                f.text_msg("salut")
                    .event_id(thread_event_id_0)
                    .in_thread(thread_root, thread_root)
                    .into_event(),
            ],
        };

        thread_event_cache
            .handle_joined_room_update(timeline, MaybeReceiptEventContent::none())
            .await
            .unwrap();

        assert_matches!(
            thread_stream.recv().await,
            Ok(ThreadEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs { diffs, .. })) => {
                assert_eq!(diffs.len(), 2);
                assert_matches!(&diffs[0], VectorDiff::Clear);
                assert_matches!(&diffs[1], VectorDiff::Append { values: events } => {
                    assert_eq!(events.len(), 1);
                    assert_eq!(events[0].event_id(), Some(thread_event_id_0));
                });
            }
        );
        assert!(thread_stream.is_empty());

        // Check the storage.
        let linked_chunk = from_all_chunks::<3, _, _>(
            event_cache_store
                .load_all_chunks(LinkedChunkId::Thread(room_id, thread_root))
                .await
                .unwrap(),
        )
        .unwrap()
        .unwrap();

        assert_eq!(linked_chunk.chunks().count(), 2);

        let mut chunks = linked_chunk.chunks();

        // We start with the gap.
        assert_matches!(chunks.next().unwrap().content(), ChunkContent::Gap(gap) => {
            assert_eq!(gap.token, "raclette");
        });

        // Then we have the stored event.
        assert_matches!(chunks.next().unwrap().content(), ChunkContent::Items(events) => {
            assert_eq!(events.len(), 1);
            assert_eq!(events[0].event_id(), Some(thread_event_id_0));
        });

        // That's all, folks!
        assert!(chunks.next().is_none());
    }

    #[async_test]
    async fn test_write_to_storage_strips_bundled_relations() {
        let sender = user_id!("@mnt_io:matrix.org");
        let room_id = room_id!("!r0");
        let thread_root = event_id!("$t0_ev0");
        let thread_event_id_0 = event_id!("$t0_ev1");

        let f = EventFactory::new().room(room_id).sender(sender);

        let event_cache_store = Arc::new(MemoryStore::new());

        let client = MockClientBuilder::new(None)
            .on_builder(|builder| {
                builder
                    .store_config(
                        StoreConfig::new(CrossProcessLockConfig::multi_process("hodor"))
                            .event_cache_store(event_cache_store.clone()),
                    )
                    .with_threading_support(ThreadingSupport::Enabled { with_subscriptions: true })
            })
            .build()
            .await;

        let event_cache = client.event_cache().unwrap();

        client.base_client().get_or_create_room(room_id, RoomState::Joined);

        let (thread_event_cache, _drop_handles) =
            event_cache.thread(room_id, thread_root).await.unwrap();

        // Propagate an update for a message with bundled relations.
        let timeline = Timeline {
            limited: false,
            prev_batch: None,
            events: vec![
                f.text_msg("s 'up")
                    .event_id(thread_event_id_0)
                    .with_bundled_edit(f.text_msg("Hello, Kind Sir").sender(sender))
                    .in_thread(thread_root, thread_root)
                    .into_event(),
            ],
        };

        thread_event_cache
            .handle_joined_room_update(timeline, MaybeReceiptEventContent::none())
            .await
            .unwrap();

        // The in-memory linked chunk keeps the bundled relation.
        {
            let (events, _) = thread_event_cache.subscribe().await.unwrap();

            assert_eq!(events.len(), 1);

            let event = events[0].raw().deserialize().unwrap();
            assert_eq!(event.event_id(), thread_event_id_0);
            assert_let!(
                AnySyncTimelineEvent::MessageLike(AnySyncMessageLikeEvent::RoomMessage(msg)) =
                    event
            );
            assert!(msg.as_original().unwrap().unsigned.relations.replace.is_some());
        }

        // The one in storage does not.
        let linked_chunk = from_all_chunks::<3, _, _>(
            event_cache_store
                .load_all_chunks(LinkedChunkId::Thread(room_id, thread_root))
                .await
                .unwrap(),
        )
        .unwrap()
        .unwrap();

        assert_eq!(linked_chunk.chunks().count(), 1);

        let mut chunks = linked_chunk.chunks();
        assert_matches!(chunks.next().unwrap().content(), ChunkContent::Items(events) => {
            assert_eq!(events.len(), 1);

            let event = events[0].raw().deserialize().unwrap();
            assert_eq!(event.event_id(), thread_event_id_0);

            assert_let!(AnySyncTimelineEvent::MessageLike(AnySyncMessageLikeEvent::RoomMessage(msg)) = event);
            assert!(msg.as_original().unwrap().unsigned.relations.replace.is_none());
        });

        // That's all, folks!
        assert!(chunks.next().is_none());
    }

    #[async_test]
    async fn test_clear() {
        let room_id = room_id!("!r0");
        let f = EventFactory::new().room(room_id).sender(user_id!("@mnt_io:matrix.org"));

        let thread_root = event_id!("$t0_ev0");
        let thread_event_id_0 = event_id!("$t0_ev1");
        let thread_event_id_1 = event_id!("$t0_ev2");

        let thread_event_0 = f
            .text_msg("foo")
            .event_id(thread_event_id_0)
            .in_thread(thread_root, thread_root)
            .into_event();
        let thread_event_1 = f
            .text_msg("bar")
            .event_id(thread_event_id_1)
            .in_thread(thread_root, thread_event_id_0)
            .into_event();

        // Prefill the store with some data.
        let event_cache_store = Arc::new(MemoryStore::new());
        event_cache_store
            .handle_linked_chunk_updates(
                LinkedChunkId::Thread(room_id, thread_root),
                vec![
                    // An empty items chunk.
                    Update::NewItemsChunk {
                        previous: None,
                        new: ChunkIdentifier::new(0),
                        next: None,
                    },
                    // A gap chunk.
                    Update::NewGapChunk {
                        previous: Some(ChunkIdentifier::new(0)),
                        // Chunk IDs aren't supposed to be ordered, so use a random value here.
                        new: ChunkIdentifier::new(42),
                        next: None,
                        gap: Gap { token: "comté".to_owned() },
                    },
                    // Another items chunk, non-empty this time.
                    Update::NewItemsChunk {
                        previous: Some(ChunkIdentifier::new(42)),
                        new: ChunkIdentifier::new(1),
                        next: None,
                    },
                    Update::PushItems {
                        at: Position::new(ChunkIdentifier::new(1), 0),
                        items: vec![thread_event_0.clone()],
                    },
                    // And another items chunk, non-empty again.
                    Update::NewItemsChunk {
                        previous: Some(ChunkIdentifier::new(1)),
                        new: ChunkIdentifier::new(2),
                        next: None,
                    },
                    Update::PushItems {
                        at: Position::new(ChunkIdentifier::new(2), 0),
                        items: vec![thread_event_1.clone()],
                    },
                ],
            )
            .await
            .unwrap();

        let client = MockClientBuilder::new(None)
            .on_builder(|builder| {
                builder
                    .store_config(
                        StoreConfig::new(CrossProcessLockConfig::multi_process("hodor"))
                            .event_cache_store(event_cache_store.clone()),
                    )
                    .with_threading_support(ThreadingSupport::Enabled { with_subscriptions: true })
            })
            .build()
            .await;

        let event_cache = client.event_cache().unwrap();

        client.base_client().get_or_create_room(room_id, RoomState::Joined);

        let (thread_event_cache, _drop_handles) =
            event_cache.thread(room_id, thread_root).await.unwrap();
        let (thread_events, mut thread_stream) = thread_event_cache.subscribe().await.unwrap();

        let mut generic_stream = event_cache.subscribe_to_room_generic_updates();

        // The thread knows about all cached events.
        {
            assert!(thread_event_cache.find_event(thread_event_id_0).await.unwrap().is_some());
            assert!(thread_event_cache.find_event(thread_event_id_1).await.unwrap().is_some());
        }

        // But only part of events are loaded from the store.
        {
            // The thread must contain only one event because only one chunk has
            // been loaded.
            assert_eq!(thread_events.len(), 1);
            assert_eq!(thread_events[0].event_id().unwrap(), thread_event_id_1);

            assert!(thread_stream.is_empty());
        }

        // Let's load more chunks to load all events.
        {
            thread_event_cache.pagination().run_backwards_once(20).await.unwrap();

            assert_matches!(
                thread_stream.recv().await,
                Ok(ThreadEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs { diffs, .. })) => {
                    assert_eq!(diffs.len(), 1);
                    assert_matches!(&diffs[0], VectorDiff::Insert { index: 0, value: event } => {
                        // Here you are `thread_event_0`!
                        assert_eq!(event.event_id(), Some(thread_event_id_0));
                    });
                }
            );
            assert!(thread_stream.is_empty());

            assert_matches!(
                generic_stream.recv().await,
                Ok(RoomEventCacheGenericUpdate { room_id: expected_room_id }) => {
                    assert_eq!(room_id, expected_room_id);
                }
            );
            assert!(generic_stream.is_empty());
        }

        // After clearing,…
        event_cache.clear_all_rooms().await.unwrap();

        //… we get an update that the content has been cleared.
        assert_matches!(
            thread_stream.recv().await,
            Ok(ThreadEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs { diffs, .. })) => {
                assert_eq!(diffs.len(), 1);
                assert_matches!(&diffs[0], VectorDiff::Clear);
            }
        );

        // … same with a generic update. (update for the clearing of the room)
        assert_matches!(
            generic_stream.recv().await,
            Ok(RoomEventCacheGenericUpdate { room_id: received_room_id }) => {
                assert_eq!(received_room_id, room_id);
            }
        );
        // (update for the clearing of the thread)
        assert_matches!(
            generic_stream.recv().await,
            Ok(RoomEventCacheGenericUpdate { room_id: received_room_id }) => {
                assert_eq!(received_room_id, room_id);
            }
        );
        assert!(generic_stream.is_empty());

        // Events individually are forgotten by the event cache, after clearing
        // the threads.
        assert!(thread_event_cache.find_event(thread_event_id_0).await.unwrap().is_none());
        assert!(thread_event_cache.find_event(thread_event_id_1).await.unwrap().is_none());

        // And their presence in a linked chunk is forgotten.
        let (thread_events, _) = thread_event_cache.subscribe().await.unwrap();
        assert!(thread_events.is_empty());

        // The event cache store is totally empty.
        assert!(
            event_cache_store
                .load_all_chunks(LinkedChunkId::Thread(room_id, thread_root))
                .await
                .unwrap()
                .is_empty()
        );
    }

    #[async_test]
    async fn test_load_from_storage() {
        let room_id = room_id!("!r0");
        let f = EventFactory::new().room(room_id).sender(user_id!("@mnt_io:matrix.org"));

        let event_cache_store = Arc::new(MemoryStore::new());

        let thread_root = event_id!("$t0");
        let thread_event_id_0 = event_id!("$t0_ev0");
        let thread_event_id_1 = event_id!("$t0_ev1");

        let thread_event_0 = f
            .text_msg("hello world")
            .event_id(thread_event_id_0)
            .in_thread(thread_root, thread_root)
            .into_event();
        let thread_event_1 = f
            .text_msg("how's it going")
            .event_id(thread_event_id_1)
            .in_thread(thread_root, thread_event_id_1)
            .into_event();

        // Prefill the store with some data. The room usually has all events
        // duplicated from the threads. It's important to make the test pass
        // when checking the generic update.
        let updates = vec![
            // An empty items chunk.
            Update::NewItemsChunk { previous: None, new: ChunkIdentifier::new(0), next: None },
            // A gap chunk.
            Update::NewGapChunk {
                previous: Some(ChunkIdentifier::new(0)),
                // Chunk IDs aren't supposed to be ordered, so use a random value here.
                new: ChunkIdentifier::new(42),
                next: None,
                gap: Gap { token: "gruyère".to_owned() },
            },
            // Another items chunk, non-empty this time.
            Update::NewItemsChunk {
                previous: Some(ChunkIdentifier::new(42)),
                new: ChunkIdentifier::new(1),
                next: None,
            },
            Update::PushItems {
                at: Position::new(ChunkIdentifier::new(1), 0),
                items: vec![thread_event_0.clone()],
            },
            // And another items chunk, non-empty again.
            Update::NewItemsChunk {
                previous: Some(ChunkIdentifier::new(1)),
                new: ChunkIdentifier::new(2),
                next: None,
            },
            Update::PushItems {
                at: Position::new(ChunkIdentifier::new(2), 0),
                items: vec![thread_event_1.clone()],
            },
        ];
        event_cache_store
            .handle_linked_chunk_updates(LinkedChunkId::Room(room_id), updates.clone())
            .await
            .unwrap();
        event_cache_store
            .handle_linked_chunk_updates(LinkedChunkId::Thread(room_id, thread_root), updates)
            .await
            .unwrap();

        let client = MockClientBuilder::new(None)
            .on_builder(|builder| {
                builder
                    .store_config(
                        StoreConfig::new(CrossProcessLockConfig::multi_process("hodor"))
                            .event_cache_store(event_cache_store.clone()),
                    )
                    .with_threading_support(ThreadingSupport::Enabled { with_subscriptions: true })
            })
            .build()
            .await;

        let event_cache = client.event_cache().unwrap();

        client.base_client().get_or_create_room(room_id, RoomState::Joined);

        // Let's check whether the generic updates are received for the
        // initialisation.
        let mut generic_stream = event_cache.subscribe_to_room_generic_updates();
        let (thread_event_cache, _drop_handles) =
            event_cache.thread(room_id, thread_root).await.unwrap();
        let (thread_events, mut thread_stream) = thread_event_cache.subscribe().await.unwrap();

        // The room **and** the thread have been loaded. Two generic updates
        // must have been triggered.
        for _ in 0..2 {
            assert_matches!(
                generic_stream.recv().await,
                Ok(RoomEventCacheGenericUpdate { room_id: expected_room_id }) => {
                    assert_eq!(room_id, expected_room_id);
                }
            );
        }
        assert!(generic_stream.is_empty());

        // The initial events contain one event because only the last chunk is
        // loaded by default.
        assert_eq!(thread_events.len(), 1);
        assert_eq!(thread_events[0].event_id().unwrap(), thread_event_id_1);
        assert!(thread_stream.is_empty());

        // The thread knows all events in the storage though, even if they
        // aren't loaded.
        assert!(thread_event_cache.find_event(thread_event_id_0).await.unwrap().is_some());
        assert!(thread_event_cache.find_event(thread_event_id_1).await.unwrap().is_some());

        // Let's paginate to load more events.
        thread_event_cache.pagination().run_backwards_once(20).await.unwrap();

        assert_matches!(
            thread_stream.recv().await,
            Ok(ThreadEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs { diffs, .. })) => {
                assert_eq!(diffs.len(), 1);
                assert_matches!(&diffs[0], VectorDiff::Insert { index: 0, value: event } => {
                    assert_eq!(event.event_id(), Some(thread_event_id_0));
                });
            }
        );
        assert!(thread_stream.is_empty());

        // A generic update is triggered too.
        assert_matches!(
            generic_stream.recv().await,
            Ok(RoomEventCacheGenericUpdate { room_id: expected_room_id }) => {
                assert_eq!(expected_room_id, room_id);
            }
        );
        assert!(generic_stream.is_empty());

        // A new update with one of these events leads to deduplication.
        let timeline = Timeline { limited: false, prev_batch: None, events: vec![thread_event_1] };

        thread_event_cache
            .handle_joined_room_update(timeline, MaybeReceiptEventContent::none())
            .await
            .unwrap();

        // Just checking the generic update is correct. There is a duplicate
        // event, so no generic changes whatsoever!
        assert!(generic_stream.recv().now_or_never().is_none());

        // The stream doesn't report these changes _yet_. Use the events vector
        // given when subscribing, to check that the events correspond to their
        // new positions. The duplicated item is removed (so it's not the first
        // element anymore), and it's added to the back of the list.
        let (thread_events, _) = thread_event_cache.subscribe().await.unwrap();
        assert_eq!(thread_events.len(), 2);
        assert_eq!(thread_events[0].event_id(), Some(thread_event_id_0));
        assert_eq!(thread_events[1].event_id(), Some(thread_event_id_1));
    }

    #[async_test]
    async fn test_load_from_storage_resilient_to_failure() {
        let room_id = room_id!("!r0");
        let f = EventFactory::new().room(room_id).sender(user_id!("@mnt_io:matrix.org"));

        let thread_root = event_id!("$t0");
        let thread_event_id_0 = event_id!("$t0_ev0");

        let thread_event_0 = f
            .text_msg("hello world")
            .event_id(thread_event_id_0)
            .in_thread(thread_root, thread_root)
            .into_event();

        // Prefill the store with invalid data: two chunks that form a cycle.
        let event_cache_store = Arc::new(MemoryStore::new());
        event_cache_store
            .handle_linked_chunk_updates(
                LinkedChunkId::Thread(room_id, thread_root),
                vec![
                    Update::NewItemsChunk {
                        previous: None,
                        new: ChunkIdentifier::new(0),
                        next: None,
                    },
                    Update::PushItems {
                        at: Position::new(ChunkIdentifier::new(0), 0),
                        items: vec![thread_event_0],
                    },
                    Update::NewItemsChunk {
                        previous: Some(ChunkIdentifier::new(0)),
                        new: ChunkIdentifier::new(1),
                        next: Some(ChunkIdentifier::new(0)),
                    },
                ],
            )
            .await
            .unwrap();

        let client = MockClientBuilder::new(None)
            .on_builder(|builder| {
                builder
                    .store_config(
                        StoreConfig::new(CrossProcessLockConfig::multi_process("holder"))
                            .event_cache_store(event_cache_store.clone()),
                    )
                    .with_threading_support(ThreadingSupport::Enabled { with_subscriptions: true })
            })
            .build()
            .await;

        let event_cache = client.event_cache().unwrap();

        client.base_client().get_or_create_room(room_id, RoomState::Joined);

        let (thread_event_cache, _drop_handles) =
            event_cache.thread(room_id, thread_root).await.unwrap();
        let (thread_events, _) = thread_event_cache.subscribe().await.unwrap();

        // Because the persisted content was invalid, the thread store is reset:
        // there are no events in the cache.
        assert!(thread_events.is_empty());

        // Storage doesn't contain anything. It would also be valid that it
        // contains a single initial empty items chunk.
        let raw_chunks = event_cache_store
            .load_all_chunks(LinkedChunkId::Thread(room_id, thread_root))
            .await
            .unwrap();
        assert!(raw_chunks.is_empty());
    }

    #[async_test]
    async fn test_reload_when_dirty() {
        let user_id = user_id!("@mnt_io:matrix.org");
        let room_id = room_id!("!raclette:patate.ch");

        let event_factory = EventFactory::new().room(room_id).sender(user_id);

        let thread_root = event_id!("$t0");
        let thread_event_id_0 = event_id!("$t0_ev0");
        let thread_event_id_1 = event_id!("$t0_ev1");

        let thread_event_0 = event_factory
            .text_msg("comté")
            .event_id(thread_event_id_0)
            .in_thread(thread_root, thread_root)
            .into_event();
        let thread_event_1 = event_factory
            .text_msg("morbier")
            .event_id(thread_event_id_1)
            .in_thread(thread_root, thread_event_id_0)
            .into_event();

        // The storage shared by the two clients.
        let event_cache_store = MemoryStore::new();
        event_cache_store
            .handle_linked_chunk_updates(
                LinkedChunkId::Thread(room_id, thread_root),
                vec![
                    Update::NewItemsChunk {
                        previous: None,
                        new: ChunkIdentifier::new(0),
                        next: None,
                    },
                    Update::PushItems {
                        at: Position::new(ChunkIdentifier::new(0), 0),
                        items: vec![thread_event_0],
                    },
                    Update::NewItemsChunk {
                        previous: Some(ChunkIdentifier::new(0)),
                        new: ChunkIdentifier::new(1),
                        next: None,
                    },
                    Update::PushItems {
                        at: Position::new(ChunkIdentifier::new(1), 0),
                        items: vec![thread_event_1],
                    },
                ],
            )
            .await
            .unwrap();

        // Client for the process 0.
        let client_p0 = MockClientBuilder::new(None)
            .on_builder(|builder| {
                builder
                    .store_config(
                        StoreConfig::new(CrossProcessLockConfig::multi_process("process #0"))
                            .event_cache_store(event_cache_store.clone()),
                    )
                    .with_threading_support(ThreadingSupport::Enabled { with_subscriptions: true })
            })
            .build()
            .await;

        // Client for the process 1.
        let client_p1 = MockClientBuilder::new(None)
            .on_builder(|builder| {
                builder
                    .store_config(
                        StoreConfig::new(CrossProcessLockConfig::multi_process("process #1"))
                            .event_cache_store(event_cache_store),
                    )
                    .with_threading_support(ThreadingSupport::Enabled { with_subscriptions: true })
            })
            .build()
            .await;

        // Subscribe the event caches, and create the room.
        let (thread_event_cache_p0, thread_event_cache_p1) = {
            let event_cache_p0 = client_p0.event_cache().unwrap();
            let event_cache_p1 = client_p1.event_cache().unwrap();

            client_p0.base_client().get_or_create_room(room_id, RoomState::Joined);
            client_p1.base_client().get_or_create_room(room_id, RoomState::Joined);

            let (thread_event_cache_p0, _drop_handles) =
                event_cache_p0.thread(room_id, thread_root).await.unwrap();
            let (thread_event_cache_p1, _drop_handles) =
                event_cache_p1.thread(room_id, thread_root).await.unwrap();

            (thread_event_cache_p0, thread_event_cache_p1)
        };

        // Okay. We are ready for the test!
        //
        // First off, let's check `thread_event_cache_p0` has access to the
        // first event loaded in-memory, then do a pagination, and see more
        // events.
        let mut updates_stream_p0 = {
            let thread_event_cache = &thread_event_cache_p0;

            let (initial_updates, mut updates_stream) =
                thread_event_cache_p0.subscribe().await.unwrap();

            // Initial updates contain `thread_event_id_1` only.
            assert_eq!(initial_updates.len(), 1);
            assert_eq!(initial_updates[0].event_id(), Some(thread_event_id_1));
            assert!(updates_stream.is_empty());

            // Load one more event with a backpagination.
            thread_event_cache.pagination().run_backwards_once(1).await.unwrap();

            // A new update for `ev_id_0` must be present.
            assert_matches!(
                updates_stream.recv().await.unwrap(),
                ThreadEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs { diffs, .. }) => {
                    assert_eq!(diffs.len(), 1, "{diffs:#?}");
                    assert_matches!(
                        &diffs[0],
                        VectorDiff::Insert { index: 0, value: event } => {
                            assert_eq!(event.event_id(), Some(thread_event_id_0));
                        }
                    );
                }
            );

            updates_stream
        };

        // Second, let's check `thread_event_cache_p1` has the same accesses.
        let mut updates_stream_p1 = {
            let thread_event_cache = &thread_event_cache_p1;
            let (initial_updates, mut updates_stream) =
                thread_event_cache_p1.subscribe().await.unwrap();

            // Initial updates contain `thread_event_id_1` only.
            assert_eq!(initial_updates.len(), 1);
            assert_eq!(initial_updates[0].event_id(), Some(thread_event_id_1));
            assert!(updates_stream.is_empty());

            // Load one more event with a backpagination.
            thread_event_cache.pagination().run_backwards_once(1).await.unwrap();

            // A new update for `thread_event_id_0` must be present.
            assert_matches!(
                updates_stream.recv().await.unwrap(),
                ThreadEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs { diffs, .. }) => {
                    assert_eq!(diffs.len(), 1, "{diffs:#?}");
                    assert_matches!(
                        &diffs[0],
                        VectorDiff::Insert { index: 0, value: event } => {
                            assert_eq!(event.event_id(), Some(thread_event_id_0));
                        }
                    );
                }
            );

            updates_stream
        };

        // Do this a couple times, for the fun.
        for _ in 0..3 {
            // Third, because `thread_event_cache_p1` has locked the store, the
            // lock is dirty for `thread_event_cache_p0`, so it will shrink to
            // its last chunk for the thread!
            {
                let thread_event_cache = &thread_event_cache_p0;
                let updates_stream = &mut updates_stream_p0;

                // `thread_event_id_1` must be loaded in memory, just like
                // before. However, `thread_event_id_0` must NOT be loaded in
                // memory. It WAS loaded, but the state has been reloaded to its
                // last chunk.
                let (initial_updates, _) = thread_event_cache.subscribe().await.unwrap();

                assert_eq!(initial_updates.len(), 1);
                assert_eq!(initial_updates[0].event_id(), Some(thread_event_id_1));

                // The reload can be observed via the updates too.
                assert_matches!(
                    updates_stream.recv().await.unwrap(),
                    ThreadEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs { diffs, .. }) => {
                        assert_eq!(diffs.len(), 2, "{diffs:#?}");
                        assert_matches!(&diffs[0], VectorDiff::Clear);
                        assert_matches!(
                            &diffs[1],
                            VectorDiff::Append { values: events } => {
                                assert_eq!(events.len(), 1);
                                assert_eq!(events[0].event_id(), Some(thread_event_id_1));
                            }
                        );
                    }
                );
                assert_matches!(
                    updates_stream.recv().await.unwrap(),
                    ThreadEventCacheUpdate::UpdateSummary(summary) => {
                        assert_eq!(summary.latest_reply.as_deref(), Some(thread_event_id_1));
                        assert_eq!(summary.num_replies, 2);
                    }
                );

                // Load one more event with a backpagination.
                thread_event_cache.pagination().run_backwards_once(1).await.unwrap();

                // `thread_event_id_0` must now be loaded in memory. The
                // pagination can be observed via the updates.
                assert_matches!(
                    updates_stream.recv().await.unwrap(),
                    ThreadEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs { diffs, .. }) => {
                        assert_eq!(diffs.len(), 1, "{diffs:#?}");
                        assert_matches!(
                            &diffs[0],
                            VectorDiff::Insert { index: 0, value: event } => {
                                assert_eq!(event.event_id(), Some(thread_event_id_0));
                            }
                        );
                    }
                );

                assert!(updates_stream.is_empty());
            }

            // Fourth, because `thread_event_cache_p0` has locked the store
            // again, the lock is dirty for `thread_event_cache_p1` too!, so it
            // will shrink to its last chunk for the thread!
            {
                let thread_event_cache = &thread_event_cache_p1;
                let updates_stream = &mut updates_stream_p1;

                // `thread_event_id_1` must be loaded in memory, just like
                // before. However, `thread_event_id_0` must NOT be loaded in
                // memory. It WAS loaded, but the state has shrunk to its last
                // chunk.
                let (initial_updates, _) = thread_event_cache.subscribe().await.unwrap();

                assert_eq!(initial_updates.len(), 1);
                assert_eq!(initial_updates[0].event_id(), Some(thread_event_id_1));

                // The reload can be observed via the updates too.
                assert_matches!(
                    updates_stream.recv().await.unwrap(),
                    ThreadEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs { diffs, .. }) => {
                        assert_eq!(diffs.len(), 2, "{diffs:#?}");
                        assert_matches!(&diffs[0], VectorDiff::Clear);
                        assert_matches!(
                            &diffs[1],
                            VectorDiff::Append { values: events } => {
                                assert_eq!(events.len(), 1);
                                assert_eq!(events[0].event_id(), Some(thread_event_id_1));
                            }
                        );
                    }
                );
                assert_matches!(
                    updates_stream.recv().await.unwrap(),
                    ThreadEventCacheUpdate::UpdateSummary(summary) => {
                        assert_eq!(summary.latest_reply.as_deref(), Some(thread_event_id_1));
                        assert_eq!(summary.num_replies, 2);
                    }
                );

                // Load one more event with a backpagination.
                thread_event_cache.pagination().run_backwards_once(1).await.unwrap();

                // `thread_event_id_0` must now be loaded in memory. The
                // pagination can be observed via the updates.
                assert_matches!(
                    updates_stream.recv().await.unwrap(),
                    ThreadEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs { diffs, .. }) => {
                        assert_eq!(diffs.len(), 1, "{diffs:#?}");
                        assert_matches!(
                            &diffs[0],
                            VectorDiff::Insert { index: 0, value: event } => {
                                assert_eq!(event.event_id(), Some(thread_event_id_0));
                            }
                        );
                    }
                );
            }
        }
    }

    #[async_test]
    async fn test_auto_shrink_after_all_subscribers_are_gone() {
        let room_id = room_id!("!r0");
        let thread_id = event_id!("$t0");

        let f = EventFactory::new().room(room_id).sender(*ALICE);

        let event_id_0 = event_id!("$ev0");
        let event_id_1 = event_id!("$ev1");

        let thread_root =
            f.text_msg("gr00t").event_id(thread_id).in_thread(thread_id, thread_id).into_event();
        let event_0 =
            f.text_msg("hello").event_id(event_id_0).in_thread(thread_id, event_id_0).into_event();
        let event_1 =
            f.text_msg("world").event_id(event_id_1).in_thread(thread_id, event_id_1).into_event();

        // Fill the event cache store with an initial linked chunk with 2 events
        // chunks.
        let event_cache_store = MemoryStore::new();
        event_cache_store
            .handle_linked_chunk_updates(
                LinkedChunkId::Thread(room_id, thread_id),
                vec![
                    Update::NewItemsChunk {
                        previous: None,
                        new: ChunkIdentifier::new(0),
                        next: None,
                    },
                    Update::PushItems {
                        at: Position::new(ChunkIdentifier::new(0), 0),
                        items: vec![thread_root, event_0],
                    },
                    Update::NewItemsChunk {
                        previous: Some(ChunkIdentifier::new(0)),
                        new: ChunkIdentifier::new(1),
                        next: None,
                    },
                    Update::PushItems {
                        at: Position::new(ChunkIdentifier::new(1), 0),
                        items: vec![event_1],
                    },
                ],
            )
            .await
            .unwrap();

        let client = MockClientBuilder::new(None)
            .on_builder(move |builder| {
                builder.store_config(
                    StoreConfig::new(CrossProcessLockConfig::MultiProcess {
                        holder_name: "foo".to_owned(),
                    })
                    .event_cache_store(event_cache_store),
                )
            })
            .build()
            .await;

        let event_cache = client.event_cache().unwrap();

        client.base_client().get_or_create_room(room_id, RoomState::Joined);

        let (thread_event_cache, _drop_handles) =
            event_cache.thread(room_id, thread_id).await.unwrap();

        // Sanity check: lazily loaded, so only includes one item at start.
        let (events1, mut stream1) = thread_event_cache.subscribe().await.unwrap();
        assert_eq!(events1.len(), 1);
        assert_eq!(events1[0].event_id(), Some(event_id_1));
        assert!(stream1.is_empty());

        let mut generic_stream = event_cache.subscribe_to_room_generic_updates();

        // Force loading the full linked chunk by back-paginating.
        let outcome = thread_event_cache.pagination().run_backwards_once(20).await.unwrap();
        assert_eq!(outcome.events.len(), 2);
        assert_eq!(outcome.events[0].event_id(), Some(event_id_0));
        assert_eq!(outcome.events[1].event_id(), Some(thread_id));
        assert!(outcome.reached_start);

        // We also get an update about the loading from the store. Ignore it,
        // for this test's sake.
        assert_let_timeout!(
            Ok(ThreadEventCacheUpdate::UpdateTimelineEvents(TimelineVectorDiffs { diffs, .. })) =
                stream1.recv()
        );
        assert_eq!(diffs.len(), 2);
        assert_matches!(&diffs[0], VectorDiff::Insert { index: 0, value } => {
            assert_eq!(value.event_id(), Some(thread_id));
        });
        assert_matches!(&diffs[1], VectorDiff::Insert { index: 1, value } => {
            assert_eq!(value.event_id(), Some(event_id_0));
        });

        assert!(stream1.is_empty());

        assert_let_timeout!(
            Ok(RoomEventCacheGenericUpdate { room_id: expected_room_id }) = generic_stream.recv()
        );
        assert_eq!(expected_room_id, room_id);
        assert!(generic_stream.is_empty());

        // Have another subscriber. Since it's not the first one, and the
        // previous one loaded some more events, the second subscribers sees
        // them all.
        let (events2, stream2) = thread_event_cache.subscribe().await.unwrap();
        assert_eq!(events2.len(), 3);
        assert_eq!(events2[0].event_id(), Some(thread_id));
        assert_eq!(events2[1].event_id(), Some(event_id_0));
        assert_eq!(events2[2].event_id(), Some(event_id_1));
        assert!(stream2.is_empty());

        // Grab a receiver for testing no diffs is sent.
        let subscriber = {
            let state = thread_event_cache.inner.state.read().await.unwrap();
            state.update_sender.new_thread_receiver()
        };

        // Drop the first stream, and wait a bit.
        drop(stream1);
        yield_now().await;

        // The second stream remains undisturbed.
        assert!(stream2.is_empty());

        // Now drop the second stream, and wait a bit.
        drop(stream2);
        yield_now().await;

        // The linked chunk must have auto-shrunk by now.

        {
            // Check the inner state: there's no more shared auto-shrinker.
            let state = thread_event_cache.inner.state.read().await.unwrap();
            assert_eq!(state.subscribers_handle().count(), 0);

            // No diff is sent when the linked chunk has auto-shrunk.
            assert!(subscriber.is_empty());
            assert!(generic_stream.is_empty());
        }

        // Getting the events will only give us the latest chunk.
        let events3 = thread_event_cache
            .inner
            .state
            .read()
            .await
            .unwrap()
            .thread_linked_chunk()
            .events()
            .map(|(_position, item)| item.clone())
            .collect::<Vec<_>>();
        assert_eq!(events3.len(), 1);
        assert_eq!(events3[0].event_id(), Some(event_id_1));
    }
}