From 1c87d613f499530b39702eb7d3009aad33242709 Mon Sep 17 00:00:00 2001
From: Arthur Zakirov <z-arthur@yandex.ru>
Date: Thu, 20 Apr 2017 12:38:51 +0300
Subject: [PATCH] Check target LSN. Archived WAL segment should contain it.

---
 backup.c       | 17 +++++++++++----
 parsexlog.c    | 59 ++++++++++++++++++++++++++++++++++++++++++++++++--
 pg_probackup.h |  2 ++
 restore.c      |  2 +-
 4 files changed, 73 insertions(+), 7 deletions(-)

diff --git a/backup.c b/backup.c
index 277f1b90..ed3930bf 100644
--- a/backup.c
+++ b/backup.c
@@ -239,7 +239,7 @@ do_backup_database(parray *backup_list)
 
 		if (ptrack_lsn > prev_backup->stop_lsn)
 		{
-			elog(ERROR, "lsn from ptrack_control %lx differs from lsn of previous ptrack backup %lx.\n"
+			elog(ERROR, "LSN from ptrack_control %lx differs from LSN of previous ptrack backup %lx.\n"
 						"Create new full backup before an incremental one.",
 						ptrack_lsn, prev_backup->start_lsn);
 		}
@@ -714,7 +714,8 @@ pg_ptrack_get_and_clear(Oid tablespace_oid, Oid db_oid, Oid rel_oid,
 }
 
 /*
- * TODO Add comment
+ * Wait for target 'lsn' to be archived in archive 'wal' directory with
+ * WAL segment file.
  */
 static void
 wait_archive_lsn(XLogRecPtr lsn, bool prev_segno)
@@ -729,7 +730,7 @@ wait_archive_lsn(XLogRecPtr lsn, bool prev_segno)
 
 	tli = get_current_timeline(false);
 
-	/* Compute the name of the WAL file containig requested lsn */
+	/* Compute the name of the WAL file containig requested LSN */
 	XLByteToSeg(lsn, targetSegNo);
 	if (prev_segno)
 		targetSegNo--;
@@ -747,7 +748,7 @@ wait_archive_lsn(XLogRecPtr lsn, bool prev_segno)
 
 		/* Inform user if WAL segment is absent in first attempt */
 		if (try_count == 1)
-			elog(INFO, "wait for lsn %X/%X in archived WAL segment %s",
+			elog(INFO, "wait for LSN %X/%X in archived WAL segment %s",
 				 (uint32) (lsn >> 32), (uint32) lsn, wal_path);
 
 		if (archive_timeout > 0 && try_count > archive_timeout)
@@ -755,6 +756,14 @@ wait_archive_lsn(XLogRecPtr lsn, bool prev_segno)
 				 "switched WAL segment %s could not be archived in %d seconds",
 				 wal_file, archive_timeout);
 	}
+
+	/*
+	 * WAL segment was archived. Check LSN on it if we waited current WAL
+	 * segment, not previous.
+	 */
+	if (!prev_segno && !wal_contains_lsn(arclog_path, lsn, tli))
+		elog(ERROR, "WAL segment %s doesn't contain target LSN %X/%X",
+			 wal_file, (uint32) (lsn >> 32), (uint32) lsn);
 }
 
 /*
diff --git a/parsexlog.c b/parsexlog.c
index 205bc607..22f734af 100644
--- a/parsexlog.c
+++ b/parsexlog.c
@@ -303,6 +303,7 @@ read_recovery_info(const char *archivedir, TimeLineID tli,
 	XLogRecPtr	startpoint = stop_lsn;
 	XLogReaderState *xlogreader;
 	XLogPageReadPrivate private;
+	bool		res;
 
 	private.archivedir = archivedir;
 	private.tli = tli;
@@ -342,12 +343,66 @@ read_recovery_info(const char *archivedir, TimeLineID tli,
 			*recovery_time = timestamptz_to_time_t(last_time);
 			*recovery_xid = XLogRecGetXid(xlogreader);
 
-			return true;
+			/* Found timestamp in WAL record 'record' */
+			res = true;
+			goto cleanup;
 		}
 	} while (startpoint >= start_lsn);
 
 	/* Didn't find timestamp from WAL records between start_lsn and stop_lsn */
-	return false;
+	res = false;
+
+cleanup:
+	XLogReaderFree(xlogreader);
+	if (xlogreadfd != -1)
+	{
+		close(xlogreadfd);
+		xlogreadfd = -1;
+		xlogexists = false;
+	}
+
+	return res;
+}
+
+/*
+ * Check if WAL segment file 'wal_path' contains 'target_lsn'.
+ */
+bool
+wal_contains_lsn(const char *archivedir, XLogRecPtr target_lsn,
+				 TimeLineID target_tli)
+{
+	XLogReaderState *xlogreader;
+	XLogPageReadPrivate private;
+	char	   *errormsg;
+	bool		res;
+
+	private.archivedir = archivedir;
+	private.tli = target_tli;
+
+	xlogreader = XLogReaderAllocate(&SimpleXLogPageRead, &private);
+	if (xlogreader == NULL)
+		elog(ERROR, "out of memory");
+
+	res = XLogReadRecord(xlogreader, target_lsn, &errormsg) != NULL;
+	if (!res)
+	{
+		if (errormsg)
+			elog(ERROR, "could not read WAL record at %X/%X: %s",
+				 (uint32) (target_lsn >> 32), (uint32) (target_lsn),
+				 errormsg);
+
+		/* Didn't find 'target_lsn' and there is no error, return false */
+	}
+
+	XLogReaderFree(xlogreader);
+	if (xlogreadfd != -1)
+	{
+		close(xlogreadfd);
+		xlogreadfd = -1;
+		xlogexists = false;
+	}
+
+	return res;
 }
 
 /* XLogreader callback function, to read a WAL page */
diff --git a/pg_probackup.h b/pg_probackup.h
index 1d0948c1..d6bf3313 100644
--- a/pg_probackup.h
+++ b/pg_probackup.h
@@ -368,6 +368,8 @@ extern bool read_recovery_info(const char *archivedir, TimeLineID tli,
 							   XLogRecPtr start_lsn, XLogRecPtr stop_lsn,
 							   time_t *recovery_time,
 							   TransactionId *recovery_xid);
+extern bool wal_contains_lsn(const char *archivedir, XLogRecPtr target_lsn,
+							 TimeLineID target_tli);
 
 /* in util.c */
 extern TimeLineID get_current_timeline(bool safe);
diff --git a/restore.c b/restore.c
index 71fe3d4f..be6fb5c1 100644
--- a/restore.c
+++ b/restore.c
@@ -790,7 +790,7 @@ readTimeLineHistory_probackup(TimeLineID targetTLI)
 	/* append target timeline */
 	entry = pgut_new(TimeLineHistoryEntry);
 	entry->tli = targetTLI;
-	/* lsn in target timeline is valid */
+	/* LSN in target timeline is valid */
 	entry->end = (uint32) (-1UL << 32) | -1UL;
 	parray_insert(result, 0, entry);