Worker seems to be not closing the LISTEN/NOTIFY connection on stop

[mikz]

Hey,

I'm starting some workers in tests to e2e validate the system and then I have trouble tearing down the database. It seems the Worker.stop does not stop the connection created in _maintain_notifications is not stored:

chancy/chancy/worker.py

Lines 445 to 470 in b0e3e94

	async def _maintain_notifications(self):
	"""
	Listen for notifications from the database.
	Improves the reactivity of a worker by allowing it to almost immediately
	react to cluster events using Postgres's LISTEN/NOTIFY feature.
	.. note::
	This feature utilizes a permanent connection to the database
	separate from the shared connection pool and is not counted against
	the pool's connection limit.
	"""
	connection = await AsyncConnection.connect(
	self.chancy.dsn, autocommit=True
	)
	await 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():
	j = json.loads(notification.payload)
	await self.hub.emit(j.pop("t"), j)

so later in stop it is not closed:

chancy/chancy/worker.py

Lines 805 to 849 in b0e3e94

	async def stop(self) -> bool:
	"""
	Stop the worker.
	Attempts to stop the worker gracefully, sending a CancelledError to all
	running tasks and waiting up to `shutdown_timeout` seconds for them to
	complete before returning.
	Returns True if the worker was stopped cleanly, or False if the worker
	returned due to the timeout expiring.
	"""
	try:
	async with asyncio.timeout(self.shutdown_timeout) as cm:
	# Stop accepting new queues and queue changes.
	try:
	await self.manager.cancel("queues")
	except KeyError:
	pass
	# Delete all the queues we know about so the executors can
	# clean up.
	self._queues.clear()
	while self._executors:
	await asyncio.sleep(0.1)
	# And finally axe everything else started by this worker.
	await self.manager.cancel_all()
	except TimeoutError:
	# We check this instead of depending on the exception in case the
	# exception wasn't really raised by us but a nested timeout.
	if cm.expired():
	await self.hub.emit(
	"worker.shutdown_timeout",
	{
	"worker": self,
	},
	)
	return False
	raise
	await self.hub.emit(
	"worker.stopped",
	{
	"worker": self,
	},
	)
	return True

I'd like to confirm it is not intentional before I make a PR to fix it.

[TkTech]

Definite oversight. I'll need to think about how to structure a connection leakage test so we can catch this kind of issue in the future.

[mikz]

in Ruby on Rails world all the libraries I've used are going through shared connection pool. Here if they'd all be taken from a pool (even if just a local to the worker), then it would be easier to audit.
I believe it is ok to take a long running connection from the pool and never audit it - as the pool ensures the library does not come over a limit - and ideally a shared pool like #70 would make it possible to ensure the app does not go over a certain connection limit and it can be monitored.

But yeah, having a smoke test to ensure this happens would be cool. There is a way on linux to use procfs and count opened connections, but maybe it would be better to get the connections from postgres itself (minus the one that makes the query to the stats, which should be easy to identify).