Warn when there are active listeners on a connection that is released

Closes: MagicStack#190.

Author: elprans

asyncpg/connection.py

@@ -8,6 +8,7 @@
 import asyncio
 import collections
 import collections.abc
+import itertools
 import struct
 import time
 import warnings
@@ -1172,6 +1173,29 @@ def _set_proxy(self, proxy):
 
         self._proxy = proxy
 
+    def _check_listeners(self, listeners, listener_type):
+        if listeners:
+            count = len(listeners)
+
+            w = exceptions.InterfaceWarning(
+                '{conn!r} is being released to the pool but has {c} active '
+                '{type} listener{s}'.format(
+                    conn=self, c=count, type=listener_type,
+                    s='s' if count > 1 else ''))
+
+            warnings.warn(w)
+
+    def _on_release(self, stacklevel=1):
+        # Invalidate external references to the connection.
+        self._pool_release_ctr += 1
+        # Called when the connection is about to be released to the pool.
+        # Let's check that the user has not left any listeners on it.
+        self._check_listeners(
+            list(itertools.chain.from_iterable(self._listeners.values())),
+            'notification')
+        self._check_listeners(
+            self._log_listeners, 'log')
+
     def _drop_local_statement_cache(self):
         self._stmt_cache.clear()
 

asyncpg/connresource.py

@@ -27,10 +27,10 @@ class ConnectionResource:
 
     def __init__(self, connection):
         self._connection = connection
-        self._con_release_ctr = getattr(connection, '_pool_release_ctr', None)
+        self._con_release_ctr = connection._pool_release_ctr
 
     def _check_conn_validity(self, meth_name):
-        con_release_ctr = getattr(self._connection, '_pool_release_ctr', None)
+        con_release_ctr = self._connection._pool_release_ctr
         if con_release_ctr != self._con_release_ctr:
             raise exceptions.InterfaceError(
                 'cannot call {}.{}(): '

asyncpg/exceptions/_base.py

@@ -10,7 +10,7 @@
 
 
 __all__ = ('PostgresError', 'FatalPostgresError', 'UnknownPostgresError',
-           'InterfaceError', 'PostgresLogMessage')
+           'InterfaceError', 'InterfaceWarning', 'PostgresLogMessage')
 
 
 def _is_asyncpg_class(cls):
@@ -159,9 +159,36 @@ class UnknownPostgresError(FatalPostgresError):
     """An error with an unknown SQLSTATE code."""
 
 
-class InterfaceError(Exception):
+class InterfaceMessage:
+    def __init__(self, *, detail=None, hint=None):
+        self.detail = detail
+        self.hint = hint
+
+    def __str__(self):
+        msg = self.args[0]
+        if self.detail:
+            msg += '\nDETAIL:  {}'.format(self.detail)
+        if self.hint:
+            msg += '\nHINT:  {}'.format(self.hint)
+
+        return msg
+
+
+class InterfaceError(InterfaceMessage, Exception):
     """An error caused by improper use of asyncpg API."""
 
+    def __init__(self, msg, *, detail=None, hint=None):
+        InterfaceMessage.__init__(self, detail=detail, hint=hint)
+        Exception.__init__(self, msg)
+
+
+class InterfaceWarning(InterfaceMessage, UserWarning):
+    """A warning caused by an improper use of asyncpg API."""
+
+    def __init__(self, msg, *, detail=None, hint=None):
+        InterfaceMessage.__init__(self, detail=detail, hint=hint)
+        Warning.__init__(self, msg)
+
 
 class PostgresLogMessage(PostgresMessage):
     """A base class for non-error server messages."""

asyncpg/pool.py

@@ -175,8 +175,7 @@ async def release(self):
         assert self._in_use
         self._in_use = False
 
-        # Invalidate external references to the connection.
-        self._con._pool_release_ctr += 1
+        self._con._on_release()
 
         if self._con.is_closed():
             self._con = None

tests/test_listeners.py

@@ -76,6 +76,20 @@ def listener1(*args):
                     await q1.get(),
                     (con1, con2.get_server_pid(), 'ipc', 'hello'))
 
+                await con1.remove_listener('ipc', listener1)
+
+    async def test_dangling_listener_warns(self):
+        async with self.create_pool(database='postgres') as pool:
+            with self.assertWarnsRegex(
+                    exceptions.InterfaceWarning,
+                    '.*Connection.*is being released to the pool but '
+                    'has 1 active notification listener'):
+                async with pool.acquire() as con:
+                    def listener1(*args):
+                        pass
+
+                    await con.add_listener('ipc', listener1)
+
 
 class TestLogListeners(tb.ConnectedTestCase):
 
@@ -231,3 +245,15 @@ def notice_callb(con, message):
         with self.assertRaises(exceptions.InvalidCharacterValueForCastError):
             await raise_message('', '22018')
         self.assertTrue(q1.empty())
+
+    async def test_dangling_log_listener_warns(self):
+        async with self.create_pool(database='postgres') as pool:
+            with self.assertWarnsRegex(
+                    exceptions.InterfaceWarning,
+                    '.*Connection.*is being released to the pool but '
+                    'has 1 active log listener'):
+                async with pool.acquire() as con:
+                    def listener1(*args):
+                        pass
+
+                    con.add_log_listener(listener1)