pgbench: Fix assertion failure with multiple \syncpipeline in pipeline mode.

Previously, when pgbench ran a custom script that triggered retriable errors
(e.g., deadlocks) followed by multiple \syncpipeline commands in pipeline mode,
the following assertion failure could occur:

    Assertion failed: (res == ((void*)0)), function discardUntilSync, file pgbench.c, line 3594.

The issue was that discardUntilSync() assumed a pipeline sync result
(PGRES_PIPELINE_SYNC) would always be followed by either another sync result
or NULL. This assumption was incorrect: when multiple sync requests were sent,
a sync result could instead be followed by another result type. In such cases,
discardUntilSync() mishandled the results, leading to the assertion failure.

This commit fixes the issue by making discardUntilSync() correctly handle cases
where a pipeline sync result is followed by other result types. It now continues
discarding results until another pipeline sync followed by NULL is reached.

Backpatched to v17, where support for \syncpipeline command in pgbench was
introduced.

Author: Yugo Nagata <nagata@sraoss.co.jp>
Reviewed-by: Chao Li <lic@highgo.com>
Reviewed-by: Fujii Masao <masao.fujii@gmail.com>
Discussion: https://postgr.es/m/20251111105037.f3fc554616bc19891f926c5b@sraoss.co.jp
Backpatch-through: 17

1 files changed, 28 insertions, 11 deletions

diff --git a/src/bin/pgbench/pgbench.c b/src/bin/pgbench/pgbench.c
index d8764ba6fe0..a425176ecdc 100644
--- a/src/bin/pgbench/pgbench.c
+++ b/src/bin/pgbench/pgbench.c
@@ -3563,14 +3563,18 @@ doRetry(CState *st, pg_time_usec_t *now)
 }
 
 /*
- * Read results and discard it until a sync point.
+ * Read and discard results until the last sync point.
  */
 static int
 discardUntilSync(CState *st)
 {
 	bool		received_sync = false;
 
-	/* send a sync */
+	/*
+	 * Send a Sync message to ensure at least one PGRES_PIPELINE_SYNC is
+	 * received and to avoid an infinite loop, since all earlier ones may have
+	 * already been received.
+	 */
 	if (!PQpipelineSync(st->con))
 	{
 		pg_log_error("client %d aborted: failed to send a pipeline sync",
@@ -3578,29 +3582,42 @@ discardUntilSync(CState *st)
 		return 0;
 	}
 
-	/* receive PGRES_PIPELINE_SYNC and null following it */
+	/*
+	 * Continue reading results until the last sync point, i.e., until
+	 * reaching null just after PGRES_PIPELINE_SYNC.
+	 */
 	for (;;)
 	{
 		PGresult   *res = PQgetResult(st->con);
 
+		if (PQstatus(st->con) == CONNECTION_BAD)
+		{
+			pg_log_error("client %d aborted while rolling back the transaction after an error; perhaps the backend died while processing",
+						 st->id);
+			PQclear(res);
+			return 0;
+		}
+
 		if (PQresultStatus(res) == PGRES_PIPELINE_SYNC)
 			received_sync = true;
-		else if (received_sync)
+		else if (received_sync && res == NULL)
 		{
 			/*
-			 * PGRES_PIPELINE_SYNC must be followed by another
-			 * PGRES_PIPELINE_SYNC or NULL; otherwise, assert failure.
-			 */
-			Assert(res == NULL);
-
-			/*
 			 * Reset ongoing sync count to 0 since all PGRES_PIPELINE_SYNC
 			 * results have been discarded.
 			 */
 			st->num_syncs = 0;
-			PQclear(res);
 			break;
 		}
+		else
+		{
+			/*
+			 * If a PGRES_PIPELINE_SYNC is followed by something other than
+			 * PGRES_PIPELINE_SYNC or NULL, another PGRES_PIPELINE_SYNC will
+			 * appear later. Reset received_sync to false to wait for it.
+			 */
+			received_sync = false;
+		}
 		PQclear(res);
 	}