Catch up MSC4354 Sticky Events to remote homeservers that have missed them.

[reivilibre]

Part of: MSC4354 (Sticky Events)

Our locally-sent Sticky Events need to be reliably (within bounds of possibility) sent to remote homeservers.

This PR adds the mechanism to make this happen.

Notable omissions:

• catching up historical sticky events to homeservers that only just join the room

---

Review Guidance

The following 2 commits are refactorings and hopefully stand alone, with no behavioural changes. I suggest considering them on their own.

• Replace _TransactionQueueManager with prepare/complete transaction methods
• Pull out user_is_local_like_pattern

After that, I recommend looking at the database schema (delta introducing a table) to understand the data model
for tracking backlogged sticky events. That's more or less the central part of this PR.

It may be helpful to remember that Sticky Events are events that are temporarily 'promoted' such that we want to ensure that all users and all homeservers receive them,
without letting them fall down 'gaps' of unretrieved history. Essentially, it's a temporary delivery reliability increase.

There are then roughly three logical pieces to review:

• how, when sending federation transactions, we retrieve events from the sticky event backlog and record that
  we have succeeded in sending them
• how we become aware of a new backlog of sticky events. (Bear in mind that, unless the remote homeserver goes down, the intention
  is for sticky events to not have any special processing)
  The docstring on mark_backlogged_sticky_events_after_catchup_transaction contains a diagrammatical explanation of this
  mechanism and how it works.
• how we round-robin between sending out sticky event backlog transactions and 'main queue' (normal) transactions containing
  real-time PDUs and EDUs

Or if you prefer, it may be easiest to just start looking at the storage changes
to get an idea of the storage contract exposed by the database layer, since
the federation transmission logic itself is (I think) fairly understandable,
as long as you trust that contract.

Happy to pair, explain, add comments, anything; just let me know!

---

1. Replace _TransactionQueueManager with prepare/complete transaction methods
   Split out at Refactor the federation transmission code to delineate transaction preparation and completion. #20166

2. Pull out user_is_local_like_pattern

3. unwrap utility

4. Add destination_room_sticky_events_backlog table

5. Introduce a StickyEventStreamPosition NewType

6. Add store methods

7. Wire sticky event backlog transactions into the per-destination queue

8. Add mark of backlog after catch-up transaction

9. Add tests

[github-actions]

Schema Diff

Please check that this looks as expected!

common (sqlite)

Unchanged

common (postgres)

Unchanged

main (sqlite)

--- /tmp/schema_diff_kgtpsnw9/before/main/full_schemas/9999/full.sql.sqlite	2026-09-01 10:44:01.740546407 +0000
+++ /tmp/schema_diff_kgtpsnw9/after/main/full_schemas/9999/full.sql.sqlite	2026-09-01 10:43:47.334400460 +0000
@@ -1113,20 +1113,45 @@
 CREATE INDEX sticky_events_room_idx ON sticky_events (room_id, event_stream_ordering);
 CREATE TABLE sliding_sync_connection_lazy_members (
     connection_key BIGINT NOT NULL REFERENCES sliding_sync_connections(connection_key) ON DELETE CASCADE,
     connection_position BIGINT REFERENCES sliding_sync_connection_positions(connection_position) ON DELETE CASCADE,
     room_id TEXT NOT NULL,
     user_id TEXT NOT NULL,
     last_seen_ts BIGINT NOT NULL
 );
 CREATE UNIQUE INDEX sliding_sync_connection_lazy_members_idx ON sliding_sync_connection_lazy_members (connection_key, room_id, user_id);
 CREATE INDEX sliding_sync_connection_lazy_members_pos_idx ON sliding_sync_connection_lazy_members (connection_key, connection_position) WHERE connection_position IS NOT NULL;
+CREATE TABLE destination_room_sticky_events_backlog (
+    -- Server name of the remote homeserver.
+    destination TEXT NOT NULL,
+
+    -- Room ID in which sticky events have been missed.
+    room_id TEXT NOT NULL
+        -- Only track this information for rooms we know about;
+        -- if we delete a room locally then also delete the tracking info.
+        REFERENCES rooms(room_id) ON DELETE CASCADE,
+
+    -- Position in the sticky events stream, corresponding to the
+    -- `sticky_events.stream_id` of the first sticky event that
+    -- has yet to be sent.
+    --
+    -- Because sticky events must be sent in order (as per MSC4354),
+    -- all subsequent sticky events for the same room with higher
+    -- `stream_id`s are also unsent.
+    --
+    -- Not a foreign key because we must still support expiration of sticky events.
+    sticky_events_stream_position INTEGER NOT NULL,
+
+    -- It's enough to track one position per (destination, room_id) pair.
+    PRIMARY KEY (destination, room_id)
+);
+CREATE INDEX destination_room_sticky_events_backlog_room_id ON destination_room_sticky_events_backlog (room_id);
 CREATE TABLE quarantined_media_changes (
     -- Position in the quarantined media stream
     stream_id INTEGER NOT NULL PRIMARY KEY,
 
     -- Name of the worker sending this (makes us compatible with multiple writers)
     instance_name TEXT NOT NULL,
 
     -- Media origin. NULL if local media.
     -- We store the origin and media_id as media is scoped to the origin and are uniquely identified by (origin, media_id).
     origin TEXT NULL,

main (postgres)

--- /tmp/schema_diff_kgtpsnw9/before/main/full_schemas/9999/full.sql.postgres	2026-09-01 10:44:02.463551993 +0000
+++ /tmp/schema_diff_kgtpsnw9/after/main/full_schemas/9999/full.sql.postgres	2026-09-01 10:43:48.131409717 +0000
@@ -193,20 +193,25 @@
     stream_id bigint NOT NULL,
     CONSTRAINT delayed_events_stream_pos_lock_check CHECK ((lock = 'X'::bpchar))
 );
 CREATE TABLE deleted_pushers (
     stream_id bigint NOT NULL,
     app_id text NOT NULL,
     pushkey text NOT NULL,
     user_id text NOT NULL,
     instance_name text
 );
+CREATE TABLE destination_room_sticky_events_backlog (
+    destination text NOT NULL,
+    room_id text NOT NULL,
+    sticky_events_stream_position integer NOT NULL
+);
 CREATE TABLE destination_rooms (
     destination text NOT NULL,
     room_id text NOT NULL,
     stream_ordering bigint NOT NULL
 );
 COMMENT ON TABLE destination_rooms IS 'Information about transmission of PDUs in a given room to a given remote homeserver.';
 COMMENT ON COLUMN destination_rooms.destination IS 'server name of remote homeserver in question';
 COMMENT ON COLUMN destination_rooms.room_id IS 'room ID in question';
 COMMENT ON COLUMN destination_rooms.stream_ordering IS '`stream_ordering` of the most recent PDU in this room that needs to be sent (by us) to this homeserver.
 This can only be pointing to our own PDU because we are only responsible for sending our own PDUs.';
@@ -1431,20 +1436,22 @@
 ALTER TABLE ONLY current_state_events
     ADD CONSTRAINT current_state_events_event_id_key UNIQUE (event_id);
 ALTER TABLE ONLY current_state_events
     ADD CONSTRAINT current_state_events_room_id_type_state_key_key UNIQUE (room_id, type, state_key);
 ALTER TABLE ONLY dehydrated_devices
     ADD CONSTRAINT dehydrated_devices_pkey PRIMARY KEY (user_id);
 ALTER TABLE ONLY delayed_events
     ADD CONSTRAINT delayed_events_pkey PRIMARY KEY (user_localpart, delay_id);
 ALTER TABLE ONLY delayed_events_stream_pos
     ADD CONSTRAINT delayed_events_stream_pos_lock_key UNIQUE (lock);
+ALTER TABLE ONLY destination_room_sticky_events_backlog
+    ADD CONSTRAINT destination_room_sticky_events_backlog_pkey PRIMARY KEY (destination, room_id);
 ALTER TABLE ONLY destination_rooms
     ADD CONSTRAINT destination_rooms_pkey PRIMARY KEY (destination, room_id);
 ALTER TABLE ONLY destinations
     ADD CONSTRAINT destinations_pkey PRIMARY KEY (destination);
 ALTER TABLE ONLY device_lists_changes_converted_stream_position
     ADD CONSTRAINT device_lists_changes_converted_stream_position_lock_key UNIQUE (lock);
 ALTER TABLE ONLY device_lists_changes_in_room_max_pruned_stream_id
     ADD CONSTRAINT device_lists_changes_in_room_max_pruned_stream_id_lock_key UNIQUE (lock);
 ALTER TABLE ONLY device_lists_remote_pending
     ADD CONSTRAINT device_lists_remote_pending_pkey PRIMARY KEY (stream_id);
@@ -1643,20 +1650,21 @@
 CREATE INDEX current_state_delta_stream_idx ON current_state_delta_stream USING btree (stream_id);
 CREATE INDEX current_state_delta_stream_room_idx ON current_state_delta_stream USING btree (room_id, stream_id);
 CREATE INDEX current_state_events_member_index ON current_state_events USING btree (state_key) WHERE (type = 'm.room.member'::text);
 CREATE INDEX current_state_events_members_room_index ON current_state_events USING btree (room_id, membership) WHERE (type = 'm.room.member'::text);
 CREATE INDEX current_state_events_stream_ordering_idx ON current_state_events USING btree (event_stream_ordering);
 CREATE UNIQUE INDEX delayed_events_idx ON delayed_events USING btree (delay_id);
 CREATE INDEX delayed_events_is_processed ON delayed_events USING btree (is_processed);
 CREATE INDEX delayed_events_room_state_event_idx ON delayed_events USING btree (room_id, event_type, state_key) WHERE (state_key IS NOT NULL);
 CREATE INDEX delayed_events_send_ts ON delayed_events USING btree (send_ts);
 CREATE INDEX deleted_pushers_stream_id ON deleted_pushers USING btree (stream_id);
+CREATE INDEX destination_room_sticky_events_backlog_room_id ON destination_room_sticky_events_backlog USING btree (room_id);
 CREATE INDEX destination_rooms_room_id ON destination_rooms USING btree (room_id);
 CREATE INDEX device_auth_providers_devices ON device_auth_providers USING btree (user_id, device_id);
 CREATE INDEX device_auth_providers_sessions ON device_auth_providers USING btree (auth_provider_id, auth_provider_session_id);
 CREATE INDEX device_federation_inbox_received_ts_index ON device_federation_inbox USING btree (received_ts);
 CREATE INDEX device_federation_inbox_sender_id ON device_federation_inbox USING btree (origin, message_id);
 CREATE INDEX device_federation_outbox_destination_id ON device_federation_outbox USING btree (destination, stream_id);
 CREATE INDEX device_federation_outbox_id ON device_federation_outbox USING btree (stream_id);
 CREATE INDEX device_inbox_stream_id_user_id ON device_inbox USING btree (stream_id, user_id);
 CREATE INDEX device_inbox_user_stream_id ON device_inbox USING btree (user_id, device_id, stream_id);
 CREATE INDEX device_lists_changes_in_room_by_room_idx ON device_lists_changes_in_room USING btree (room_id, stream_id);
@@ -1846,20 +1854,22 @@
 CREATE UNIQUE INDEX worker_read_write_locks_mode_type ON worker_read_write_locks_mode USING btree (lock_name, lock_key, write_lock);
 CREATE UNIQUE INDEX worker_read_write_locks_write ON worker_read_write_locks USING btree (lock_name, lock_key) WHERE write_lock;
 CREATE TRIGGER check_event_stream_ordering BEFORE INSERT OR UPDATE ON current_state_events FOR EACH ROW EXECUTE FUNCTION check_event_stream_ordering();
 CREATE TRIGGER check_event_stream_ordering BEFORE INSERT OR UPDATE ON local_current_membership FOR EACH ROW EXECUTE FUNCTION check_event_stream_ordering();
 CREATE TRIGGER check_event_stream_ordering BEFORE INSERT OR UPDATE ON room_memberships FOR EACH ROW EXECUTE FUNCTION check_event_stream_ordering();
 CREATE TRIGGER check_partial_state_events BEFORE INSERT OR UPDATE ON partial_state_events FOR EACH ROW EXECUTE FUNCTION check_partial_state_events();
 CREATE TRIGGER delete_read_write_lock_parent_trigger AFTER DELETE ON worker_read_write_locks FOR EACH ROW EXECUTE FUNCTION delete_read_write_lock_parent();
 CREATE TRIGGER upsert_read_write_lock_parent_trigger BEFORE INSERT ON worker_read_write_locks FOR EACH ROW EXECUTE FUNCTION upsert_read_write_lock_parent();
 ALTER TABLE ONLY access_tokens
     ADD CONSTRAINT access_tokens_refresh_token_id_fkey FOREIGN KEY (refresh_token_id) REFERENCES refresh_tokens(id) ON DELETE CASCADE;
+ALTER TABLE ONLY destination_room_sticky_events_backlog
+    ADD CONSTRAINT destination_room_sticky_events_backlog_room_id_fkey FOREIGN KEY (room_id) REFERENCES rooms(room_id) ON DELETE CASCADE;
 ALTER TABLE ONLY destination_rooms
     ADD CONSTRAINT destination_rooms_destination_fkey FOREIGN KEY (destination) REFERENCES destinations(destination);
 ALTER TABLE ONLY destination_rooms
     ADD CONSTRAINT destination_rooms_room_id_fkey FOREIGN KEY (room_id) REFERENCES rooms(room_id);
 ALTER TABLE ONLY event_edges
     ADD CONSTRAINT event_edges_event_id_fkey FOREIGN KEY (event_id) REFERENCES events(event_id);
 ALTER TABLE ONLY event_failed_pull_attempts
     ADD CONSTRAINT event_failed_pull_attempts_room_id_fkey FOREIGN KEY (room_id) REFERENCES rooms(room_id);
 ALTER TABLE ONLY event_forward_extremities
     ADD CONSTRAINT event_forward_extremities_event_id FOREIGN KEY (event_id) REFERENCES events(event_id) DEFERRABLE INITIALLY DEFERRED;

state (sqlite)

Unchanged

state (postgres)

Unchanged