synapse/tests/storage/test_federation_store.py

#
# This file is licensed under the Affero General Public License (AGPL) version 3.
#
# Copyright (C) 2026 Element Creations Ltd
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# See the GNU Affero General Public License for more details:
# <https://www.gnu.org/licenses/agpl-3.0.html>.
#

"""The Rust federation store (`rust/src/storage/federation_store.rs`) against
the Python store methods it ports, on the same database. These need the native
Rust Postgres driver:

    SYNAPSE_TEST_RUST_DRIVER=1 POSTGRES=1 run-trial.sh tests.storage.test_federation_store
"""

from typing import Any
from unittest import SkipTest

from signedjson.key import (
    encode_verify_key_base64,
    generate_signing_key,
    get_verify_key,
)

from twisted.internet.testing import MemoryReactor

from synapse.handlers.e2e_keys import SignatureListItem
from synapse.replication.tcp.streams._base import StickyEventStreamPosition
from synapse.rest import admin
from synapse.rest.client import login, room
from synapse.server import HomeServer
from synapse.storage.rust_pool import RustConnectionPool
from synapse.synapse_rust.federation import FederationStoreForTest
from synapse.types import JsonDict, RoomID
from synapse.util.clock import Clock
from synapse.util.duration import Duration

from tests import unittest

DEST = "remote.example.com"
OTHER = "other.example.com"
ROOM = "!room:test"


def _sorted_edus(edus: list[tuple[str, JsonDict]]) -> list[tuple[str, JsonDict]]:
    """The EDUs in a fixed order. Python orders the users as the database
    returns them, and Rust by their first update."""
    return sorted(
        edus,
        key=lambda edu: (edu[0], edu[1]["user_id"], edu[1].get("stream_id", 0)),
    )


class FederationStoreTestCase(unittest.HomeserverTestCase):
    """Each test runs a Rust query and the Python store method against the
    same fixture and compares."""

    def default_config(self) -> dict[str, Any]:
        config = super().default_config()
        # `get_device_updates_by_remote` only works on a federation sender.
        config["federation_sender_instances"] = ["master"]
        return config

    def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
        hs = self.setup_test_homeserver()
        if not isinstance(
            hs.get_datastores().main.db_pool._db_pool, RustConnectionPool
        ):
            raise SkipTest("needs the native Rust Postgres driver")
        return hs

    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
        self.store = hs.get_datastores().main
        db_pool = self.store.db_pool._db_pool
        assert isinstance(db_pool, RustConnectionPool)
        assert db_pool._pool is not None
        self.rust = FederationStoreForTest(db_pool._pool, False, hs.hostname)
        self.rust_with_names = FederationStoreForTest(db_pool._pool, True, hs.hostname)

    # To-device messages.

    def _to_device_edu(self, n: int) -> JsonDict:
        return {
            "sender": "@alice:test",
            "type": "m.test",
            "message_id": str(n),
            "messages": {"@u:remote": {"D": {"n": n}}},
        }

    def _queue_to_device(self, destination: str, n: int) -> None:
        self.get_success(
            self.store.add_remote_messages_from_client_to_device_inbox(
                {destination: self._to_device_edu(n)}
            )
        )

    def test_new_device_msgs_for_remote(self) -> None:
        for n in range(3):
            self._queue_to_device(DEST, n)
            self._queue_to_device(OTHER, n)
        current = self.store.get_to_device_stream_token()

        for last in (0, current - 3, current):
            for limit in (0, 1, 2, 10):
                self.assertEqual(
                    self.rust.get_new_device_msgs_for_remote(
                        DEST, last, current, limit
                    ),
                    self.get_success(
                        self.store.get_new_device_msgs_for_remote(
                            DEST, last, current, limit
                        )
                    ),
                    (last, limit),
                )

    def test_delete_device_msgs_for_remote(self) -> None:
        for n in range(3):
            self._queue_to_device(DEST, n)
        self._queue_to_device(OTHER, 0)
        current = self.store.get_to_device_stream_token()

        self.rust.delete_device_msgs_for_remote(DEST, current - 2)

        messages, _ = self.get_success(
            self.store.get_new_device_msgs_for_remote(DEST, 0, current, 10)
        )
        self.assertEqual(messages, [self._to_device_edu(2)])
        messages, _ = self.get_success(
            self.store.get_new_device_msgs_for_remote(OTHER, 0, current, 10)
        )
        self.assertEqual(len(messages), 1)

    # Device list updates.

    def _poke(self, user_id: str, device_id: str, host: str = DEST) -> None:
        self.get_success(
            self.store.add_device_change_to_streams(user_id, [device_id], [ROOM])
        )
        self.get_success(
            self.store.add_device_list_outbound_pokes(
                user_id=user_id,
                device_id=device_id,
                room_id=ROOM,
                hosts=[host],
                context={},
            )
        )

    def _cross_signing_key(self, user_id: str, key_type: str) -> str:
        """Give the user a cross-signing key, and return its device ID."""
        public = encode_verify_key_base64(get_verify_key(generate_signing_key("x")))
        key = {
            "user_id": user_id,
            "usage": [key_type],
            "keys": {f"ed25519:{public}": public},
        }
        self.get_success(self.store.set_e2e_cross_signing_key(user_id, key_type, key))
        return public

    def _device_keys(self, user_id: str, device_id: str) -> None:
        self.get_success(
            self.store.set_e2e_device_keys(
                user_id,
                device_id,
                self.clock.time_msec(),
                {
                    "user_id": user_id,
                    "device_id": device_id,
                    "algorithms": ["m.olm.v1.curve25519-aes-sha2"],
                    "keys": {f"ed25519:{device_id}": "key"},
                    "signatures": {user_id: {f"ed25519:{device_id}": "self"}},
                },
            )
        )

    def _device_list_fixture(self) -> None:
        """Pokes to DEST for: devices with and without keys and names, a
        device's repeated update, a deleted device, a hidden device, a
        cross-signed device, and both kinds of cross-signing key update. And
        one poke for another destination."""
        alice = "@alice:test"
        bob = "@bob:test"

        self.get_success(self.store.store_device(alice, "A1", "Alice's phone"))
        self.get_success(self.store.store_device(alice, "A2", None))
        self.get_success(self.store.store_device(bob, "B1", "Bob's laptop"))
        self._device_keys(alice, "A1")
        self._device_keys(bob, "B1")

        master = self._cross_signing_key(alice, "master")
        self_signing = self._cross_signing_key(alice, "self_signing")
        self.get_success(
            self.store.store_e2e_cross_signing_signatures(
                alice,
                [
                    SignatureListItem(
                        signing_key_id=f"ed25519:{self_signing}",
                        target_user_id=alice,
                        target_device_id="A1",
                        signature="cross",  # type: ignore[arg-type]
                    )
                ],
            )
        )

        self._poke(alice, "A1")
        self._poke(bob, "B1")
        self._poke(alice, "A2")
        self._poke(alice, master)
        self._poke(alice, "A1")
        self._poke(bob, "GONE")
        self._poke(alice, self_signing)
        self._poke(bob, "B1", host=OTHER)

        # A hidden device is sent as deleted.
        self.get_success(self.store.store_device(bob, "HIDDEN", None))
        self.get_success(
            self.store.db_pool.simple_update_one(
                "devices", {"user_id": bob, "device_id": "HIDDEN"}, {"hidden": True}
            )
        )
        self._poke(bob, "HIDDEN")

    def _assert_device_updates(self, from_stream_id: int, limit: int) -> None:
        now = self.store.get_device_stream_token().stream
        python = self.get_success(
            self.store.get_device_updates_by_remote(DEST, from_stream_id, limit)
        )
        for rust, names in ((self.rust, False), (self.rust_with_names, True)):
            self.hs.config.federation.allow_device_name_lookup_over_federation = names
            python = self.get_success(
                self.store.get_device_updates_by_remote(DEST, from_stream_id, limit)
            )
            stream_id, edus = rust.get_device_updates_by_remote(
                DEST, from_stream_id, now, limit
            )
            self.assertEqual(stream_id, python[0], (from_stream_id, limit, names))
            self.assertEqual(
                _sorted_edus(edus),
                _sorted_edus(python[1]),
                (from_stream_id, limit, names),
            )

    def test_device_updates_by_remote(self) -> None:
        self._device_list_fixture()
        for limit in (1, 2, 3, 100):
            self._assert_device_updates(0, limit)

        # The fixture has every kind of update.
        _, edus = self.rust_with_names.get_device_updates_by_remote(
            DEST, 0, self.store.get_device_stream_token().stream, 100
        )
        types = [edu_type for edu_type, _ in edus]
        self.assertEqual(types.count("m.signing_key_update"), 1)
        self.assertEqual(types.count("org.matrix.signing_key_update"), 1)
        updates = {
            content["device_id"]: content
            for edu_type, content in edus
            if edu_type == "m.device_list_update"
        }
        self.assertEqual(set(updates), {"A1", "A2", "B1", "GONE", "HIDDEN"})
        self.assertEqual(updates["A1"]["device_display_name"], "Alice's phone")
        self.assertIn("cross", str(updates["A1"]["keys"]["signatures"]))
        self.assertNotIn("keys", updates["A2"])
        self.assertTrue(updates["GONE"]["deleted"])
        self.assertTrue(updates["HIDDEN"]["deleted"])
        # A1 was updated again after A2, so it follows A2.
        self.assertEqual(updates["A1"]["prev_id"], [updates["A2"]["stream_id"]])

    def test_device_updates_by_remote_paginates(self) -> None:
        """Walk the stream a few updates at a time, marking each batch sent,
        so the prev IDs come from the last batch."""
        self._device_list_fixture()
        position = 0
        for _ in range(10):
            self._assert_device_updates(position, 3)
            stream_id, edus = self.rust.get_device_updates_by_remote(
                DEST, position, self.store.get_device_stream_token().stream, 3
            )
            if not edus:
                break
            self.rust.mark_as_sent_devices_by_remote(DEST, stream_id)
            position = stream_id
        else:
            self.fail("never caught up")

    def test_mark_as_sent_devices_by_remote(self) -> None:
        """The pokes sent are deleted, and each user's last one recorded."""
        self._device_list_fixture()
        stream_id, _ = self.get_success(
            self.store.get_device_updates_by_remote(DEST, 0, limit=4)
        )
        pokes = self.get_success(
            self.store.db_pool.simple_select_list(
                "device_lists_outbound_pokes",
                {"destination": DEST},
                ["user_id", "stream_id"],
            )
        )
        expected: dict[str, int] = {}
        for user_id, poke_stream_id in pokes:
            if poke_stream_id <= stream_id:
                expected[user_id] = max(expected.get(user_id, 0), poke_stream_id)

        self.rust.mark_as_sent_devices_by_remote(DEST, stream_id)

        remaining = self.get_success(
            self.store.db_pool.simple_select_list(
                "device_lists_outbound_pokes",
                {"destination": DEST},
                ["stream_id"],
            )
        )
        self.assertTrue(remaining)
        self.assertTrue(all(row[0] > stream_id for row in remaining))
        last_success = self.get_success(
            self.store.db_pool.simple_select_list(
                "device_lists_outbound_last_success",
                {"destination": DEST},
                ["user_id", "stream_id"],
            )
        )
        self.assertEqual(dict(last_success), expected)
        # Another destination's pokes are left alone.
        other = self.get_success(
            self.store.db_pool.simple_select_list(
                "device_lists_outbound_pokes", {"destination": OTHER}, ["stream_id"]
            )
        )
        self.assertEqual(len(other), 1)

        # Marking again moves a user's last success on.
        self.rust.mark_as_sent_devices_by_remote(
            DEST, self.store.get_device_stream_token().stream
        )
        last_success = self.get_success(
            self.store.db_pool.simple_select_list(
                "device_lists_outbound_last_success",
                {"destination": DEST},
                ["user_id", "stream_id"],
            )
        )
        self.assertEqual(
            {user_id for user_id, _ in last_success}, {"@alice:test", "@bob:test"}
        )
        self.assertTrue(
            all(stream > expected.get(user_id, 0) for user_id, stream in last_success)
        )

    def test_set_destination_last_successful_stream_ordering(self) -> None:
        self.rust.set_destination_last_successful_stream_ordering(DEST, 5)
        self.assertEqual(
            self.get_success(
                self.store.get_destination_last_successful_stream_ordering(DEST)
            ),
            5,
        )
        self.rust.set_destination_last_successful_stream_ordering(DEST, 7)
        self.assertEqual(
            self.get_success(
                self.store.get_destination_last_successful_stream_ordering(DEST)
            ),
            7,
        )


class CatchUpStoreTestCase(unittest.HomeserverTestCase):
    """The catch-up queries against Python's, on the same database."""

    servlets = [
        admin.register_servlets,
        login.register_servlets,
        room.register_servlets,
    ]

    def default_config(self) -> dict[str, Any]:
        config = super().default_config()
        config["experimental_features"] = {"msc4354_enabled": True}
        return config

    def make_homeserver(self, reactor: MemoryReactor, clock: Clock) -> HomeServer:
        hs = self.setup_test_homeserver()
        if not isinstance(
            hs.get_datastores().main.db_pool._db_pool, RustConnectionPool
        ):
            raise SkipTest("needs the native Rust Postgres driver")
        return hs

    def prepare(self, reactor: MemoryReactor, clock: Clock, hs: HomeServer) -> None:
        self.store = hs.get_datastores().main
        db_pool = self.store.db_pool._db_pool
        assert isinstance(db_pool, RustConnectionPool)
        assert db_pool._pool is not None
        self.rust = FederationStoreForTest(db_pool._pool, False, hs.hostname)

        self.user = self.register_user("alice", "pass")
        self.tok = self.login("alice", "pass")
        self.rooms = [
            self.helper.create_room_as(self.user, tok=self.tok) for _ in range(3)
        ]

    def _send(self, room_id: str, sticky: bool) -> int:
        """Send a message, record it as due to both destinations, and return
        its stream ordering."""
        content = {"body": "hi", "msgtype": "m.text"}
        if sticky:
            event_id = self.helper.send_sticky_event(
                room_id,
                "m.room.message",
                duration=Duration(minutes=10),
                content=content,
                tok=self.tok,
            )["event_id"]
        else:
            event_id = self.helper.send_event(
                room_id, "m.room.message", content, tok=self.tok
            )["event_id"]
        event = self.get_success(self.store.get_event(event_id))
        stream_ordering = event.internal_metadata.stream_ordering
        assert stream_ordering is not None
        self.get_success(
            self.store.store_destination_rooms_entries(
                [DEST, OTHER], room_id, stream_ordering
            )
        )
        return stream_ordering

    def test_last_successful_stream_ordering(self) -> None:
        self.assertIsNone(
            self.rust.get_destination_last_successful_stream_ordering(DEST)
        )
        self.get_success(
            self.store.set_destination_last_successful_stream_ordering(DEST, 9)
        )
        self.assertEqual(
            self.rust.get_destination_last_successful_stream_ordering(DEST), 9
        )

    def test_catch_up_room_event_ids(self) -> None:
        orderings = [
            self._send(room_id, sticky=False)
            for room_id in self.rooms
            for _ in range(2)
        ]
        for last in [0, *orderings]:
            for destination in (DEST, OTHER, "unknown"):
                self.assertEqual(
                    self.rust.get_catch_up_room_event_ids(destination, last),
                    self.get_success(
                        self.store.get_catch_up_room_event_ids(destination, last)
                    ),
                )

    def test_catch_up_room_event_ids_limit(self) -> None:
        """More rooms are due than one batch takes."""
        for i in range(55):
            event_id = self.helper.send(self.rooms[0], f"m{i}", tok=self.tok)[
                "event_id"
            ]
            event = self.get_success(self.store.get_event(event_id))
            stream_ordering = event.internal_metadata.stream_ordering
            assert stream_ordering is not None
            # A room for each, as the catch-up query takes one event per room.
            room_id = f"!r{i}:test"
            self.get_success(
                self.store.db_pool.simple_insert(
                    "rooms", {"room_id": room_id, "is_public": False}
                )
            )
            self.get_success(
                self.store.store_destination_rooms_entries(
                    [DEST], room_id, stream_ordering
                )
            )
        python = self.get_success(self.store.get_catch_up_room_event_ids(DEST, 0))
        self.assertEqual(len(python), 50)
        self.assertEqual(self.rust.get_catch_up_room_event_ids(DEST, 0), python)

    def test_prev_events_for_room(self) -> None:
        for room_id in self.rooms:
            self._send(room_id, sticky=False)
        for room_id in [*self.rooms, "!unknown:test"]:
            self.assertEqual(
                self.rust.get_prev_events_for_room(room_id),
                self.get_success(self.store.get_prev_events_for_room(room_id)),
            )

    def _backlog(self) -> list[tuple[str, str, int]]:
        rows = self.get_success(
            self.store.db_pool.simple_select_list(
                "destination_rooms_sticky_events_backlog",
                None,
                ["destination", "room_id", "sticky_events_stream_position"],
            )
        )
        return sorted(rows)

    def _clear_backlog(self) -> None:
        self.get_success(
            self.store.db_pool.runInteraction(
                "clear_backlog",
                lambda txn: txn.execute(
                    "DELETE FROM destination_rooms_sticky_events_backlog"
                ),
            )
        )

    def test_mark_backlogged_sticky_events_after_catchup_transaction(self) -> None:
        orderings = []
        for i in range(9):
            orderings.append(self._send(self.rooms[i % 3], sticky=i % 2 == 0))

        cases = [
            (0, orderings[-1] + 1, []),
            (0, orderings[-1] + 1, orderings[::3]),
            (orderings[2], orderings[7], [orderings[4]]),
            (orderings[5], orderings[6], []),
        ]
        backlogged = False
        for old, new, sent in cases:
            # With no backlog, one before any new position, which stays, and
            # one further on, which is moved back.
            for existing in (None, 0, 1_000_000):
                self._clear_backlog()
                if existing is not None:
                    self.get_success(
                        self.store.db_pool.simple_insert(
                            "destination_rooms_sticky_events_backlog",
                            {
                                "destination": DEST,
                                "room_id": self.rooms[0],
                                "sticky_events_stream_position": existing,
                            },
                        )
                    )
                self.get_success(
                    self.store.mark_backlogged_sticky_events_after_catchup_transaction(
                        DEST,
                        old_last_successfully_sent_stream_ordering=old,
                        new_last_successfully_sent_stream_ordering=new,
                        event_stream_orderings_sent_in_transaction=sent,
                    )
                )
                python = self._backlog()
                if existing is None and python:
                    backlogged = True

                self._clear_backlog()
                if existing is not None:
                    self.get_success(
                        self.store.db_pool.simple_insert(
                            "destination_rooms_sticky_events_backlog",
                            {
                                "destination": DEST,
                                "room_id": self.rooms[0],
                                "sticky_events_stream_position": existing,
                            },
                        )
                    )
                self.rust.mark_backlogged_sticky_events_after_catchup_transaction(
                    DEST, old, new, sent
                )
                self.assertEqual(self._backlog(), python, (old, new, sent, existing))
        # The cases did backlog something of their own.
        self.assertTrue(backlogged)

    def _seed_backlog(self) -> list[int]:
        """Sticky and plain events in each room, with every room backlogged
        for `DEST` from its start."""
        orderings = [self._send(self.rooms[i % 3], sticky=i % 2 == 0) for i in range(9)]
        self.get_success(
            self.store.mark_backlogged_sticky_events_after_catchup_transaction(
                DEST,
                old_last_successfully_sent_stream_ordering=0,
                new_last_successfully_sent_stream_ordering=orderings[-1] + 1,
                event_stream_orderings_sent_in_transaction=[],
            )
        )
        return orderings

    def test_backlogged_sticky_events_for_destination(self) -> None:
        self._seed_backlog()
        now = self.clock.time_msec()
        for limit in (1, 2, 50):
            python = self.get_success(
                self.store.get_backlogged_sticky_events_for_destination(
                    DEST, limit=limit
                )
            )
            assert python is not None
            room_id, position, event_ids = python
            self.assertEqual(
                self.rust.get_backlogged_sticky_events_for_destination(
                    DEST, limit, now
                ),
                (room_id.to_string(), position, event_ids),
            )
        self.assertIsNone(
            self.rust.get_backlogged_sticky_events_for_destination(OTHER, 50, now)
        )

    def test_backlogged_sticky_events_expire(self) -> None:
        """Once the sticky events expire, the backlog is cleaned up."""
        self._seed_backlog()
        self.assertNotEqual(self._backlog(), [])
        later = self.clock.time_msec() + 60 * 60 * 1000
        self.assertIsNone(
            self.rust.get_backlogged_sticky_events_for_destination(DEST, 50, later)
        )
        self.assertEqual(self._backlog(), [])

    def test_backlogged_sticky_events_cleanup(self) -> None:
        """A room whose backlog is past its sticky events is cleaned up. One
        whose position is at its last sticky event is kept, and sent."""
        self._seed_backlog()
        rows = self.get_success(
            self.store.db_pool.simple_select_list(
                "sticky_events", None, ("room_id", "stream_id", "event_id")
            )
        )
        last: dict[str, tuple[int, str]] = {}
        for room_id, stream_id, event_id in rows:
            if room_id not in last or stream_id > last[room_id][0]:
                last[room_id] = (stream_id, event_id)
        # The backlog query takes the first room it finds, which is by room
        # ID. Leave the last room with something to send, so the others are
        # cleaned up first.
        *empty, kept = sorted(self.rooms)
        seeded = [(DEST, room_id, last[room_id][0] + 1) for room_id in empty]
        seeded.append((DEST, kept, last[kept][0]))
        now = self.clock.time_msec()

        self._clear_backlog()
        self._seed_backlog_rows(seeded)
        python = self.get_success(
            self.store.get_backlogged_sticky_events_for_destination(DEST, limit=50)
        )
        assert python is not None
        python_rows = self._backlog()

        self._clear_backlog()
        self._seed_backlog_rows(seeded)
        rust = self.rust.get_backlogged_sticky_events_for_destination(DEST, 50, now)
        self.assertEqual(rust, (kept, last[kept][0], [last[kept][1]]))
        self.assertEqual((python[0].to_string(), python[1], python[2]), rust)
        self.assertEqual(self._backlog(), python_rows)
        self.assertEqual(self._backlog(), [(DEST, kept, last[kept][0])])

    def test_mark_backlogged_sticky_events_sent(self) -> None:
        self._seed_backlog()
        backlog = self.rust.get_backlogged_sticky_events_for_destination(
            DEST, 50, self.clock.time_msec()
        )
        assert backlog is not None
        room_id, position, _ = backlog

        self.rust.mark_backlogged_sticky_events_sent(DEST, room_id, position)
        rust = self._backlog()
        self._clear_backlog()
        self._seed_backlog_rows(rust)
        self.get_success(
            self.store.mark_backlogged_sticky_events_sent(
                DEST, RoomID.from_string(room_id), StickyEventStreamPosition(position)
            )
        )
        self.assertEqual(self._backlog(), rust)
        self.assertIn((DEST, room_id, position + 1), rust)

    def _seed_backlog_rows(self, rows: list[tuple[str, str, int]]) -> None:
        for destination, room_id, position in rows:
            self.get_success(
                self.store.db_pool.simple_upsert(
                    "destination_rooms_sticky_events_backlog",
                    {"destination": destination, "room_id": room_id},
                    {"sticky_events_stream_position": position},
                )
            )

    def test_catch_up_outstanding_destinations(self) -> None:
        # PDUs for `DEST` and `OTHER`, a to-device message for a third, a
        # sticky backlog for a fourth, and a fifth that is backed off.
        self._send(self.rooms[0], sticky=False)
        self.get_success(
            self.store.add_remote_messages_from_client_to_device_inbox(
                {
                    "todevice.example.com": {
                        "sender": "@alice:test",
                        "type": "m.test",
                        "message_id": "1",
                        "messages": {"@u:todevice.example.com": {"D": {}}},
                    }
                }
            )
        )
        self._seed_backlog_rows([("sticky.example.com", self.rooms[0], 1)])
        self.get_success(
            self.store.store_destination_rooms_entries(
                ["backoff.example.com"], self.rooms[0], 1_000_000
            )
        )
        now = self.clock.time_msec()
        self.get_success(
            self.store.set_destination_retry_timings(
                "backoff.example.com", now, now, 60_000
            )
        )

        for after in (None, "", DEST, "zzz"):
            for at in (now, now + 120_000):
                # Python's reads the clock, so call its transaction directly.
                python = self.get_success(
                    self.store.db_pool.runInteraction(
                        "outstanding",
                        self.store._get_catch_up_outstanding_destinations_txn,
                        at,
                        after,
                    )
                )
                self.assertEqual(
                    self.rust.get_catch_up_outstanding_destinations(after, at),
                    python,
                    (after, at),
                )
        self.assertIn(
            "backoff.example.com",
            self.rust.get_catch_up_outstanding_destinations(None, now + 120_000),
        )

        # More than one page.
        self.get_success(
            self.store.store_destination_rooms_entries(
                [f"d{i:02}.example.com" for i in range(30)], self.rooms[0], 1_000_000
            )
        )
        after = None
        pages = []
        while True:
            python = self.get_success(
                self.store.db_pool.runInteraction(
                    "outstanding",
                    self.store._get_catch_up_outstanding_destinations_txn,
                    now,
                    after,
                )
            )
            self.assertEqual(
                self.rust.get_catch_up_outstanding_destinations(after, now), python
            )
            if not python:
                break
            pages.append(python)
            after = python[-1]
        self.assertGreater(len(pages), 1)
        self.assertEqual(len(pages[0]), 25)
        self.assertNotIn(
            "backoff.example.com",
            self.rust.get_catch_up_outstanding_destinations(None, now),
        )