fix(pgsql-ffto): stop ReadyForQuery finalizing pipelined extended queries (test_ffto_pgsql_pipeline-t)

include/PgSQLFFTO.hpp

@@ -77,6 +77,14 @@ class PgSQLFFTO : public TrafficObserver {
     uint64_t m_affected_rows {0};          ///< Accumulated affected rows for the current query.
     uint64_t m_rows_sent {0};              ///< Accumulated rows sent for the current query.
     bool m_current_finalize_on_sync {false}; ///< Whether current query finalizes on ReadyForQuery ('Z').
+    /// Whether the current query has received its own response terminator
+    /// (CommandComplete, EmptyQueryResponse or PortalSuspended). This, not
+    /// m_current_finalize_on_sync, is what qualifies a ReadyForQuery to
+    /// finalize: a ReadyForQuery left over from an earlier exchange can arrive
+    /// while a freshly activated query is still awaiting its first response,
+    /// and finalizing there would report zeroed counters and drop the real
+    /// response that follows.
+    bool m_response_seen {false};
 
     struct PendingQuery {
         std::string query;

lib/PgSQLFFTO.cpp

@@ -160,6 +160,7 @@ void PgSQLFFTO::track_query(std::string query, bool finalize_on_sync) {
         m_current_finalize_on_sync = pending.finalize_on_sync;
         m_affected_rows = 0;
         m_rows_sent = 0;
+        m_response_seen = false;
         m_state = AWAITING_RESPONSE;
         return;
     }
@@ -173,6 +174,7 @@ void PgSQLFFTO::clear_current_query() {
     m_affected_rows = 0;
     m_rows_sent = 0;
     m_current_finalize_on_sync = false;
+    m_response_seen = false;
 }
 
 void PgSQLFFTO::activate_next_query() {
@@ -189,6 +191,7 @@ void PgSQLFFTO::activate_next_query() {
     m_current_finalize_on_sync = next_query.finalize_on_sync;
     m_affected_rows = 0;
     m_rows_sent = 0;
+    m_response_seen = false;
     m_state = AWAITING_RESPONSE;
 }
 
@@ -258,11 +261,56 @@ void PgSQLFFTO::process_server_message(char type, const unsigned char* payload,
         uint64_t rows = extract_pg_rows_affected(payload, len, is_select);
         if (is_select) m_rows_sent += rows;
         else m_affected_rows += rows;
+        m_response_seen = true;
+        if (!m_current_finalize_on_sync) {
+            finalize_current_query();
+        }
+    } else if (type == 'I' || type == 's') {
+        // EmptyQueryResponse ('I') replaces CommandComplete when the query text
+        // is empty once comments/whitespace are stripped, and PortalSuspended
+        // ('s') replaces it when a row-limited Execute stops early. Both
+        // terminate the current query's response just as CommandComplete does,
+        // with no row counts to add. Without this, an extended Execute answered
+        // by either one would never be finalized -- neither branch below fires
+        // for it -- and the stalled query would block the pending queue and
+        // swallow the NEXT query's CommandComplete.
+        //
+        // ProxySQL does not emit PortalSuspended today (Execute's max-rows
+        // field is parsed but never acted on, issue #5900), so 's' is inert
+        // until that is fixed; handling it now costs two lines and avoids
+        // reintroducing the stall when it is.
+        m_response_seen = true;
         if (!m_current_finalize_on_sync) {
             finalize_current_query();
         }
     } else if (type == 'Z') {
-        finalize_current_query();
+        // ReadyForQuery terminates an exchange, and is the finalizer for SIMPLE
+        // queries -- their CommandComplete deliberately does not finalize (see
+        // the m_current_finalize_on_sync check above).
+        //
+        // The qualifying condition is m_response_seen, NOT
+        // m_current_finalize_on_sync. A ReadyForQuery belongs to the exchange
+        // that produced it, but the query that happens to be current when it
+        // arrives may already belong to the NEXT one: a client that pipelines
+        // without reading its replies leaves an earlier exchange's
+        // ReadyForQuery in flight, and an extended CommandComplete can activate
+        // the following queued query before it lands. Finalizing then reports
+        // that query with zeroed counters and pops the queue, so its real
+        // response is attributed to whatever comes after and the last response
+        // in the batch is dropped once the queue drains to IDLE -- silent,
+        // plausible-looking corruption of stats_pgsql_query_digest rather than
+        // an obvious failure.
+        //
+        // Gating on "has this query actually seen its own response terminator"
+        // is correct for every ordering: a simple query that got its
+        // CommandComplete finalizes here as before; an extended Execute has
+        // already finalized itself above and cannot still be current; and a
+        // query still awaiting its first response is left alone for the
+        // CommandComplete that is genuinely its own. An abandoned batch is
+        // still cleared by ErrorResponse ('E' below).
+        if (m_response_seen) {
+            finalize_current_query();
+        }
     } else if (type == 'E') {
         if (!m_current_query.empty() && m_query_start_time != 0) {
             unsigned long long duration = monotonic_time() - m_query_start_time;

test/tap/tests/test_ffto_pgsql_pipeline-t.cpp

@@ -14,6 +14,8 @@
  * @par Test scenarios
  *  1. 3 different queries pipelined before Sync → 3 separate digests
  *  2. Same prepared statement executed 10 times in pipeline → count_star=10
+ *  3. Extended Execute + Sync + simple Query pipelined together → each digest
+ *     keeps its own row counts across the exchange boundary
  *
  * @pre  ProxySQL running with a PostgreSQL backend.
  *
@@ -38,12 +40,13 @@
  * @brief Total number of planned TAP assertions.
  *
  * Breakdown:
- *  - Setup:                 1 (connect)
+ *  - Setup:                  1 (connect)
  *  - Scenario 1 (3 queries): 3 x 3 = 9 (3 verify_pg_digest calls)
  *  - Scenario 2 (10x exec):  1 x 3 = 3 (1 verify_pg_digest call)
- *  Total = 13
+ *  - Scenario 3 (boundary):  2 x 3 = 6 (2 verify_pg_digest calls)
+ *  Total = 19
  */
-static constexpr int kPlannedTests = 13;
+static constexpr int kPlannedTests = 19;
 
 #define FAIL_AND_SKIP_REMAINING(cleanup_label, fmt, ...) \
     do { \
@@ -149,10 +152,26 @@ int main(int argc, char** argv) {
     }
     ok(pgc != NULL && pgc->isConnected(), "Connected via pg_lite_client");
 
-    /* Create test table via simple query */
+    /* Create test table via simple query.
+     *
+     * Each response MUST be consumed. execute() only writes the Query message;
+     * it does not read the reply, so firing three of them back-to-back leaves
+     * three unread ReadyForQuery messages in the socket. The next
+     * consumeInputUntilReady() -- the one after the Parse batch below -- stops
+     * at the FIRST ReadyForQuery it sees, which would be DROP TABLE's, not the
+     * one it is waiting for. From that point the client is a full exchange
+     * behind the server and keeps pipelining anyway, so an old ReadyForQuery is
+     * still in flight when the Bind/Execute batch starts. Whether that stray
+     * message reaches the proxy before or after the batch's first Execute is
+     * pure timing, which is exactly how this test failed intermittently in CI
+     * (SELECT/INSERT/UPDATE stats each shifted by one position) while passing
+     * locally. */
     pgc->execute("DROP TABLE IF EXISTS ffto_pg_pipe");
+    pgc->consumeInputUntilReady();
     pgc->execute("CREATE TABLE ffto_pg_pipe (id INT PRIMARY KEY, val TEXT)");
+    pgc->consumeInputUntilReady();
     pgc->execute("INSERT INTO ffto_pg_pipe VALUES (1,'a'), (2,'b'), (3,'c')");
+    pgc->consumeInputUntilReady();
 
     /* ================================================================
      * Scenario 1:  3 different queries pipelined before Sync
@@ -225,6 +244,57 @@ int main(int argc, char** argv) {
 
     verify_pg_digest(admin, "SELECT val FROM ffto_pg_pipe WHERE id = $1", 10, 0, 10);
 
+    /* ================================================================
+     * Scenario 3:  extended Execute + Sync + SIMPLE query, pipelined
+     *
+     * Regression guard for stats attribution across an exchange boundary.
+     * The client sends Bind/Execute, Sync, and then a simple Query without
+     * reading anything in between, so two exchanges are in flight at once
+     * and the server answers:
+     *
+     *   CommandComplete(Execute), ReadyForQuery(Sync),
+     *   CommandComplete(Query),   ReadyForQuery(Query)
+     *
+     * The Execute's CommandComplete finalizes the Execute and activates the
+     * simple query, which is queued behind it. The ReadyForQuery that then
+     * arrives belongs to the *Sync*, i.e. to the exchange that just ended --
+     * not to the simple query now sitting in front of it. Finalizing on it
+     * reports the simple query with zero rows and pops the queue, so the
+     * CommandComplete that really is its own is discarded and its digest
+     * silently records rows_sent=0.
+     *
+     * Unlike scenarios 1 and 2 this ordering is deterministic, not a race:
+     * the response sequence above follows purely from what the client sends.
+     * ================================================================ */
+    diag("--- Scenario 3: extended Execute + Sync + simple query ---");
+    clear_pg_stats(admin);
+
+    try {
+        /* Exchange 1: extended Bind/Execute terminated by Sync. */
+        pgc->bindStatement("pipe_sel", "",
+            {{std::string("2"), 0}}, {}, false);
+        pgc->executePortal("", 0, false);
+        pgc->sendSync();
+
+        /* Exchange 2: a simple query, sent WITHOUT reading exchange 1's
+         * replies -- that overlap is the whole point of the scenario. */
+        pgc->execute("SELECT count(*) FROM ffto_pg_pipe");
+
+        /* Now drain both exchanges: one ReadyForQuery each. */
+        pgc->consumeInputUntilReady();
+        pgc->consumeInputUntilReady();
+    } catch (const PgException& e) {
+        diag("Pipeline scenario 3 failed: %s", e.what());
+        FAIL_AND_SKIP_REMAINING(cleanup, "Pipelined exchange-boundary test failed");
+    }
+
+    /* The extended Execute is unaffected -- it is finalized by its own
+     * CommandComplete before the boundary is crossed. */
+    verify_pg_digest(admin, "SELECT val FROM ffto_pg_pipe WHERE id = $1", 1, 0, 1);
+    /* The simple query is the one that gets zeroed when ReadyForQuery is
+     * allowed to finalize a query that has not yet seen its own response. */
+    verify_pg_digest(admin, "SELECT count(*) FROM ffto_pg_pipe", 1, 0, 1);
+
 cleanup:
     if (pgc) { delete pgc; }
     if (admin) mysql_close(admin);