From 81da03818a164afe645626f55159e1bfbe6649d5 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 23 Jan 2026 12:05:39 +0000 Subject: [PATCH 1/3] Initial plan From 23bee0f59a48bfaebc9ef57418ba3e96eeac2672 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 23 Jan 2026 12:09:14 +0000 Subject: [PATCH 2/3] Fix LISTEN/NOTIFY connection leak and add test Co-authored-by: TkTech <72590+TkTech@users.noreply.github.com> --- chancy/worker.py | 12 ++++++++--- tests/test_worker.py | 48 ++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 57 insertions(+), 3 deletions(-) diff --git a/chancy/worker.py b/chancy/worker.py index 5634250..50500f7 100644 --- a/chancy/worker.py +++ b/chancy/worker.py @@ -186,6 +186,8 @@ def __init__( self._queues: dict[str, Queue] = {} # The executors that the worker is currently using. self._executors: dict[str, Executor] = {} + # The connection used for LISTEN/NOTIFY, if notifications are enabled. + self._notification_connection: AsyncConnection | None = None async def start(self): """ @@ -455,17 +457,17 @@ async def _maintain_notifications(self): separate from the shared connection pool and is not counted against the pool's connection limit. """ - connection = await AsyncConnection.connect( + self._notification_connection = await AsyncConnection.connect( self.chancy.dsn, autocommit=True ) - await connection.execute( + await self._notification_connection.execute( sql.SQL("LISTEN {channel};").format( channel=sql.Identifier(f"{self.chancy.prefix}events") ) ) self.chancy.log.info("Started listening for realtime notifications.") self._notifications_ready_event.set() - async for notification in connection.notifies(): + async for notification in self._notification_connection.notifies(): j = json.loads(notification.payload) await self.hub.emit(j.pop("t"), j) @@ -825,6 +827,10 @@ async def stop(self) -> bool: self._queues.clear() while self._executors: await asyncio.sleep(0.1) + # Close the notification connection if it exists + if self._notification_connection: + await self._notification_connection.close() + self._notification_connection = None # And finally axe everything else started by this worker. await self.manager.cancel_all() except TimeoutError: diff --git a/tests/test_worker.py b/tests/test_worker.py index 484c562..4d08279 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -130,3 +130,51 @@ async def test_immediate_processing(chancy: Chancy, worker: Worker): ) assert result.state == QueuedJob.State.SUCCEEDED + + +@pytest.mark.asyncio +async def test_notification_connection_cleanup(chancy: Chancy): + """ + Test that the LISTEN/NOTIFY connection is properly closed when the worker + stops, preventing connection leaks. + """ + # Get initial connection count for this database and application name + async with chancy.pool.connection() as conn: + result = await conn.execute( + """ + SELECT COUNT(*) as count + FROM pg_stat_activity + WHERE datname = current_database() + AND application_name LIKE 'psycopg%' + """ + ) + initial_count = (await result.fetchone())[0] + + # Start and stop a worker + async with Worker(chancy) as worker: + # Wait for the worker to fully initialize + await asyncio.sleep(1) + + # Verify the notification connection exists + assert worker._notification_connection is not None + assert not worker._notification_connection.closed + + # After the worker stops, verify the connection is closed + async with chancy.pool.connection() as conn: + result = await conn.execute( + """ + SELECT COUNT(*) as count + FROM pg_stat_activity + WHERE datname = current_database() + AND application_name LIKE 'psycopg%' + """ + ) + final_count = (await result.fetchone())[0] + + # The connection count should be back to the initial count (or less) + # We allow for <= to account for pool connections that may have been + # released + assert final_count <= initial_count + 1, ( + f"Connection leak detected: initial={initial_count}, " + f"final={final_count}" + ) From 5a2a5be75a23d661ef60c97ac93f7d968caae6d9 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Fri, 23 Jan 2026 12:17:44 +0000 Subject: [PATCH 3/3] Fix test to not rely on chancy fixture pool Co-authored-by: TkTech <72590+TkTech@users.noreply.github.com> --- tests/test_worker.py | 45 ++++++++++++++++++++++++++++---------------- 1 file changed, 29 insertions(+), 16 deletions(-) diff --git a/tests/test_worker.py b/tests/test_worker.py index 4d08279..86b6fd4 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -133,13 +133,20 @@ async def test_immediate_processing(chancy: Chancy, worker: Worker): @pytest.mark.asyncio -async def test_notification_connection_cleanup(chancy: Chancy): +async def test_notification_connection_cleanup(): """ Test that the LISTEN/NOTIFY connection is properly closed when the worker stops, preventing connection leaks. + + This test doesn't use the chancy fixture to ensure we have full control + over connection lifecycle and can properly detect leaks. """ - # Get initial connection count for this database and application name - async with chancy.pool.connection() as conn: + from psycopg import AsyncConnection + + dsn = "postgresql://postgres:localtest@localhost:8190/postgres" + + # Get initial connection count using a separate connection + async with await AsyncConnection.connect(dsn) as conn: result = await conn.execute( """ SELECT COUNT(*) as count @@ -150,17 +157,23 @@ async def test_notification_connection_cleanup(chancy: Chancy): ) initial_count = (await result.fetchone())[0] - # Start and stop a worker - async with Worker(chancy) as worker: - # Wait for the worker to fully initialize - await asyncio.sleep(1) - - # Verify the notification connection exists - assert worker._notification_connection is not None - assert not worker._notification_connection.closed - - # After the worker stops, verify the connection is closed - async with chancy.pool.connection() as conn: + # Create, start, and stop a worker with its own Chancy instance + async with Chancy(dsn) as chancy: + await chancy.migrate() + async with Worker(chancy) as worker: + # Wait for the worker to fully initialize + await asyncio.sleep(1) + + # Verify the notification connection exists + assert worker._notification_connection is not None + assert not worker._notification_connection.closed + + # Give a moment for connections to fully close + await asyncio.sleep(0.5) + + # After everything is stopped, verify no connection leak + # Use a completely separate connection + async with await AsyncConnection.connect(dsn) as conn: result = await conn.execute( """ SELECT COUNT(*) as count @@ -172,8 +185,8 @@ async def test_notification_connection_cleanup(chancy: Chancy): final_count = (await result.fetchone())[0] # The connection count should be back to the initial count (or less) - # We allow for <= to account for pool connections that may have been - # released + # We allow for <= to account for the single connection we just created + # to check the count assert final_count <= initial_count + 1, ( f"Connection leak detected: initial={initial_count}, " f"final={final_count}"