diff --git a/app/api/channel.py b/app/api/channel.py index 269e0ce..4c45276 100644 --- a/app/api/channel.py +++ b/app/api/channel.py @@ -66,6 +66,12 @@ class BaseChannelNotifier: ) ) + async def aclose(self) -> None: + """Release the Redis connection held by this notifier.""" + if self.redis is not None: + await self.redis.aclose() + self.redis = None + class BaseChannelListener: def __init__(self, channel: str, redis_url=None): @@ -125,7 +131,6 @@ class BaseChannelListener: except AttributeError as e: logger.error(f"AttributeError while handling channel message: {e}") except Exception as e: - print(type(e)) logger.error( f"Error handling channel message: {e}. Error type: {type(e)}" ) @@ -134,7 +139,10 @@ class BaseChannelListener: except Exception as e: logger.error(f"Error listening to channel: {e}") finally: - await pubsub.unsubscribe() + # release the pubsub connection and the listener's Redis client; + # aclose() unsubscribes and returns the connection to the pool + await pubsub.aclose() + await self.aclose() logger.info(f"Stopped listening on channel: {self.channel}") return Ok(None) @@ -146,6 +154,12 @@ class BaseChannelListener: return Ok(None) + async def aclose(self) -> None: + """Release the Redis connection held by this listener.""" + if self.redis is not None: + await self.redis.aclose() + self.redis = None + async def handle_event(self, event): """Process incoming events - to be implemented by subclasses""" logger.debug(f"Received event: {event}") diff --git a/app/apps/cache.py b/app/apps/cache.py index 6781de8..7ae4f68 100644 --- a/app/apps/cache.py +++ b/app/apps/cache.py @@ -4,6 +4,7 @@ This module contains the caching implementation for app status data. It provides functions to retrieve, store, and monitor app status data in Redis. """ +import asyncio from typing import Optional from loguru import logger @@ -197,10 +198,20 @@ async def watch_app_status_changes(): try: while True: listener = AppStatusUpdateListener() - await listener.connect() - # This will loop forever + match await listener.connect(): + case Err(report): + logger.error( + f"App status listener failed to connect: {report.format()}" + ) + await listener.aclose() + # back off before retrying so a Redis outage doesn't + # spin this loop + await asyncio.sleep(5) + continue + # This will loop forever (listen() closes its own resources) await listener.listen() logger.error("Recreating app status channel listener, as it stopped!") + await asyncio.sleep(1) except Exception as e: return Err(Report(f"App status channel listener error: {e}", error=e)) diff --git a/app/apps/tasks_impl/app_manage.py b/app/apps/tasks_impl/app_manage.py index d27134d..d77d513 100644 --- a/app/apps/tasks_impl/app_manage.py +++ b/app/apps/tasks_impl/app_manage.py @@ -219,8 +219,9 @@ async def app_manage_task_impl( await _send_finish_message(id, mode, notifier) + await notifier.aclose() if redis_client: - await redis_client.close() + await redis_client.aclose() async def _send_finish_message( diff --git a/app/apps/tasks_impl/app_status_update.py b/app/apps/tasks_impl/app_status_update.py index 1d03cd9..1fe5bb2 100644 --- a/app/apps/tasks_impl/app_status_update.py +++ b/app/apps/tasks_impl/app_status_update.py @@ -222,8 +222,9 @@ async def update_app_state_task_impl( case Err(message): _log_notify_listeners_error(message) + await channel_notifier.aclose() if redis_client: - await redis_client.close() + await redis_client.aclose() def _log_notify_listeners_error(message: Report): diff --git a/tests/test_app_manage_lock.py b/tests/test_app_manage_lock.py index f1e5bad..409632f 100644 --- a/tests/test_app_manage_lock.py +++ b/tests/test_app_manage_lock.py @@ -30,6 +30,9 @@ async def test_lock_already_held_does_not_release_or_finish(monkeypatch): sent.append(contents) return Ok(None) + async def aclose(self): + pass + async def fake_acquire(key, lock_ttl, redis): return Ok(False) # lock already held by the running install @@ -74,6 +77,9 @@ async def test_acquired_lock_is_released_and_finished(monkeypatch): sent.append(contents) return Ok(None) + async def aclose(self): + pass + async def fake_acquire(key, lock_ttl, redis): return Ok(True) # lock acquired by this task diff --git a/tests/test_channel_cleanup.py b/tests/test_channel_cleanup.py new file mode 100644 index 0000000..b1ea9f7 --- /dev/null +++ b/tests/test_channel_cleanup.py @@ -0,0 +1,54 @@ +""" +Regression test: the channel listener must release its Redis resources. + +listen() only called pubsub.unsubscribe() in its finally block; it never +closed the pubsub connection or the listener's own Redis client, leaking one +connection per install/uninstall and per app-status listener recreation. +""" + +from app.api.channel import BaseChannelListener + + +class _FakePubSub: + def __init__(self, on_poll): + self._on_poll = on_poll + self.aclosed = False + + async def subscribe(self, channel): + pass + + async def get_message(self, **kwargs): + self._on_poll() + return None + + async def unsubscribe(self): + pass + + async def aclose(self): + self.aclosed = True + + +class _FakeRedis: + def __init__(self, pubsub): + self._pubsub = pubsub + self.aclosed = False + + def pubsub(self): + return self._pubsub + + async def aclose(self): + self.aclosed = True + + +async def test_listen_closes_pubsub_and_redis(): + listener = BaseChannelListener("test-channel") + + # stop the listen loop after the first poll + pubsub = _FakePubSub(on_poll=lambda: setattr(listener, "running", False)) + redis = _FakeRedis(pubsub) + listener.redis = redis + + await listener.listen() + + assert pubsub.aclosed is True, "pubsub connection must be closed" + assert redis.aclosed is True, "listener's redis client must be closed"