Update device inbox stream change caches before advancing ID generator

Author: Fizzadar

synapse/storage/databases/main/deviceinbox.py

@@ -138,16 +138,23 @@ def process_replication_rows(
         if stream_name == ToDeviceStream.NAME:
             # If replication is happening than postgres must be being used.
             assert isinstance(self._device_inbox_id_gen, MultiWriterIdGenerator)
-            self._device_inbox_id_gen.advance(instance_name, token)
             for row in rows:
+                # NOTE: here we always tell both stream change caches, either about
+                # the entity or just the known position.
                 if row.entity.startswith("@"):
                     self._device_inbox_stream_cache.entity_has_changed(
                         row.entity, token
                     )
+                    self._device_federation_outbox_stream_cache.have_seen_position(
+                        token
+                    )
                 else:
+                    self._device_inbox_stream_cache.have_seen_position(token)
                     self._device_federation_outbox_stream_cache.entity_has_changed(
                         row.entity, token
                     )
+            # Important that the ID gen advances after stream change caches
+            self._device_inbox_id_gen.advance(instance_name, token)
         return super().process_replication_rows(stream_name, instance_name, token, rows)
 
     def get_to_device_stream_token(self) -> int:

synapse/util/caches/stream_change_cache.py

@@ -215,6 +215,9 @@ def entity_has_changed(self, entity: EntityType, stream_pos: int) -> None:
         self._entity_to_key[entity] = stream_pos
         self._evict()
 
+        self.have_seen_position(stream_pos)
+
+    def have_seen_position(self, stream_pos: int) -> None:
         if stream_pos > self.max_stream_pos:
             self.max_stream_pos = stream_pos