Make pg_receivexlog and pg_basebackup -X stream work across timeline switches.

This mirrors the changes done earlier to the server in standby mode. When
receivelog reaches the end of a timeline, as reported by the server, it
fetches the timeline history file of the next timeline, and restarts
streaming from the new timeline by issuing a new START_STREAMING command.

When pg_receivexlog crosses a timeline, it leaves the .partial suffix on the
last segment on the old timeline. This helps you to tell apart a partial
segment left in the directory because of a timeline switch, and a completed
segment. If you just follow a single server, it won't make a difference, but
it can be significant in more complicated scenarios where new WAL is still
generated on the old timeline.

This includes two small changes to the streaming replication protocol:
First, when you reach the end of timeline while streaming, the server now
sends the TLI of the next timeline in the server's history to the client.
pg_receivexlog uses that as the next timeline, so that it doesn't need to
parse the timeline history file like a standby server does. Second, when
BASE_BACKUP command sends the begin and end WAL positions, it now also sends
the timeline IDs corresponding the positions.

1 files changed, 53 insertions, 46 deletions

diff --git a/src/bin/pg_basebackup/pg_receivexlog.c b/src/bin/pg_basebackup/pg_receivexlog.c
index 7f2db1946e7..33dbc50389b 100644
--- a/src/bin/pg_basebackup/pg_receivexlog.c
+++ b/src/bin/pg_basebackup/pg_receivexlog.c
@@ -39,8 +39,7 @@ volatile bool time_to_abort = false;
 
 
 static void usage(void);
-static XLogRecPtr FindStreamingStart(XLogRecPtr currentpos,
-				   uint32 currenttimeline);
+static XLogRecPtr FindStreamingStart(uint32 *tli);
 static void StreamLog();
 static bool stop_streaming(XLogRecPtr segendpos, uint32 timeline,
 			   bool segment_finished);
@@ -70,14 +69,31 @@ usage(void)
 }
 
 static bool
-stop_streaming(XLogRecPtr segendpos, uint32 timeline, bool segment_finished)
+stop_streaming(XLogRecPtr xlogpos, uint32 timeline, bool segment_finished)
 {
+	static uint32 prevtimeline = 0;
+	static XLogRecPtr prevpos = InvalidXLogRecPtr;
+
+	/* we assume that we get called once at the end of each segment */
 	if (verbose && segment_finished)
 		fprintf(stderr, _("%s: finished segment at %X/%X (timeline %u)\n"),
-				progname,
-				(uint32) (segendpos >> 32), (uint32) segendpos,
+				progname, (uint32) (xlogpos >> 32), (uint32) xlogpos,
 				timeline);
 
+	/*
+	 * Note that we report the previous, not current, position here. That's
+	 * the exact location where the timeline switch happend. After the switch,
+	 * we restart streaming from the beginning of the segment, so xlogpos can
+	 * smaller than prevpos if we just switched to new timeline.
+	 */
+	if (prevtimeline != 0 && prevtimeline != timeline)
+		fprintf(stderr, _("%s: switched to timeline %u at %X/%X\n"),
+				progname, timeline,
+				(uint32) (prevpos >> 32), (uint32) prevpos);
+
+	prevtimeline = timeline;
+	prevpos = xlogpos;
+
 	if (time_to_abort)
 	{
 		fprintf(stderr, _("%s: received interrupt signal, exiting\n"),
@@ -88,20 +104,19 @@ stop_streaming(XLogRecPtr segendpos, uint32 timeline, bool segment_finished)
 }
 
 /*
- * Determine starting location for streaming, based on:
- * 1. If there are existing xlog segments, start at the end of the last one
- *	  that is complete (size matches XLogSegSize)
- * 2. If no valid xlog exists, start from the beginning of the current
- *	  WAL segment.
+ * Determine starting location for streaming, based on any existing xlog
+ * segments in the directory. We start at the end of the last one that is
+ * complete (size matches XLogSegSize), on the timeline with highest ID.
+ *
+ * If there are no WAL files in the directory, returns InvalidXLogRecPtr.
  */
 static XLogRecPtr
-FindStreamingStart(XLogRecPtr currentpos, uint32 currenttimeline)
+FindStreamingStart(uint32 *tli)
 {
 	DIR		   *dir;
 	struct dirent *dirent;
-	int			i;
-	bool		b;
 	XLogSegNo	high_segno = 0;
+	uint32		high_tli = 0;
 
 	dir = opendir(basedir);
 	if (dir == NULL)
@@ -120,26 +135,13 @@ FindStreamingStart(XLogRecPtr currentpos, uint32 currenttimeline)
 					seg;
 		XLogSegNo	segno;
 
-		if (strcmp(dirent->d_name, ".") == 0 ||
-			strcmp(dirent->d_name, "..") == 0)
-			continue;
-
-		/* xlog files are always 24 characters */
-		if (strlen(dirent->d_name) != 24)
-			continue;
-
-		/* Filenames are always made out of 0-9 and A-F */
-		b = false;
-		for (i = 0; i < 24; i++)
-		{
-			if (!(dirent->d_name[i] >= '0' && dirent->d_name[i] <= '9') &&
-				!(dirent->d_name[i] >= 'A' && dirent->d_name[i] <= 'F'))
-			{
-				b = true;
-				break;
-			}
-		}
-		if (b)
+		/*
+		 * Check if the filename looks like an xlog file, or a .partial file.
+		 * Xlog files are always 24 characters, and .partial files are 32
+		 * characters.
+		 */
+		if (strlen(dirent->d_name) != 24 ||
+			!strspn(dirent->d_name, "0123456789ABCDEF") == 24)
 			continue;
 
 		/*
@@ -154,10 +156,6 @@ FindStreamingStart(XLogRecPtr currentpos, uint32 currenttimeline)
 		}
 		segno = ((uint64) log) << 32 | seg;
 
-		/* Ignore any files that are for another timeline */
-		if (tli != currenttimeline)
-			continue;
-
 		/* Check if this is a completed segment or not */
 		snprintf(fullpath, sizeof(fullpath), "%s/%s", basedir, dirent->d_name);
 		if (stat(fullpath, &statbuf) != 0)
@@ -170,9 +168,10 @@ FindStreamingStart(XLogRecPtr currentpos, uint32 currenttimeline)
 		if (statbuf.st_size == XLOG_SEG_SIZE)
 		{
 			/* Completed segment */
-			if (segno > high_segno)
+			if (segno > high_segno || (segno == high_segno && tli > high_tli))
 			{
 				high_segno = segno;
+				high_tli = tli;
 				continue;
 			}
 		}
@@ -199,10 +198,11 @@ FindStreamingStart(XLogRecPtr currentpos, uint32 currenttimeline)
 
 		XLogSegNoOffsetToRecPtr(high_segno, 0, high_ptr);
 
+		*tli = high_tli;
 		return high_ptr;
 	}
 	else
-		return currentpos;
+		return InvalidXLogRecPtr;
 }
 
 /*
@@ -212,8 +212,10 @@ static void
 StreamLog(void)
 {
 	PGresult   *res;
-	uint32		timeline;
 	XLogRecPtr	startpos;
+	uint32		starttli;
+	XLogRecPtr	serverpos;
+	uint32		servertli;
 	uint32		hi,
 				lo;
 
@@ -243,7 +245,7 @@ StreamLog(void)
 				progname, PQntuples(res), PQnfields(res), 1, 3);
 		disconnect_and_exit(1);
 	}
-	timeline = atoi(PQgetvalue(res, 0, 1));
+	servertli = atoi(PQgetvalue(res, 0, 1));
 	if (sscanf(PQgetvalue(res, 0, 2), "%X/%X", &hi, &lo) != 2)
 	{
 		fprintf(stderr,
@@ -251,13 +253,18 @@ StreamLog(void)
 				progname, PQgetvalue(res, 0, 2));
 		disconnect_and_exit(1);
 	}
-	startpos = ((uint64) hi) << 32 | lo;
+	serverpos = ((uint64) hi) << 32 | lo;
 	PQclear(res);
 
 	/*
 	 * Figure out where to start streaming.
 	 */
-	startpos = FindStreamingStart(startpos, timeline);
+	startpos = FindStreamingStart(&starttli);
+	if (startpos == InvalidXLogRecPtr)
+	{
+		startpos = serverpos;
+		starttli = servertli;
+	}
 
 	/*
 	 * Always start streaming at the beginning of a segment
@@ -271,10 +278,10 @@ StreamLog(void)
 		fprintf(stderr,
 				_("%s: starting log streaming at %X/%X (timeline %u)\n"),
 				progname, (uint32) (startpos >> 32), (uint32) startpos,
-				timeline);
+				starttli);
 
-	ReceiveXlogStream(conn, startpos, timeline, NULL, basedir,
-					  stop_streaming, standby_message_timeout, false);
+	ReceiveXlogStream(conn, startpos, starttli, NULL, basedir,
+					  stop_streaming, standby_message_timeout, ".partial");
 
 	PQfinish(conn);
 }