Check WALs in validate command. Also support 64bit XID from PGPRO_EE. issue #5

Author: stalkerg

backup.c

@@ -64,7 +64,7 @@ static void pg_start_backup(const char *label, bool smooth, pgBackup *backup);
 static void pg_stop_backup(pgBackup *backup);
 static bool pg_is_standby(void);
 static void get_lsn(PGconn *conn, PGresult *res, XLogRecPtr *lsn, bool stop_backup);
-static void get_xid(PGresult *res, uint32 *xid);
+static void get_xid(PGresult *res, TransactionId *xid);
 static void pg_ptrack_clear(void);
 static bool pg_ptrack_support(void);
 static bool pg_ptrack_enable(void);
@@ -1042,14 +1042,14 @@ get_lsn(PGconn *conn, PGresult *res, XLogRecPtr *lsn, bool stop_backup)
  * Get XID from result of txid_current() after pg_stop_backup().
  */
 static void
-get_xid(PGresult *res, uint32 *xid)
+get_xid(PGresult *res, TransactionId *xid)
 {
 	if (res == NULL || PQntuples(res) != 1 || PQnfields(res) != 1)
 		elog(ERROR,
 			"result of txid_current() is invalid: %s",
 			PQerrorMessage(connection));
 
-	if (sscanf(PQgetvalue(res, 0, 0), "%u", xid) != 1)
+	if (sscanf(PQgetvalue(res, 0, 0), XID_FMT, xid) != 1)
 	{
 		elog(ERROR,
 			"result of txid_current() is invalid: %s",

catalog.c

@@ -331,21 +331,21 @@ catalog_read_ini(const char *path)
 
 	pgut_option options[] =
 	{
-		{ 's', 0, "backup-mode"			, NULL, SOURCE_ENV },
-		{ 'u', 0, "timelineid"			, NULL, SOURCE_ENV },
-		{ 's', 0, "start-lsn"			, NULL, SOURCE_ENV },
-		{ 's', 0, "stop-lsn"			, NULL, SOURCE_ENV },
-		{ 't', 0, "start-time"			, NULL, SOURCE_ENV },
-		{ 't', 0, "end-time"			, NULL, SOURCE_ENV },
-		{ 'u', 0, "recovery-xid"				, NULL, SOURCE_ENV },
-		{ 't', 0, "recovery-time"				, NULL, SOURCE_ENV },
-		{ 'I', 0, "data-bytes"		, NULL, SOURCE_ENV },
-		{ 'u', 0, "block-size"			, NULL, SOURCE_ENV },
-		{ 'u', 0, "xlog-block-size"		, NULL, SOURCE_ENV },
-		{ 'u', 0, "checksum_version"		, NULL, SOURCE_ENV },
-		{ 'u', 0, "stream"				, NULL, SOURCE_ENV },
-		{ 's', 0, "status"				, NULL, SOURCE_ENV },
-		{ 0 }
+		{'s', 0, "backup-mode",			NULL, SOURCE_ENV},
+		{'u', 0, "timelineid",			NULL, SOURCE_ENV},
+		{'s', 0, "start-lsn",			NULL, SOURCE_ENV},
+		{'s', 0, "stop-lsn",			NULL, SOURCE_ENV},
+		{'t', 0, "start-time",			NULL, SOURCE_ENV},
+		{'t', 0, "end-time",			NULL, SOURCE_ENV},
+		{'U', 0, "recovery-xid",		NULL, SOURCE_ENV},
+		{'t', 0, "recovery-time",		NULL, SOURCE_ENV},
+		{'I', 0, "data-bytes",			NULL, SOURCE_ENV},
+		{'u', 0, "block-size",			NULL, SOURCE_ENV},
+		{'u', 0, "xlog-block-size",		NULL, SOURCE_ENV},
+		{'u', 0, "checksum_version",	NULL, SOURCE_ENV},
+		{'u', 0, "stream",				NULL, SOURCE_ENV},
+		{'s', 0, "status",				NULL, SOURCE_ENV},
+		{0}
 	};
 
 	if (access(path, F_OK) != 0)

parsexlog.c

@@ -17,6 +17,7 @@
 
 #include "commands/dbcommands_xlog.h"
 #include "catalog/storage_xlog.h"
+#include "access/transam.h"
 
 /*
  * RmgrNames is an array of resource manager names, to make error messages
@@ -100,6 +101,85 @@ extractPageMap(const char *archivedir, XLogRecPtr startpoint, TimeLineID tli,
 	}
 }
 
+void
+validate_wal(pgBackup *backup,
+			 const char *archivedir,
+			 XLogRecPtr startpoint,
+			 time_t target_time,
+			 TransactionId recovery_target_xid,
+			 TimeLineID tli)
+{
+	XLogRecord *record;
+	XLogReaderState *xlogreader;
+	char	   *errormsg;
+	XLogPageReadPrivate private;
+	TransactionId last_xid = InvalidTransactionId;
+	TimestampTz last_time = 0;
+	char	timestamp[100];
+
+	private.archivedir = archivedir;
+	private.tli = tli;
+	xlogreader = XLogReaderAllocate(&SimpleXLogPageRead, &private);
+	if (xlogreader == NULL)
+		elog(ERROR, "out of memory");
+
+	while (true)
+	{
+		record = XLogReadRecord(xlogreader, startpoint, &errormsg);
+		bool timestamp_record;
+		if (record == NULL)
+		{
+			XLogRecPtr	errptr;
+
+			errptr = startpoint ? startpoint : xlogreader->EndRecPtr;
+			if (recovery_target_xid == InvalidTransactionId && target_time == 0)
+			{
+				break;
+			}
+
+			if (errormsg)
+				elog(ERROR, "stop check WALs because could not read WAL record at %X/%X: %s\nend time:%s end xid:" XID_FMT,
+						 (uint32) (errptr >> 32), (uint32) (errptr),
+						 errormsg,
+						 timestamp,
+						 last_xid);
+			else
+				elog(ERROR, "could not read WAL record at %X/%X\nend time:%s end xid:" XID_FMT,
+						 (uint32) (errptr >> 32),
+						 (uint32) (errptr),
+						 timestamp,
+						 last_xid);
+		}
+
+		timestamp_record = getRecordTimestamp(xlogreader, &last_time);
+		last_xid = XLogRecGetXid(xlogreader);
+		if (recovery_target_xid != InvalidTransactionId && recovery_target_xid == last_xid)
+			break;
+
+		if (target_time != 0 && timestamp_record && timestamptz_to_time_t(last_time) >= target_time)
+			break;
+
+		startpoint = InvalidXLogRecPtr; /* continue reading at next record */
+	}
+
+	if (last_time > 0)
+		time2iso(timestamp, lengthof(timestamp), timestamptz_to_time_t(last_time));
+	else
+		time2iso(timestamp, lengthof(timestamp), backup->recovery_time);
+	if (last_xid == InvalidTransactionId)
+		last_xid = backup->recovery_xid;
+
+	elog(INFO, "Validate WAL stoped on %s time and xid:" XID_FMT, timestamp, last_xid);
+
+	/* clean */
+	XLogReaderFree(xlogreader);
+	if (xlogreadfd != -1)
+	{
+		close(xlogreadfd);
+		xlogreadfd = -1;
+	}
+}
+
 /* XLogreader callback function, to read a WAL page */
 static int
 SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
@@ -138,7 +218,7 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr,
 
 		if (xlogreadfd < 0)
 		{
-			elog(WARNING, "could not open WAL segment \"%s\": %s",
+			elog(INFO, "could not open WAL segment \"%s\": %s",
 				 xlogfpath, strerror(errno));
 			return -1;
 		}

pg_probackup.h

@@ -45,6 +45,10 @@
 #define DIR_PERMISSION		(0700)
 #define FILE_PERMISSION		(0600)
 
+#ifndef PGPRO_EE
+#define XID_FMT "%u"
+#endif
+
 /* backup mode file */
 typedef struct pgFile
 {
@@ -107,33 +111,33 @@ typedef enum BackupMode
 typedef struct pgBackup
 {
 	/* Backup Level */
-	BackupMode	backup_mode;
+	BackupMode		backup_mode;
 
 	/* Status - one of BACKUP_STATUS_xxx */
 	BackupStatus	status;
 
 	/* Timestamp, etc. */
-	TimeLineID	tli;
-	XLogRecPtr	start_lsn;
-	XLogRecPtr	stop_lsn;
-	time_t		start_time;
-	time_t		end_time;
-	time_t		recovery_time;
-	uint32		recovery_xid;
+	TimeLineID		tli;
+	XLogRecPtr		start_lsn;
+	XLogRecPtr		stop_lsn;
+	time_t			start_time;
+	time_t			end_time;
+	time_t			recovery_time;
+	TransactionId	recovery_xid;
 
 	/* Different sizes (-1 means nothing was backed up) */
 	/*
 	 * Amount of raw data. For a full backup, this is the total amount of
 	 * data while for a differential backup this is just the difference
 	 * of data taken.
 	 */
-	int64		data_bytes;
+	int64			data_bytes;
 
 	/* data/wal block size for compatibility check */
-	uint32		block_size;
-	uint32		wal_block_size;
-	uint32		checksum_version;
-	bool		stream;
+	uint32			block_size;
+	uint32			wal_block_size;
+	uint32			checksum_version;
+	bool			stream;
 } pgBackup;
 
 typedef struct pgBackupOption
@@ -156,11 +160,11 @@ typedef struct pgTimeLine
 
 typedef struct pgRecoveryTarget
 {
-	bool		time_specified;
-	time_t		recovery_target_time;
-	bool		xid_specified;
-	unsigned int	recovery_target_xid;
-	bool		recovery_target_inclusive;
+	bool			time_specified;
+	time_t			recovery_target_time;
+	bool			xid_specified;
+	TransactionId	recovery_target_xid;
+	bool			recovery_target_inclusive;
 } pgRecoveryTarget;
 
 typedef union DataPage
@@ -313,8 +317,16 @@ extern bool copy_file(const char *from_root, const char *to_root,
 extern bool calc_file(pgFile *file);
 
 /* parsexlog.c */
-extern void extractPageMap(const char *datadir, XLogRecPtr startpoint,
-						   TimeLineID tli, XLogRecPtr endpoint);
+extern void extractPageMap(const char *datadir,
+						   XLogRecPtr startpoint,
+						   TimeLineID tli,
+						   XLogRecPtr endpoint);
+extern void validate_wal(pgBackup *backup,
+						 const char *archivedir,
+						 XLogRecPtr startpoint,
+						 time_t target_time,
+						 TransactionId recovery_target_xid,
+						 TimeLineID tli);
 
 /* in util.c */
 extern TimeLineID get_current_timeline(bool safe);
@@ -328,6 +340,7 @@ extern uint32 get_data_checksum_version(bool safe);
 extern char *base36enc(long unsigned int value);
 extern long unsigned int base36dec(const char *text);
 extern uint64 get_system_identifier(bool safe);
+extern pg_time_t timestamptz_to_time_t(TimestampTz t);
 
 /* in status.c */
 extern bool is_pg_running(void);

restore.c

@@ -777,9 +777,9 @@ checkIfCreateRecoveryConf(const char *target_time,
                    const char *target_xid,
                    const char *target_inclusive)
 {
-	time_t		dummy_time;
-	unsigned int	dummy_xid;
-	bool		dummy_bool;
+	time_t			dummy_time;
+	TransactionId	dummy_xid;
+	bool			dummy_bool;
 	pgRecoveryTarget *rt;
 
 	/* Initialize pgRecoveryTarget */
@@ -801,7 +801,11 @@ checkIfCreateRecoveryConf(const char *target_time,
 	if (target_xid)
 	{
 		rt->xid_specified = true;
+#ifdef PGPRO_EE
+		if (parse_uint64(target_xid, &dummy_xid))
+#else
 		if (parse_uint32(target_xid, &dummy_xid))
+#endif
 			rt->recovery_target_xid = dummy_xid;
 		else
 			elog(ERROR, "cannot create recovery.conf with %s", target_xid);

util.c

@@ -168,6 +168,22 @@ time2iso(char *buf, size_t len, time_t time)
 	strftime(buf, len, "%Y-%m-%d %H:%M:%S", tm);
 }
 
+/* copied from timestamp.c */
+pg_time_t
+timestamptz_to_time_t(TimestampTz t)
+{
+	pg_time_t	result;
+
+#ifdef HAVE_INT64_TIMESTAMP
+	result = (pg_time_t) (t / USECS_PER_SEC +
+				 ((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY));
+#else
+	result = (pg_time_t) (t +
+				 ((POSTGRES_EPOCH_JDATE - UNIX_EPOCH_JDATE) * SECS_PER_DAY));
+#endif
+	return result;
+}
+
 const char *
 status2str(BackupStatus status)
 {

validate.c

@@ -193,6 +193,15 @@ int do_validate(time_t backup_id,
 	}
 
 	/* and now we must check WALs */
+	{
+		pgBackup *backup = (pgBackup *) parray_get(backups, last_restored_index);
+		validate_wal(backup,
+					 arclog_path,
+					 backup->start_lsn,
+					 rt->recovery_target_time,
+					 rt->recovery_target_xid,
+					 target_tli);
+	}
 
 	/* release catalog lock */
 	catalog_unlock();