diff --git a/include/PgSQLFFTO.hpp b/include/PgSQLFFTO.hpp index 73e3ab4201..8b3f5dbf3b 100644 --- a/include/PgSQLFFTO.hpp +++ b/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; diff --git a/lib/PgSQLFFTO.cpp b/lib/PgSQLFFTO.cpp index a042fbd4f3..e00603d48c 100644 --- a/lib/PgSQLFFTO.cpp +++ b/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; diff --git a/test/tap/tests/test_ffto_pgsql_pipeline-t.cpp b/test/tap/tests/test_ffto_pgsql_pipeline-t.cpp index 37f19c4c01..e898a0db43 100644 --- a/test/tap/tests/test_ffto_pgsql_pipeline-t.cpp +++ b/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);