NiceGUI has Redis connection leak via tab storage causes service degradation

@@ -304,6 +304,7 @@ def handle_disconnect(self, socket_id: str) -> None:
         document_id = self._socket_to_document_id.pop(socket_id)
         self._cancel_delete_task(document_id)
         self._num_connections[document_id] -= 1
+        tab_id_to_close = self.tab_id
         self.tab_id = None
 
         for t in self.disconnect_handlers:
@@ -316,6 +317,7 @@ async def delete_content() -> None:
             if self._num_connections[document_id] == 0:
                 self._num_connections.pop(document_id)
                 self._delete_tasks.pop(document_id)
+                await core.app.storage.close_tab(tab_id_to_close)
                 self.delete()
         self._delete_tasks[document_id] = \
             background_tasks.create(delete_content(), name=f'delete content {document_id}')

@@ -1,11 +1,14 @@
+import asyncio
+import contextlib
+from typing import Optional
+
 from .. import background_tasks, core, json, optional_features
 from ..logging import log
 from .persistent_dict import PersistentDict
 
 try:
     import redis as redis_sync
     import redis.asyncio as redis
-    import redis.exceptions as redis_exceptions
     optional_features.register('redis')
 except ImportError:
     pass
@@ -26,7 +29,7 @@ def __init__(self, *, url: str, id: str, key_prefix: str = 'nicegui:') -> None:
         self.redis_client = redis.from_url(self.url, **self._redis_client_params)
         self.pubsub = self.redis_client.pubsub()
         self.key = key_prefix + id
-        self._should_listen = True
+        self._listener_task: Optional[asyncio.Task] = None
         super().__init__(data={}, on_change=self.publish)
 
     async def initialize(self) -> None:
@@ -51,12 +54,7 @@ def initialize_sync(self) -> None:
     def _start_listening(self) -> None:
         async def listen():
             try:
-                if not self._should_listen:
-                    return
                 await self.pubsub.subscribe(self.key + 'changes')
-                if not self._should_listen:
-                    await self.pubsub.unsubscribe()
-                    return
                 async for message in self.pubsub.listen():
                     t = message['type']
                     if t == 'message':
@@ -65,15 +63,21 @@ async def listen():
                             self.update(new_data)
                     elif t in ('unsubscribe', 'punsubscribe') and message.get('data') == 0:
                         break
-            except Exception as e:
-                if isinstance(e, redis_exceptions.ConnectionError) and not self._should_listen:
-                    return  # NOTE: on quick instantiation cycles, unsubscribe event might not be received before the connection is closed
+            except asyncio.CancelledError:
+                pass  # Expected during close()
+            except Exception:
                 log.exception(f'Unexpected error in Redis listener for {self.key}')
+            finally:
+                if self.pubsub.subscribed:
+                    await self.pubsub.unsubscribe()
+
+        def create_listener_task() -> None:
+            self._listener_task = background_tasks.create(listen(), name=f'redis-listen-{self.key}')
 
         if core.loop and core.loop.is_running():
-            background_tasks.create(listen(), name=f'redis-listen-{self.key}')
+            create_listener_task()
         else:
-            core.app.on_startup(listen())
+            core.app.on_startup(create_listener_task)
 
     def publish(self) -> None:
         """Publish the data to Redis and notify other instances."""
@@ -91,11 +95,12 @@ async def backup() -> None:
 
     async def close(self) -> None:
         """Close Redis connection and subscription."""
-        self._should_listen = False
-        if self.pubsub.subscribed:
-            await self.pubsub.unsubscribe()
-        await self.pubsub.close()
-        await self.redis_client.close()
+        if self._listener_task and not self._listener_task.done():
+            self._listener_task.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await self._listener_task
+        await self.pubsub.aclose()
+        await self.redis_client.aclose()
 
     def clear(self) -> None:
         super().clear()

@@ -164,6 +164,11 @@ def copy_tab(self, old_tab_id: str, tab_id: str) -> None:
                 self._tabs[tab_id] = ObservableDict()
             self._tabs[tab_id].update(self._tabs[old_tab_id])
 
+    async def close_tab(self, tab_id: Optional[str]) -> None:
+        """Close the tab storage. (For internal use only.)"""
+        if tab_id and isinstance(tab := self._tabs.get(tab_id), PersistentDict):
+            await tab.close()
+
     def clear(self) -> None:
         """Clears all storage."""
         self._general.clear()