Try to avoid race conditions with supergroup upgrades

mautrix_telegram/abstract_user.py

@@ -596,15 +596,27 @@ class AbstractUser(ABC):
             await portal.delete_telegram_user(self.tgid, sender=None)
         elif chan := getattr(update, "mau_channel", None):
             if not portal.mxid:
-                self.log.info(
-                    "Creating Matrix room with data fetched by Telethon due to UpdateChannel"
-                )
-                await portal.create_matrix_room(self, chan)
+                asyncio.create_task(self._delayed_create_channel(chan))
             else:
                 self.log.debug("Updating channel info with data fetched by Telethon")
                 await portal.update_info(self, chan)
                 await portal.invite_to_matrix(self.mxid)
 
+    async def _delayed_create_channel(self, chan: Channel) -> None:
+        self.log.debug("Waiting 5 seconds before handling UpdateChannel for non-existent portal")
+        await asyncio.sleep(5)
+        portal = await po.Portal.get_by_tgid(TelegramID(chan.id))
+        if portal.mxid:
+            self.log.debug(
+                "Portal started existing after waiting 5 seconds, dropping UpdateChannel"
+            )
+            return
+        else:
+            self.log.info(
+                "Creating Matrix room with data fetched by Telethon due to UpdateChannel"
+            )
+            await portal.create_matrix_room(self, chan)
+
     async def update_message(self, original_update: UpdateMessage) -> None:
         update, sender, portal = await self.get_message_details(original_update)
         if not portal:

mautrix_telegram/portal.py

@@ -480,6 +480,11 @@ class Portal(DBPortal, BasePortal):
         await self.update_info(source, entity)
 
     async def _migrate_and_save_telegram(self, new_id: TelegramID) -> None:
+        async with self._async_get_locks[(new_id,)]:
+            await self._migrate_and_save_telegram_locked(new_id)
+
+    async def _migrate_and_save_telegram_locked(self, new_id: TelegramID) -> None:
+        self.log.info(f"Starting migration to {new_id}")
         try:
             del self.by_tgid[self.tgid_full]
         except KeyError:
@@ -490,7 +495,12 @@ class Portal(DBPortal, BasePortal):
             existing = None
         self.by_tgid[(new_id, new_id)] = self
         if existing:
-            await existing.delete()
+            if existing.mxid:
+                self.log.warning(f"Deleting existing portal room {existing.mxid} for {new_id}")
+                await existing.cleanup_and_delete()
+            else:
+                self.log.debug(f"Deleting old database entry for {new_id}")
+                await existing.delete()
         old_id = self.tgid
         await self.update_id(new_id, "channel")
         self.log = self.__class__.log.getChild(self.tgid_log)