Location via proxy:   [ UP ]  
[Report a bug]   [Manage cookies]                
Skip to content

Commit 91fa853

Browse files
committed
Attempt to fix error recovery in COPY BOTH mode.
Previously, libpq and the backend had opposite ideas about whether it was necessary for the client to send a CopyDone message after receiving an ErrorResponse, making it impossible to cleanly exit COPY BOTH mode. Fix libpq so that works correctly, adopting the backend's notion that an ErrorResponse kills the copy in both directions. Adjust receivelog.c to avoid a degradation in the quality of the resulting error messages. libpqwalreceiver.c is already doing the right thing, so no adjustment needed there. Add an explicit statement to the documentation explaining how this part of the protocol is supposed to work, in the hopes of avoiding future confusion in this area. Since the consequences of all this confusion are very limited, especially in the back-branches where no client ever attempts to exit COPY BOTH mode without closing the connection entirely, no back-patch.
1 parent 43e7a66 commit 91fa853

File tree

3 files changed

+51
-36
lines changed

3 files changed

+51
-36
lines changed

doc/src/sgml/protocol.sgml

+7-2
Original file line numberDiff line numberDiff line change
@@ -1031,8 +1031,13 @@
10311031
goes into copy-in mode, and the server may not send any more CopyData
10321032
messages. After both sides have sent a CopyDone message, the copy mode
10331033
is terminated, and the backend reverts to the command-processing mode.
1034-
See <xref linkend="protocol-replication"> for more information on the
1035-
subprotocol transmitted over copy-both mode.
1034+
In the event of a backend-detected error during copy-both mode,
1035+
the backend will issue an ErrorResponse message, discard frontend messages
1036+
until a Sync message is received, and then issue ReadyForQuery and return
1037+
to normal processing. The frontend should treat receipt of ErrorResponse
1038+
as terminating the copy in both directions; no CopyDone should be sent
1039+
in this case. See <xref linkend="protocol-replication"> for more
1040+
information on the subprotocol transmitted over copy-both mode.
10361041
</para>
10371042

10381043
<para>

src/bin/pg_basebackup/receivelog.c

+28-19
Original file line numberDiff line numberDiff line change
@@ -32,10 +32,10 @@
3232
static int walfile = -1;
3333
static char current_walfile_name[MAXPGPATH] = "";
3434

35-
static bool HandleCopyStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
36-
char *basedir, stream_stop_callback stream_stop,
37-
int standby_message_timeout, char *partial_suffix,
38-
XLogRecPtr *stoppos);
35+
static PGresult *HandleCopyStream(PGconn *conn, XLogRecPtr startpos,
36+
uint32 timeline, char *basedir,
37+
stream_stop_callback stream_stop, int standby_message_timeout,
38+
char *partial_suffix, XLogRecPtr *stoppos);
3939

4040
/*
4141
* Open a new WAL file in the specified directory.
@@ -615,9 +615,10 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
615615
PQclear(res);
616616

617617
/* Stream the WAL */
618-
if (!HandleCopyStream(conn, startpos, timeline, basedir, stream_stop,
619-
standby_message_timeout, partial_suffix,
620-
&stoppos))
618+
res = HandleCopyStream(conn, startpos, timeline, basedir, stream_stop,
619+
standby_message_timeout, partial_suffix,
620+
&stoppos);
621+
if (res == NULL)
621622
goto error;
622623

623624
/*
@@ -630,7 +631,6 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
630631
* restart streaming from the next timeline.
631632
*/
632633

633-
res = PQgetResult(conn);
634634
if (PQresultStatus(res) == PGRES_TUPLES_OK)
635635
{
636636
/*
@@ -708,10 +708,11 @@ ReceiveXlogStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
708708
* The main loop of ReceiveXLogStream. Handles the COPY stream after
709709
* initiating streaming with the START_STREAMING command.
710710
*
711-
* If the COPY ends normally, returns true and sets *stoppos to the last
712-
* byte written. On error, returns false.
711+
* If the COPY ends (not necessarily successfully) due a message from the
712+
* server, returns a PGresult and sets sets *stoppos to the last byte written.
713+
* On any other sort of error, returns NULL.
713714
*/
714-
static bool
715+
static PGresult *
715716
HandleCopyStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
716717
char *basedir, stream_stop_callback stream_stop,
717718
int standby_message_timeout, char *partial_suffix,
@@ -832,9 +833,12 @@ HandleCopyStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
832833
}
833834
if (r == -1)
834835
{
836+
PGresult *res = PQgetResult(conn);
837+
835838
/*
836-
* The server closed its end of the copy stream. Close ours
837-
* if we haven't done so already, and exit.
839+
* The server closed its end of the copy stream. If we haven't
840+
* closed ours already, we need to do so now, unless the server
841+
* threw an error, in which case we don't.
838842
*/
839843
if (still_sending)
840844
{
@@ -843,18 +847,23 @@ HandleCopyStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
843847
/* Error message written in close_walfile() */
844848
goto error;
845849
}
846-
if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
850+
if (PQresultStatus(res) == PGRES_COPY_IN)
847851
{
848-
fprintf(stderr, _("%s: could not send copy-end packet: %s"),
849-
progname, PQerrorMessage(conn));
850-
goto error;
852+
if (PQputCopyEnd(conn, NULL) <= 0 || PQflush(conn))
853+
{
854+
fprintf(stderr,
855+
_("%s: could not send copy-end packet: %s"),
856+
progname, PQerrorMessage(conn));
857+
goto error;
858+
}
859+
res = PQgetResult(conn);
851860
}
852861
still_sending = false;
853862
}
854863
if (copybuf != NULL)
855864
PQfreemem(copybuf);
856865
*stoppos = blockpos;
857-
return true;
866+
return res;
858867
}
859868
if (r == -2)
860869
{
@@ -1030,5 +1039,5 @@ HandleCopyStream(PGconn *conn, XLogRecPtr startpos, uint32 timeline,
10301039
error:
10311040
if (copybuf != NULL)
10321041
PQfreemem(copybuf);
1033-
return false;
1042+
return NULL;
10341043
}

src/interfaces/libpq/fe-protocol3.c

+16-15
Original file line numberDiff line numberDiff line change
@@ -1466,7 +1466,23 @@ getCopyDataMessage(PGconn *conn)
14661466
break;
14671467
case 'd': /* Copy Data, pass it back to caller */
14681468
return msgLength;
1469+
case 'c':
1470+
/*
1471+
* If this is a CopyDone message, exit COPY_OUT mode and let
1472+
* caller read status with PQgetResult(). If we're in
1473+
* COPY_BOTH mode, return to COPY_IN mode.
1474+
*/
1475+
if (conn->asyncStatus == PGASYNC_COPY_BOTH)
1476+
conn->asyncStatus = PGASYNC_COPY_IN;
1477+
else
1478+
conn->asyncStatus = PGASYNC_BUSY;
1479+
return -1;
14691480
default: /* treat as end of copy */
1481+
/*
1482+
* Any other message terminates either COPY_IN or COPY_BOTH
1483+
* mode.
1484+
*/
1485+
conn->asyncStatus = PGASYNC_BUSY;
14701486
return -1;
14711487
}
14721488

@@ -1499,22 +1515,7 @@ pqGetCopyData3(PGconn *conn, char **buffer, int async)
14991515
*/
15001516
msgLength = getCopyDataMessage(conn);
15011517
if (msgLength < 0)
1502-
{
1503-
/*
1504-
* On end-of-copy, exit COPY_OUT or COPY_BOTH mode and let caller
1505-
* read status with PQgetResult(). The normal case is that it's
1506-
* Copy Done, but we let parseInput read that. If error, we
1507-
* expect the state was already changed.
1508-
*/
1509-
if (msgLength == -1)
1510-
{
1511-
if (conn->asyncStatus == PGASYNC_COPY_BOTH)
1512-
conn->asyncStatus = PGASYNC_COPY_IN;
1513-
else
1514-
conn->asyncStatus = PGASYNC_BUSY;
1515-
}
15161518
return msgLength; /* end-of-copy or error */
1517-
}
15181519
if (msgLength == 0)
15191520
{
15201521
/* Don't block if async read requested */

0 commit comments

Comments
 (0)