Support PgBouncer by sending only a single SYNC message per query

This is a simple implementation of PgBouncer support for asyncpg. It
doesn't have any detection features, but at least changes asyncpg
behavior in such a way that using PgBouncer is possible.
This commit gets rid of the explicit SYNC after a
parse/describe sequence and changes is to a FLUSH. This should work regardless of
the setting of statement_cache_size and whether or not it's pgbouncer or
a direct postgres connection.
With this, PgBouncer is supported when setting statement_cache_size
explicitly to 0.

Author: fvannee

asyncpg/exceptions/_base.py

@@ -142,7 +142,8 @@ def _make_constructor(cls, fields, query=None):
                   purpose;
 
                 * if you have no option of avoiding the use of pgbouncer,
-                  then you must switch pgbouncer's pool_mode to "session".
+                  then you can set statement_cache_size to 0 when creating
+                  the asyncpg connection object.
             """)
 
             dct['hint'] = hint

asyncpg/protocol/coreproto.pyx

@@ -187,19 +187,23 @@ cdef class CoreProtocol:
         elif mtype == b'T':
             # Row description
             self.result_row_desc = self.buffer.consume_message()
+            self._push_result()
 
         elif mtype == b'E':
             # ErrorResponse
             self._parse_msg_error_response(True)
-
-        elif mtype == b'Z':
-            # ReadyForQuery
-            self._parse_msg_ready_for_query()
-            self._push_result()
+            # we don't send a sync during the parse/describe sequence
+            # but send a FLUSH instead. If an error happens we need to
+            # send a SYNC explicitly in order to mark the end of the transaction.
+            # this effectively clears the error and we then wait until we get a
+            # ready for new query message
+            self._write(SYNC_MESSAGE)
+            self._set_state(PROTOCOL_CANCELLED)
 
         elif mtype == b'n':
             # NoData
             self.buffer.discard_message()
+            self._push_result()
 
     cdef _process__bind_execute(self, char mtype):
         if mtype == b'D':
@@ -853,7 +857,7 @@ cdef class CoreProtocol:
         buf.end_message()
         packet.write_buffer(buf)
 
-        packet.write_bytes(SYNC_MESSAGE)
+        packet.write_bytes(FLUSH_MESSAGE)
 
         self._write(packet)
 
@@ -1028,3 +1032,4 @@ cdef class CoreProtocol:
 
 
 cdef bytes SYNC_MESSAGE = bytes(WriteBuffer.new_message(b'S').end_message())
+cdef bytes FLUSH_MESSAGE = bytes(WriteBuffer.new_message(b'H').end_message())

asyncpg/protocol/protocol.pyx

@@ -588,6 +588,8 @@ cdef class BaseProtocol(CoreProtocol):
             })
             self.abort()
 
+        if self.state == PROTOCOL_PREPARE:
+            self._write(SYNC_MESSAGE)
         self._set_state(PROTOCOL_CANCELLED)
 
     def _on_timeout(self, fut):