Feedback-ID: ia2694551:Fastmail
Date: Fri, 1 Dec 2023 15:20:28 +0100
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
To: 
	=?utf-8?B?0KLRgNC+0YTQuNC80L7QsiDQmNCy0LDQvQ==?= <i.trofimow@yandex.ru>
Cc: "pgsql-bugs@lists.postgresql.org" <pgsql-bugs@lists.postgresql.org>
Subject: Re: BUG #17948: libpq seems to misbehave in a pipelining corner case
Message-ID: <202312011420.zmxtoxtvu6ip@alvherre.pgsql>
MIME-Version: 1.0
Content-Type: multipart/mixed; boundary="whoouggmli4wkqib"
Content-Disposition: inline
Content-Transfer-Encoding: 8bit
In-Reply-To: <82241691236930@mail.yandex.ru>
Archived-At: 
 <https://www.postgresql.org/message-id/202312011420.zmxtoxtvu6ip%40alvherre.pgsql>
Precedence: bulk


--whoouggmli4wkqib
Content-Type: text/plain; charset=utf-8
Content-Disposition: inline
Content-Transfer-Encoding: 8bit

On 2023-Aug-05, Трофимов Иван wrote:

>    We've managed to reproduce client <-> server de-synchronization triggered
>    by this, which we are seeing in production.
>    Libpq considers '< 2TDCEZ' a sufficient response to '> BDESS', when
>    according to specification one more 'Z' is expected.
>     
>    A bit more context and a MRE:
>    https://github.com/itrofimow/libpq_protocol_desync

Does this patch fix the problems?  I think the sync loss is because
we're consuming elements from the command queue even when we shouldn't,
because we don't know for an ERROR whether we should do so or not.  What
this patch does is a bit of a hack, I think, but it does solve your test
case (and it doesn't break anything else in the tree) AFAICS.  I may
have to play some more with it before gaining confidence, though.

Thanks

-- 
Álvaro Herrera         PostgreSQL Developer  —  https://www.EnterpriseDB.com/
"Hay dos momentos en la vida de un hombre en los que no debería
especular: cuando puede permitírselo y cuando no puede" (Mark Twain)

--whoouggmli4wkqib
Content-Type: text/x-diff; charset=utf-8
Content-Disposition: attachment; filename="libpq.patch"

diff --git a/src/interfaces/libpq/fe-exec.c b/src/interfaces/libpq/fe-exec.c
index 04610ccf5e..f19edaa5fd 100644
--- a/src/interfaces/libpq/fe-exec.c
+++ b/src/interfaces/libpq/fe-exec.c
@@ -2112,19 +2112,24 @@ PQgetResult(PGconn *conn)
 			break;
 
 		case PGASYNC_READY:
+			res = pqPrepareAsyncResult(conn);
 
 			/*
-			 * For any query type other than simple query protocol, we advance
-			 * the command queue here.  This is because for simple query
-			 * protocol we can get the READY state multiple times before the
-			 * command is actually complete, since the command string can
-			 * contain many queries.  In simple query protocol, the queue
-			 * advance is done by fe-protocol3 when it receives ReadyForQuery.
+			 * Tell pqCommandQueueAdvance to do its thing, if not in simple
+			 * query protocol.  XXX rewrite
+			 *
+			 * When returning a result that is not an error, for any query type
+			 * other than simple query protocol, we advance the command queue
+			 * here.  This is because for simple query protocol we can get the
+			 * READY state multiple times before the command is actually
+			 * complete, since the command string can contain many queries.  In
+			 * simple query protocol, the queue advance is done by fe-protocol3
+			 * when it receives ReadyForQuery.
 			 */
 			if (conn->cmd_queue_head &&
 				conn->cmd_queue_head->queryclass != PGQUERY_SIMPLE)
-				pqCommandQueueAdvance(conn);
-			res = pqPrepareAsyncResult(conn);
+				pqCommandQueueAdvance(conn, res->resultStatus == PGRES_FATAL_ERROR);
+
 			if (conn->pipelineStatus != PQ_PIPELINE_OFF)
 			{
 				/*
@@ -3090,15 +3095,28 @@ PQexitPipelineMode(PGconn *conn)
  * pqCommandQueueAdvance
  *		Remove one query from the command queue, when we receive
  *		all results from the server that pertain to it.
+ *
+ * Note: when the result we're processing is an error, we must only advance
+ * the queue for message types other than SYNC.  This is because we might
+ * get other errors arbitrarily (say, connection closed), but we must not go
+ * past the synchronization point until the server agrees that that's where we
+ * are.
  */
 void
-pqCommandQueueAdvance(PGconn *conn)
+pqCommandQueueAdvance(PGconn *conn, bool isError)
 {
 	PGcmdQueueEntry *prevquery;
 
 	if (conn->cmd_queue_head == NULL)
 		return;
 
+	/*
+	 * In case of error, we only allow the queue to advance if the entry
+	 * to be removed is not SYNC.
+	 */
+	if (isError && conn->cmd_queue_head->queryclass == PGQUERY_SYNC)
+		return;
+
 	/* delink from queue */
 	prevquery = conn->cmd_queue_head;
 	conn->cmd_queue_head = conn->cmd_queue_head->next;
@@ -3162,7 +3180,8 @@ pqPipelineProcessQueue(PGconn *conn)
 	 */
 	if (conn->cmd_queue_head == NULL)
 	{
-		conn->asyncStatus = PGASYNC_IDLE;
+		if (conn->pipelineStatus != PQ_PIPELINE_ABORTED)
+			conn->asyncStatus = PGASYNC_IDLE;
 		return;
 	}
 
diff --git a/src/interfaces/libpq/fe-protocol3.c b/src/interfaces/libpq/fe-protocol3.c
index 5613c56b14..017999057d 100644
--- a/src/interfaces/libpq/fe-protocol3.c
+++ b/src/interfaces/libpq/fe-protocol3.c
@@ -246,7 +246,7 @@ pqParseInput3(PGconn *conn)
 						 */
 						if (conn->cmd_queue_head &&
 							conn->cmd_queue_head->queryclass == PGQUERY_SIMPLE)
-							pqCommandQueueAdvance(conn);
+							pqCommandQueueAdvance(conn, false);
 						conn->asyncStatus = PGASYNC_IDLE;
 					}
 					break;
diff --git a/src/interfaces/libpq/libpq-int.h b/src/interfaces/libpq/libpq-int.h
index 22bc682ffc..8518ac6d76 100644
--- a/src/interfaces/libpq/libpq-int.h
+++ b/src/interfaces/libpq/libpq-int.h
@@ -698,7 +698,7 @@ extern void pqSaveMessageField(PGresult *res, char code,
 extern void pqSaveParameterStatus(PGconn *conn, const char *name,
 								  const char *value);
 extern int	pqRowProcessor(PGconn *conn, const char **errmsgp);
-extern void pqCommandQueueAdvance(PGconn *conn);
+extern void pqCommandQueueAdvance(PGconn *conn, bool isError);
 extern int	PQsendQueryContinue(PGconn *conn, const char *query);
 
 /* === in fe-protocol3.c === */

--whoouggmli4wkqib--