exactly what is COPY BOTH mode supposed to do in case of an error?

Started by Robert Haasover 12 years ago5 messages

robertmhaas@gmail.com

over 12 years ago

1 attachment(s)

It seems the backend and libpq don't agree. The backend makes no
special provision to wait for a CopyDone message if an error occurs
during copy-both. It simply sends an ErrorResponse and that's it.
libpq, on the other hand, treats either CopyDone or ErrorResponse as a
cue to transition to PGASYNC_COPY_IN (see pqGetCopyData3).

And that's a problem, because in PGASYNC_COPY_IN mode, libpq is
unwilling to process the ErrorResponse. getCopyDataMessage doesn't do
anything with it, and if the user calls PQgetResult(), pqParseInput3()
won't process it either, because it's only willing to do that when the
status is PGASYNC_BUSY. So the bottom line is that after the server
throws an error, the server gets back to thinking that the connection
is idle, while the client ends up thinking that we're still in copy-in
mode. The client can try to recover by doing PQputCopyEnd(), which
will get libpq back to an idle state, but if the extended-query
protocol is in use, the Sync it sends will cause the server to send an
extra ReadyForQuery that libpq isn't expecting. Hilarity ensues.

It's perhaps not coincidental that "48.2.5 COPY operations" is silent
about what is supposed to happen when an error occurs in copy-both
mode, though it does talk about both copy-in and copy-out. On
copy-in, it says:

In the event of a backend-detected error during copy-in mode
(including receipt of a CopyFail message), the backend will issue an
ErrorResponse message. If the COPY command was issued via an
extended-query message, the backend will now discard frontend messages
until a Sync message is received, then it will issue ReadyForQuery and
return to normal processing.

And regarding copy-out, it says:

In the event of a backend-detected error during copy-out mode, the
backend will issue an ErrorResponse message and revert to normal
processing. The frontend should treat receipt of ErrorResponse as
terminating the copy-out mode.

There's no corresponding statement about error-handling in the
copy-both case. However, since the apparent intent is that an error
message from the server trumps anything the client may have had in
mind, it seems reasonable to decide that an ErrorResponse is intended
to fully terminate copy-both mode (not just switch to copy-in) and
initiate a skip-until-sync. That's what the server actually does, but
libpq has other ideas.

One way to see the practical effect of this is to set up a streaming
replication slave but modify the backup label to reference some future
WAL location not yet written (I just changed the "0" before the slash
to a "1"). This will cause the server to throw the following error:

ereport(ERROR,
(errmsg("requested starting
point %X/%X is ahead of the WAL flush position of this server %X/%X",
(uint32)
(cmd->startpoint >> 32),
(uint32)
(cmd->startpoint),
(uint32)
(FlushPtr >> 32),
(uint32) (FlushPtr))));

Doing this will cause the PQgetCopyData in libpqrcv_receive to return
-1, so you might expected that slave would get the following error:

ereport(ERROR,
(errmsg("could not receive
data from WAL stream: %s",

PQerrorMessage(streamConn))));

But you don't, because PQgetResult() returns PGRES_COPY_IN. So the
slave thinks that the master made a *normal* termination of streaming
replication, due to a timeline change, and it prints this:

LOG: replication terminated by primary server
DETAIL: End of WAL reached on timeline 1 at 0/0

It then calls PQputCopyEnd to end a copy that the server thinks is no
longer in progress and then invokes PQgetResult again. And now it
gets the error message:

FATAL: error reading result of streaming command: ERROR: requested
starting point 1/5000000 is ahead of the WAL flush position of this
server 0/6000348

This doesn't in practice matter very much, because either way, we're
going to slam shut the server connection at this point. But it's
clearly messed up - the error is actually NOT in response to the
CopyDone we sent, but rather in response to the START_STREAMING
command that preceded it. libpq, however, refuses to receive the error
message until after the unnecessary CopyDone has been sent.

I believe that the only other place where this coding pattern arises
is in receivelog.c. That code has actually adopted the opposite
convention from the backend: it thinks it needs to send CopyDone
whether or not an error has occurred. This turns out not to matter
particularly, because the server is just going to try to close the
connection anyway. But if it did anything else after ending the copy,
I suspect the wheels would come off.

I'm attaching a patch which adopts the position that the backend is
right and libpq is wrong. The opposite approach is also possible, but
I haven't tried to implement it. Or maybe there's a third way which
is better still.

--
Robert Haas
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

Attachments:

copy-both-terminate.patchapplication/octet-stream; name=copy-both-terminate.patchDownload

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index d326266..1e2604b 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -1031,8 +1031,13 @@
     goes into copy-in mode, and the server may not send any more CopyData
     messages. After both sides have sent a CopyDone message, the copy mode
     is terminated, and the backend reverts to the command-processing mode.
-    See <xref linkend="protocol-replication"> for more information on the
-    subprotocol transmitted over copy-both mode.
+    In the event of a backend-detected error during copy-both mode,
+    the backend will issue an ErrorResponse message, discard frontend messages
+    until a Sync message is received, and then issue ReadyForQuery and return
+    to normal processing.  The frontend should treat receipt of ErrorResponse
+    as terminating the copy in both directions; no CopyDone should be sent
+    in this case.  See <xref linkend="protocol-replication"> for more
+    information on the subprotocol transmitted over copy-both mode.
    </para>
 
    <para>
diff --git a/src/bin/pg_basebackup/receivelog.c b/src/bin/pg_basebackup/receivelog.c
index b681efc..f297003 100644
--- a/src/bin/pg_basebackup/receivelog.c
+++ b/src/bin/pg_basebackup/receivelog.c
@@ -32,10 +32,10 @@
 static int	walfile = -1;
 static char	current_walfile_name[MAXPGPATH] = "";
 
-static bool HandleCopyStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
-				 char *basedir, stream_stop_callback stream_stop,
-				 int standby_message_timeout, char *partial_suffix,
-				 XLogRecPtr *stoppos);
+static PGresult *HandleCopyStream(PGconn *conn, XLogRecPtr startpos,
+				 uint32 timeline, char *basedir,
+				 stream_stop_callback stream_stop, int standby_message_timeout,
+				 char *partial_suffix, XLogRecPtr *stoppos);
 
 /*
  * Open a new WAL file in the specified directory.
@@ -615,9 +615,10 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
 		PQclear(res);
 
 		/* Stream the WAL */
-		if (!HandleCopyStream(conn, startpos, timeline, basedir, stream_stop,
-							  standby_message_timeout, partial_suffix,
-							  &stoppos))
+		res = HandleCopyStream(conn, startpos, timeline, basedir, stream_stop,
+							   standby_message_timeout, partial_suffix,
+							   &stoppos);
+		if (res == NULL)
 			goto error;
 
 		/*
@@ -630,7 +631,6 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
 		 * restart streaming from the next timeline.
 		 */
 
-		res = PQgetResult(conn);
 		if (PQresultStatus(res) == PGRES_TUPLES_OK)
 		{
 			/*
@@ -708,10 +708,11 @@ error:
  * The main loop of ReceiveXLogStream. Handles the COPY stream after
  * initiating streaming with the START_STREAMING command.
  *
- * If the COPY ends normally, returns true and sets *stoppos to the last
- * byte written. On error, returns false.
+ * If the COPY ends (not necessarily successfully) due a message from the
+ * server, returns a PGresult and sets sets *stoppos to the last byte written.
+ * On any other sort of error, returns NULL.
  */
-static bool
+static PGresult *
 HandleCopyStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
 				 char *basedir, stream_stop_callback stream_stop,
 				 int standby_message_timeout, char *partial_suffix,
@@ -832,9 +833,12 @@ HandleCopyStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
 		}
 		if (r == -1)
 		{
+			PGresult   *res = PQgetResult(conn);
+
 			/*
-			 * The server closed its end of the copy stream. Close ours
-			 * if we haven't done so already, and exit.
+			 * The server closed its end of the copy stream.  If we haven't
+			 * closed ours already, we need to do so now, unless the server
+			 * threw an error, in which case we don't.
 			 */
 			if (still_sending)
 			{
@@ -843,18 +847,23 @@ HandleCopyStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
 					/* Error message written in close_walfile() */
 					goto error;
 				}
-				if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
+				if (PQresultStatus(res) == PGRES_COPY_IN)
 				{
-					fprintf(stderr, _("%s: could not send copy-end packet: %s"),
-							progname, PQerrorMessage(conn));
-					goto error;
+					if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
+					{
+						fprintf(stderr,
+								_("%s: could not send copy-end packet: %s"),
+								progname, PQerrorMessage(conn));
+						goto error;
+					}
+					res = PQgetResult(conn);
 				}
 				still_sending = false;
 			}
 			if (copybuf != NULL)
 				PQfreemem(copybuf);
 			*stoppos = blockpos;
-			return true;
+			return res;
 		}
 		if (r == -2)
 		{
@@ -1030,5 +1039,5 @@ HandleCopyStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
 error:
 	if (copybuf != NULL)
 		PQfreemem(copybuf);
-	return false;
+	return NULL;
 }
diff --git a/src/interfaces/libpq/fe-protocol3.c b/src/interfaces/libpq/fe-protocol3.c
index 7fa090a..a7d4f40 100644
--- a/src/interfaces/libpq/fe-protocol3.c
+++ b/src/interfaces/libpq/fe-protocol3.c
@@ -1466,7 +1466,23 @@ getCopyDataMessage(PGconn *conn)
 				break;
 			case 'd':			/* Copy Data, pass it back to caller */
 				return msgLength;
+			case 'c':
+				/*
+				 * If this is a CopyDone message, exit COPY_OUT mode and let
+				 * caller read status with PQgetResult().  If we're in
+				 * COPY_BOTH mode, return to COPY_IN mode.
+				 */
+				if (conn->asyncStatus == PGASYNC_COPY_BOTH)
+					conn->asyncStatus = PGASYNC_COPY_IN;
+				else
+					conn->asyncStatus = PGASYNC_BUSY;
+				return -1;
 			default:			/* treat as end of copy */
+				/*
+				 * Any other message terminates either COPY_IN or COPY_BOTH
+				 * mode.
+				 */
+				conn->asyncStatus = PGASYNC_BUSY;
 				return -1;
 		}
 
@@ -1499,22 +1515,7 @@ pqGetCopyData3(PGconn *conn, char **buffer, int async)
 		 */
 		msgLength = getCopyDataMessage(conn);
 		if (msgLength < 0)
-		{
-			/*
-			 * On end-of-copy, exit COPY_OUT or COPY_BOTH mode and let caller
-			 * read status with PQgetResult().	The normal case is that it's
-			 * Copy Done, but we let parseInput read that.	If error, we
-			 * expect the state was already changed.
-			 */
-			if (msgLength == -1)
-			{
-				if (conn->asyncStatus == PGASYNC_COPY_BOTH)
-					conn->asyncStatus = PGASYNC_COPY_IN;
-				else
-					conn->asyncStatus = PGASYNC_BUSY;
-			}
 			return msgLength;	/* end-of-copy or error */
-		}
 		if (msgLength == 0)
 		{
 			/* Don't block if async read requested */