fix(gateway): fix Matrix E2EE silent failures after mautrix migration

gateway/platforms/matrix.py

@@ -192,6 +192,43 @@ async def find_shared_rooms(self, user_id: str) -> list:
         return list(self._joined_rooms)
 
 
+class _HermesDecryptionDispatcher:
+    """Decrypt ROOM_ENCRYPTED events; buffer failures for retry.
+
+    Replaces the default mautrix DecryptionDispatcher + a separate
+    _on_encrypted_event handler.  A single handler eliminates the dedup
+    race where _on_encrypted_event (zero awaits) marks the event_id
+    before the async DecryptionDispatcher can dispatch the decrypted event.
+    """
+
+    event_type = EventType.ROOM_ENCRYPTED
+
+    def __init__(self, client: Any, adapter: "MatrixAdapter"):
+        self.client = client
+        self._adapter = adapter
+
+    def register(self) -> None:
+        self.client.add_event_handler(self.event_type, self.handle)
+
+    async def handle(self, evt: Any) -> None:
+        try:
+            decrypted = await self.client.crypto.decrypt_megolm_event(evt)
+        except Exception:
+            room_id = str(getattr(evt, "room_id", ""))
+            event_id = str(getattr(evt, "event_id", ""))
+            logger.warning(
+                "Matrix: could not decrypt event %s in %s — buffering for retry",
+                event_id, room_id,
+            )
+            self._adapter._pending_megolm.append((room_id, evt, time.time()))
+            if len(self._adapter._pending_megolm) > _MAX_PENDING_EVENTS:
+                self._adapter._pending_megolm = (
+                    self._adapter._pending_megolm[-_MAX_PENDING_EVENTS:]
+                )
+            return
+        self.client.dispatch_event(decrypted, evt.source)
+
+
 class MatrixAdapter(BasePlatformAdapter):
     """Gateway adapter for Matrix (any homeserver)."""
 
@@ -364,6 +401,35 @@ async def _verify_device_keys_on_server(self, client: Any, olm: Any) -> bool:
                 )
                 return False
 
+            # Re-verify: the server may accept the OTKs (HTTP 200) but
+            # silently ignore new device keys when identity keys are
+            # immutable for the existing device.
+            try:
+                resp2 = await client.query_keys({client.mxid: [client.device_id]})
+                dk2 = (getattr(resp2, "device_keys", {}) or {})
+                ud2 = (dk2.get(str(client.mxid)) or {})
+                keys2 = ud2.get(str(client.device_id))
+                if keys2:
+                    server_ed2 = None
+                    for kid, kval in (getattr(keys2, "keys", {}) or {}).items():
+                        if str(kid).startswith("ed25519:"):
+                            server_ed2 = str(kval)
+                            break
+                    if server_ed2 != local_ed25519:
+                        logger.error(
+                            "Matrix: device %s has immutable identity keys on the "
+                            "server that don't match this installation. Generate a "
+                            "new access token with a fresh device: "
+                            "hermes gateway setup --platform matrix",
+                            client.device_id,
+                        )
+                        return False
+            except Exception as exc:
+                logger.error(
+                    "Matrix: failed to re-verify device keys after upload: %s", exc,
+                )
+                return False
+
         return True
 
     # ------------------------------------------------------------------
@@ -509,6 +575,13 @@ async def connect(self) -> bool:
                     return False
 
                 client.crypto = olm
+
+                # Replace the auto-registered DecryptionDispatcher with our
+                # single-path handler that buffers failures for retry.
+                from mautrix.client.client import DecryptionDispatcher
+                client.remove_dispatcher(DecryptionDispatcher)
+                _HermesDecryptionDispatcher(client, self).register()
+
                 logger.info(
                     "Matrix: E2EE enabled (store: %s%s)",
                     str(_CRYPTO_DB_PATH),
@@ -529,9 +602,6 @@ async def connect(self) -> bool:
         client.add_event_handler(EventType.REACTION, self._on_reaction)
         client.add_event_handler(IntEvt.INVITE, self._on_invite)
 
-        if self._encryption and getattr(client, "crypto", None):
-            client.add_event_handler(EventType.ROOM_ENCRYPTED, self._on_encrypted_event)
-
         # Initial sync to catch up, then start background sync.
         self._startup_ts = time.time()
         self._closing = False
@@ -540,7 +610,8 @@ async def connect(self) -> bool:
             sync_data = await client.sync(timeout=10000, full_state=True)
             if isinstance(sync_data, dict):
                 rooms_join = sync_data.get("rooms", {}).get("join", {})
-                self._joined_rooms = set(rooms_join.keys())
+                self._joined_rooms.clear()
+                self._joined_rooms.update(rooms_join.keys())
                 # Store the next_batch token so incremental syncs start
                 # from where the initial sync left off.
                 nb = sync_data.get("next_batch")
@@ -1020,12 +1091,6 @@ async def _retry_pending_decryptions(self) -> None:
                 getattr(event, "event_id", "?"),
             )
 
-            # Route to the appropriate handler.
-            # Remove from dedup set so _on_room_message doesn't drop it
-            # (the encrypted event ID was already registered by _on_encrypted_event).
-            decrypted_id = str(getattr(decrypted, "event_id", getattr(event, "event_id", "")))
-            if decrypted_id:
-                self._processed_events_set.discard(decrypted_id)
             try:
                 await self._on_room_message(decrypted)
             except Exception as exc:
@@ -1357,23 +1422,6 @@ async def _handle_media_message(
 
         await self.handle_message(msg_event)
 
-    async def _on_encrypted_event(self, event: Any) -> None:
-        """Handle encrypted events that could not be auto-decrypted."""
-        room_id = str(getattr(event, "room_id", ""))
-        event_id = str(getattr(event, "event_id", ""))
-
-        if self._is_duplicate_event(event_id):
-            return
-
-        logger.warning(
-            "Matrix: could not decrypt event %s in %s — buffering for retry",
-            event_id, room_id,
-        )
-
-        self._pending_megolm.append((room_id, event, time.time()))
-        if len(self._pending_megolm) > _MAX_PENDING_EVENTS:
-            self._pending_megolm = self._pending_megolm[-_MAX_PENDING_EVENTS:]
-
     async def _on_invite(self, event: Any) -> None:
         """Auto-join rooms when invited."""
 

tests/gateway/test_matrix.py

@@ -108,13 +108,25 @@ def __init__(self, mxid=None, device_id=None, api=None,
         def add_event_handler(self, event_type, handler):
             self._event_handlers.setdefault(event_type, []).append(handler)
 
+        def remove_dispatcher(self, dispatcher_cls):
+            pass
+
     class InternalEventType:
         INVITE = "internal.invite"
 
     mautrix_client.Client = Client
     mautrix_client.InternalEventType = InternalEventType
     mautrix.client = mautrix_client
 
+    # --- mautrix.client.client (needed for DecryptionDispatcher import) ---
+    mautrix_client_client = types.ModuleType("mautrix.client.client")
+
+    class DecryptionDispatcher:
+        event_type = "m.room.encrypted"
+
+    mautrix_client_client.DecryptionDispatcher = DecryptionDispatcher
+    mautrix_client.client = mautrix_client_client
+
     # --- mautrix.client.state_store ---
     mautrix_client_state_store = types.ModuleType("mautrix.client.state_store")
 
@@ -200,6 +212,7 @@ def create(cls, url, upgrade_table=None):
         "mautrix.api": mautrix_api,
         "mautrix.types": mautrix_types,
         "mautrix.client": mautrix_client,
+        "mautrix.client.client": mautrix_client_client,
         "mautrix.client.state_store": mautrix_client_state_store,
         "mautrix.crypto": mautrix_crypto,
         "mautrix.crypto.store": mautrix_crypto_store,
@@ -1142,26 +1155,34 @@ async def test_send_retries_after_e2ee_error(self):
 
 
 # ---------------------------------------------------------------------------
-# E2EE: MegolmEvent key request + buffering via _on_encrypted_event
+# E2EE: MegolmEvent buffering via _HermesDecryptionDispatcher
 # ---------------------------------------------------------------------------
 
 class TestMatrixMegolmEventHandling:
     @pytest.mark.asyncio
     async def test_encrypted_event_buffers_for_retry(self):
-        """_on_encrypted_event should buffer undecrypted events for retry."""
+        """_HermesDecryptionDispatcher should buffer undecryptable events."""
+        from gateway.platforms.matrix import _HermesDecryptionDispatcher
+
         adapter = _make_adapter()
         adapter._user_id = "@bot:example.org"
         adapter._startup_ts = 0.0
         adapter._dm_rooms = {}
 
+        mock_client = MagicMock()
+        mock_client.crypto = MagicMock()
+        mock_client.crypto.decrypt_megolm_event = AsyncMock(side_effect=Exception("no session"))
+
+        dispatcher = _HermesDecryptionDispatcher(mock_client, adapter)
+
         fake_event = MagicMock()
         fake_event.room_id = "!room:example.org"
         fake_event.event_id = "$encrypted_event"
         fake_event.sender = "@alice:example.org"
+        fake_event.source = {}
 
-        await adapter._on_encrypted_event(fake_event)
+        await dispatcher.handle(fake_event)
 
-        # Should have buffered the event
         assert len(adapter._pending_megolm) == 1
         room_id, event, ts = adapter._pending_megolm[0]
         assert room_id == "!room:example.org"
@@ -1170,19 +1191,26 @@ async def test_encrypted_event_buffers_for_retry(self):
     @pytest.mark.asyncio
     async def test_encrypted_event_buffer_capped(self):
         """Buffer should not grow past _MAX_PENDING_EVENTS."""
+        from gateway.platforms.matrix import _HermesDecryptionDispatcher, _MAX_PENDING_EVENTS
+
         adapter = _make_adapter()
         adapter._user_id = "@bot:example.org"
         adapter._startup_ts = 0.0
         adapter._dm_rooms = {}
 
-        from gateway.platforms.matrix import _MAX_PENDING_EVENTS
+        mock_client = MagicMock()
+        mock_client.crypto = MagicMock()
+        mock_client.crypto.decrypt_megolm_event = AsyncMock(side_effect=Exception("no session"))
+
+        dispatcher = _HermesDecryptionDispatcher(mock_client, adapter)
 
         for i in range(_MAX_PENDING_EVENTS + 10):
             evt = MagicMock()
             evt.room_id = "!room:example.org"
             evt.event_id = f"$event_{i}"
             evt.sender = "@alice:example.org"
-            await adapter._on_encrypted_event(evt)
+            evt.source = {}
+            await dispatcher.handle(evt)
 
         assert len(adapter._pending_megolm) == _MAX_PENDING_EVENTS
 
@@ -1327,11 +1355,13 @@ async def test_connect_registers_encrypted_event_handler_when_encryption_on(self
                         assert await adapter.connect() is True
 
         # Verify event handlers were registered.
-        # In mautrix the order is: add_event_handler(EventType, callback)
+        # With _HermesDecryptionDispatcher, ROOM_ENCRYPTED is registered via
+        # dispatcher.register() (add_event_handler), not the old _on_encrypted_event.
         handler_calls = mock_client.add_event_handler.call_args_list
         registered_types = [call.args[0] for call in handler_calls]
 
-        # Should have registered handlers for ROOM_MESSAGE, REACTION, INVITE, and ROOM_ENCRYPTED
+        # Should have: ROOM_MESSAGE, REACTION, INVITE (3 explicit)
+        # plus ROOM_ENCRYPTED via _HermesDecryptionDispatcher.register()
         assert len(handler_calls) >= 4  # At minimum these four
 
         await adapter.disconnect()