Fix crash when adding an index with new subscriptions in runtime (#429)

Author: droserasprout

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog],
 and this project adheres to [Semantic Versioning].
 
+## [Unreleased]
+
+### Fixed
+
+- index: Fixed crash when adding an index with new subscriptions in runtime.
+
 ## [5.2.4] - 2022-07-17
 
 ### Fixed

src/dipdup/datasources/subscription.py

@@ -53,10 +53,7 @@ def missing_subscriptions(self) -> Set[Subscription]:
         return {k for k, v in self._subscriptions.items() if k is not None and v is None}
 
     def add(self, subscription: Subscription) -> None:
-        if subscription in self._subscriptions:
-            if not self._merge_subscriptions:
-                _logger.warning(f'Subscription already exists: {subscription}')
-        else:
+        if subscription not in self._subscriptions:
             self._subscriptions[subscription] = None
 
     def remove(self, subscription: Subscription) -> None:

src/dipdup/index.py

@@ -209,13 +209,15 @@ async def process(self) -> None:
                 await self._synchronize(head_level)
             await self.state.update_status(IndexStatus.ONESHOT, head_level)
 
+        index_level = self.state.level
         sync_levels = {self.datasource.get_sync_level(s) for s in self._config.subscriptions}
-        sync_level = sync_levels.pop()
-        if sync_level is None:
+        if not sync_levels:
+            raise RuntimeError('Index has no subscriptions')
+        if None in sync_levels:
             raise RuntimeError('Call `set_sync_level` before starting IndexDispatcher')
-        if sync_levels:
-            raise RuntimeError(f'Multiple sync levels: {sync_level}, {sync_levels}')
-        index_level = self.state.level
+        # NOTE: Multiple sync levels means index with new subscriptions was added in runtime.
+        # NOTE: Choose the highest level; outdated realtime messages will be dropped from the queue anyway.
+        sync_level = max(cast(Set[int], sync_levels))
 
         if index_level < sync_level:
             self._logger.info('Index is behind datasource level, syncing: %s -> %s', index_level, sync_level)