Process query cycle messages synchronously to prevent double-delivery

.release-notes/fix-process-again-double-delivery.md

@@ -0,0 +1,3 @@
+## Fix double-delivery of pg_query_failed when close() races with error processing
+
+When `close()` was called while the session was between processing an error response and processing the subsequent ready-for-query message, the `ResultReceiver` could receive `pg_query_failed` twice — once with the original error and again with `SessionClosed`. Query cycle messages are now processed synchronously, preventing other operations from interleaving.

postgres/_response_message_parser.pony

@@ -1,47 +1,60 @@
 use "buffered"
 
 primitive _ResponseMessageParser
+  """
+  Processes buffered messages synchronously within a query cycle, yielding
+  only after ReadyForQuery. This prevents other behaviors (like `close()`)
+  from interleaving between result delivery and query dequeuing, which would
+  cause double-delivery of `pg_query_failed`.
+
+  If a callback triggers shutdown, `on_shutdown` clears the read buffer,
+  causing the next parse to return `None` and exit the loop naturally.
+  """
   fun apply(s: Session ref, readbuf: Reader) =>
-    try
-      match _ResponseParser(readbuf)?
-      | let msg: _AuthenticationMD5PasswordMessage =>
-        s.state.on_authentication_md5_password(s, msg)
-      | _AuthenticationOkMessage =>
-        s.state.on_authentication_ok(s)
-      | let msg: _CommandCompleteMessage =>
-        s.state.on_command_complete(s, msg)
-      | let msg: _DataRowMessage =>
-        s.state.on_data_row(s, msg)
-      | let err: ErrorResponseMessage =>
-        match err.code
-        | "28000" =>
-          s.state.on_authentication_failed(s,
-            InvalidAuthenticationSpecification)
+    while true do
+      try
+        match _ResponseParser(readbuf)?
+        | let msg: _AuthenticationMD5PasswordMessage =>
+          s.state.on_authentication_md5_password(s, msg)
+        | _AuthenticationOkMessage =>
+          s.state.on_authentication_ok(s)
+        | let msg: _CommandCompleteMessage =>
+          s.state.on_command_complete(s, msg)
+        | let msg: _DataRowMessage =>
+          s.state.on_data_row(s, msg)
+        | let err: ErrorResponseMessage =>
+          match err.code
+          | "28000" =>
+            s.state.on_authentication_failed(s,
+              InvalidAuthenticationSpecification)
+            return
+          | "28P01" =>
+            s.state.on_authentication_failed(s, InvalidPassword)
+            return
+          else
+            s.state.on_error_response(s, err)
+          end
+        | let msg: _ReadyForQueryMessage =>
+          s.state.on_ready_for_query(s, msg)
+          // ReadyForQuery marks the end of a query cycle. Yield to other
+          // actors before processing the next cycle.
+          s._process_again()
           return
-        | "28P01" =>
-          s.state.on_authentication_failed(s, InvalidPassword)
+        | let msg: _RowDescriptionMessage =>
+          s.state.on_row_description(s, msg)
+        | let msg: _EmptyQueryResponseMessage =>
+          s.state.on_empty_query_response(s)
+        | None =>
+          // No complete message was found. Stop parsing for now.
           return
-        else
-          s.state.on_error_response(s, err)
         end
-      | let msg: _ReadyForQueryMessage =>
-        s.state.on_ready_for_query(s, msg)
-      | let msg: _RowDescriptionMessage =>
-        s.state.on_row_description(s, msg)
-      | let msg: _EmptyQueryResponseMessage =>
-        s.state.on_empty_query_response(s)
-      | None =>
-        // No complete message was found. Stop parsing for now.
+      else
+        // An unrecoverable error was encountered while parsing. Once that
+        // happens, there's no way we are going to be able to figure out how
+        // to get the responses back into an understandable state. The only
+        // thing we can do is shut the session down.
+
+        s.state.shutdown(s)
         return
       end
-    else
-      // An unrecoverable error was encountered while parsing. Once that
-      // happens, there's no way we are going to be able to figure out how
-      // to get the responses back into an understandable state. The only
-      // thing we can do is shut the session down.
-
-      s.state.shutdown(s)
-      return
     end
-
-    s._process_again()

postgres/session.pony

@@ -111,6 +111,8 @@ class ref _SessionConnected is _AuthenticableState
     r.pg_query_failed(q, SessionNotAuthenticated)
 
   fun ref on_shutdown(s: Session ref) =>
+    // Clearing the readbuf is required for _ResponseMessageParser's
+    // synchronous loop to exit — the next parse returns None.
     _readbuf.clear()
 
   fun user(): String =>
@@ -171,6 +173,8 @@ class _SessionLoggedIn is _AuthenticatedState
     query_state.try_run_query(s, this)
 
   fun ref on_shutdown(s: Session ref) =>
+    // Clearing the readbuf is required for _ResponseMessageParser's
+    // synchronous loop to exit — the next parse returns None.
     _readbuf.clear()
     for queue_item in query_queue.values() do
       (let query, let receiver) = queue_item