Merge branch 'master' into issue_310

Author: gsmolk

src/backup.c

@@ -56,7 +56,7 @@ static void pg_stop_backup(pgBackup *backup, PGconn *pg_startbackup_conn, PGNode
 
 static XLogRecPtr wait_wal_lsn(XLogRecPtr lsn, bool is_start_lsn, TimeLineID tli,
 								bool in_prev_segment, bool segment_only,
-								int timeout_elevel, bool in_stream_dir);
+								int timeout_elevel, bool in_stream_dir, pgBackup *backup);
 
 static void check_external_for_tablespaces(parray *external_list,
 										   PGconn *backup_conn);
@@ -270,7 +270,7 @@ do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync, bool
 		 * Because WAL streaming will start after pg_start_backup() in stream
 		 * mode.
 		 */
-		wait_wal_lsn(current.start_lsn, true, current.tli, false, true, ERROR, false);
+		wait_wal_lsn(current.start_lsn, true, current.tli, false, true, ERROR, false, &current);
 	}
 
 	/* start stream replication */
@@ -281,6 +281,12 @@ do_backup_instance(PGconn *backup_conn, PGNodeInfo *nodeInfo, bool no_sync, bool
 
 		start_WAL_streaming(backup_conn, dst_backup_path, &instance_config.conn_opt,
 							current.start_lsn, current.tli);
+
+		/* Make sure that WAL streaming is working
+		 * PAGE backup in stream mode is waited twice, first for
+		 * segment in WAL archive and then for streamed segment
+		 */
+		wait_wal_lsn(current.start_lsn, true, current.tli, false, true, ERROR, true, &current);
 	}
 
 	/* initialize backup's file list */
@@ -1264,7 +1270,7 @@ pg_is_superuser(PGconn *conn)
 static XLogRecPtr
 wait_wal_lsn(XLogRecPtr target_lsn, bool is_start_lsn, TimeLineID tli,
 			 bool in_prev_segment, bool segment_only,
-			 int timeout_elevel, bool in_stream_dir)
+			 int timeout_elevel, bool in_stream_dir, pgBackup *backup)
 {
 	XLogSegNo	targetSegNo;
 	char		pg_wal_dir[MAXPGPATH];
@@ -1296,15 +1302,14 @@ wait_wal_lsn(XLogRecPtr target_lsn, bool is_start_lsn, TimeLineID tli,
 	 */
 	if (in_stream_dir)
 	{
-		pgBackupGetPath2(&current, pg_wal_dir, lengthof(pg_wal_dir),
-						 DATABASE_DIR, PG_XLOG_DIR);
+		join_path_components(pg_wal_dir, backup->database_dir, PG_XLOG_DIR);
 		join_path_components(wal_segment_path, pg_wal_dir, wal_segment);
 		wal_segment_dir = pg_wal_dir;
 	}
 	else
 	{
 		join_path_components(wal_segment_path, arclog_path, wal_segment);
-		wal_segment_dir = arclog_path;
+		wal_segment_dir = arclog_path; /* global var */
 	}
 
 	/* TODO: remove this in 3.0 (it is a cludge against some old bug with archive_timeout) */
@@ -1396,7 +1401,7 @@ wait_wal_lsn(XLogRecPtr target_lsn, bool is_start_lsn, TimeLineID tli,
 
 		sleep(1);
 		if (interrupted)
-			elog(ERROR, "Interrupted during waiting for WAL archiving");
+			elog(ERROR, "Interrupted during waiting for WAL %s", in_stream_dir ? "streaming" : "archiving");
 		try_count++;
 
 		/* Inform user if WAL segment is absent in first attempt */
@@ -1420,9 +1425,10 @@ wait_wal_lsn(XLogRecPtr target_lsn, bool is_start_lsn, TimeLineID tli,
 		{
 			if (file_exists)
 				elog(timeout_elevel, "WAL segment %s was %s, "
-					 "but target LSN %X/%X could not be archived in %d seconds",
+					 "but target LSN %X/%X could not be %s in %d seconds",
 					 wal_segment, wal_delivery_str,
-					 (uint32) (target_lsn >> 32), (uint32) target_lsn, timeout);
+					 (uint32) (target_lsn >> 32), (uint32) target_lsn,
+					 wal_delivery_str, timeout);
 			/* If WAL segment doesn't exist or we wait for previous segment */
 			else
 				elog(timeout_elevel,
@@ -1707,7 +1713,7 @@ pg_stop_backup(pgBackup *backup, PGconn *pg_startbackup_conn,
 			{
 				/* Wait for segment with current stop_lsn, it is ok for it to never arrive */
 				wait_wal_lsn(stop_backup_lsn_tmp, false, backup->tli,
-							 false, true, WARNING, stream_wal);
+							 false, true, WARNING, stream_wal, backup);
 
 				/* Get the first record in segment with current stop_lsn */
 				lsn_tmp = get_first_record_lsn(xlog_path, segno, backup->tli,
@@ -1735,7 +1741,7 @@ pg_stop_backup(pgBackup *backup, PGconn *pg_startbackup_conn,
 					 * because previous record can be the contrecord.
 					 */
 					lsn_tmp = wait_wal_lsn(stop_backup_lsn_tmp, false, backup->tli,
-											true, false, ERROR, stream_wal);
+											true, false, ERROR, stream_wal, backup);
 
 					/* sanity */
 					if (!XRecOffIsValid(lsn_tmp) || XLogRecPtrIsInvalid(lsn_tmp))
@@ -1749,7 +1755,7 @@ pg_stop_backup(pgBackup *backup, PGconn *pg_startbackup_conn,
 			{
 				/* Wait for segment with current stop_lsn */
 				wait_wal_lsn(stop_backup_lsn_tmp, false, backup->tli,
-							 false, true, ERROR, stream_wal);
+							 false, true, ERROR, stream_wal, backup);
 
 				/* Get the next closest record in segment with current stop_lsn */
 				lsn_tmp = get_next_record_lsn(xlog_path, segno, backup->tli,
@@ -1878,7 +1884,7 @@ pg_stop_backup(pgBackup *backup, PGconn *pg_startbackup_conn,
 		 */
 		if (!stop_lsn_exists)
 			stop_backup_lsn = wait_wal_lsn(stop_backup_lsn_tmp, false, backup->tli,
-											false, false, ERROR, stream_wal);
+											false, false, ERROR, stream_wal, backup);
 
 		if (stream_wal)
 		{

src/catalog.c

@@ -2457,7 +2457,7 @@ write_backup_filelist(pgBackup *backup, parray *files, const char *root,
 		{
 			len += sprintf(line+len, ",\"n_headers\":\"%i\"", file->n_headers);
 			len += sprintf(line+len, ",\"hdr_crc\":\"%u\"", file->hdr_crc);
-			len += sprintf(line+len, ",\"hdr_off\":\"%li\"", file->hdr_off);
+			len += sprintf(line+len, ",\"hdr_off\":\"%llu\"", file->hdr_off);
 			len += sprintf(line+len, ",\"hdr_size\":\"%i\"", file->hdr_size);
 		}
 

src/data.c

@@ -2160,7 +2160,7 @@ get_data_file_headers(HeaderMap *hdr_map, pgFile *file, uint32 backup_version, b
 
 	if (fseek(in, file->hdr_off, SEEK_SET))
 	{
-		elog(strict ? ERROR : WARNING, "Cannot seek to position %lu in page header map \"%s\": %s",
+		elog(strict ? ERROR : WARNING, "Cannot seek to position %llu in page header map \"%s\": %s",
 			file->hdr_off, hdr_map->path, strerror(errno));
 		goto cleanup;
 	}
@@ -2177,7 +2177,7 @@ get_data_file_headers(HeaderMap *hdr_map, pgFile *file, uint32 backup_version, b
 
 	if (fread(zheaders, 1, file->hdr_size, in) != file->hdr_size)
 	{
-		elog(strict ? ERROR : WARNING, "Cannot read header file at offset: %li len: %i \"%s\": %s",
+		elog(strict ? ERROR : WARNING, "Cannot read header file at offset: %llu len: %i \"%s\": %s",
 			file->hdr_off, file->hdr_size, hdr_map->path, strerror(errno));
 		goto cleanup;
 	}
@@ -2208,7 +2208,7 @@ get_data_file_headers(HeaderMap *hdr_map, pgFile *file, uint32 backup_version, b
 	if (hdr_crc != file->hdr_crc)
 	{
 		elog(strict ? ERROR : WARNING, "Header map for file \"%s\" crc mismatch \"%s\" "
-				"offset: %lu, len: %lu, current: %u, expected: %u",
+				"offset: %llu, len: %lu, current: %u, expected: %u",
 			file->rel_path, hdr_map->path, file->hdr_off, read_len, hdr_crc, file->hdr_crc);
 		goto cleanup;
 	}
@@ -2268,7 +2268,7 @@ write_page_headers(BackupPageHeader2 *headers, pgFile *file, HeaderMap *hdr_map,
 	{
 		elog(LOG, "Creating page header map \"%s\"", map_path);
 
-		hdr_map->fp = fopen(map_path, PG_BINARY_W);
+		hdr_map->fp = fopen(map_path, "a");
 		if (hdr_map->fp == NULL)
 			elog(ERROR, "Cannot open header file \"%s\": %s",
 				 map_path, strerror(errno));
@@ -2297,7 +2297,7 @@ write_page_headers(BackupPageHeader2 *headers, pgFile *file, HeaderMap *hdr_map,
 				 file->rel_path, z_len);
 	}
 
-	elog(VERBOSE, "Writing headers for file \"%s\" offset: %li, len: %i, crc: %u",
+	elog(VERBOSE, "Writing headers for file \"%s\" offset: %llu, len: %i, crc: %u",
 			file->rel_path, file->hdr_off, z_len, file->hdr_crc);
 
 	if (fwrite(zheaders, 1, z_len, hdr_map->fp) != z_len)

src/pg_probackup.h

@@ -208,6 +208,8 @@ do { \
 		FIN_TRADITIONAL_CRC32(crc); \
 } while (0)
 
+#define pg_off_t unsigned long long
+
 
 /* Information about single file (or dir) in backup */
 typedef struct pgFile
@@ -249,8 +251,8 @@ typedef struct pgFile
 	/* Coordinates in header map */
 	int      n_headers;		/* number of blocks in the data file in backup */
 	pg_crc32 hdr_crc;		/* CRC value of header file: name_hdr */
-	off_t    hdr_off;       /* offset in header map */
-	int      hdr_size;       /* offset in header map */
+	pg_off_t hdr_off;       /* offset in header map */
+	int      hdr_size;      /* length of headers */
 } pgFile;
 
 typedef struct page_map_entry
@@ -406,11 +408,11 @@ typedef struct PGNodeInfo
 /* structure used for access to block header map */
 typedef struct HeaderMap
 {
-	char  path[MAXPGPATH];
-	char  path_tmp[MAXPGPATH]; /* used only in merge */
-	FILE  *fp;                 /* used only for writing */
-	char  *buf;	               /* buffer */
-	off_t  offset;             /* current position in fp */
+	char     path[MAXPGPATH];
+	char     path_tmp[MAXPGPATH]; /* used only in merge */
+	FILE    *fp;                  /* used only for writing */
+	char    *buf;                 /* buffer */
+	pg_off_t offset;              /* current position in fp */
 	pthread_mutex_t mutex;
 
 } HeaderMap;

src/utils/logger.c

@@ -169,6 +169,7 @@ exit_if_necessary(int elevel)
 		{
 			/* Interrupt other possible routines */
 			thread_interrupted = true;
+			interrupted = true;
 #ifdef WIN32
 			ExitThread(elevel);
 #else

tests/backup.py

@@ -2365,7 +2365,7 @@ def test_parent_choosing_2(self):
         # Clean after yourself
         self.del_test_dir(module_name, fname)
 
-    @unittest.skip("skip")
+    # @unittest.skip("skip")
     def test_backup_with_less_privileges_role(self):
         """
         check permissions correctness from documentation:
@@ -3079,3 +3079,80 @@ def test_incr_backup_filenode_map(self):
 
         # Clean after yourself
         self.del_test_dir(module_name, fname)
+
+
+
+    # @unittest.skip("skip")
+    def test_missing_wal_segment(self):
+        """"""
+        fname = self.id().split('.')[3]
+        backup_dir = os.path.join(self.tmp_path, module_name, fname, 'backup')
+        node = self.make_simple_node(
+            base_dir=os.path.join(module_name, fname, 'node'),
+            set_replication=True,
+            ptrack_enable=self.ptrack,
+            initdb_params=['--data-checksums'],
+            pg_options={'archive_timeout': '30s'})
+
+        self.init_pb(backup_dir)
+        self.add_instance(backup_dir, 'node', node)
+        self.set_archiving(backup_dir, 'node', node)
+        node.slow_start()
+
+        node.pgbench_init(scale=10)
+
+        node.safe_psql(
+            'postgres',
+            'CREATE DATABASE backupdb')
+
+        # get segments in pg_wal, sort then and remove all but the latest
+        pg_wal_dir = os.path.join(node.data_dir, 'pg_wal')
+
+        if node.major_version >= 10:
+            pg_wal_dir = os.path.join(node.data_dir, 'pg_wal')
+        else:
+            pg_wal_dir = os.path.join(node.data_dir, 'pg_xlog')
+
+        # Full backup in streaming mode
+        gdb = self.backup_node(
+            backup_dir, 'node', node, datname='backupdb',
+            options=['--stream', '--log-level-file=INFO'], gdb=True)
+
+        # break at streaming start
+        gdb.set_breakpoint('start_WAL_streaming')
+        gdb.run_until_break()
+
+        # generate some more data
+        node.pgbench_init(scale=3)
+
+        # remove redundant WAL segments in pg_wal
+        files = os.listdir(pg_wal_dir)
+        files.sort(reverse=True)
+
+        # leave first two files in list
+        del files[:2]
+        for filename in files:
+            os.remove(os.path.join(pg_wal_dir, filename))
+
+        gdb.continue_execution_until_exit()
+
+        self.assertIn(
+            'unexpected termination of replication stream: ERROR:  requested WAL segment',
+            gdb.output)
+
+        self.assertIn(
+            'has already been removed',
+            gdb.output)
+
+        self.assertIn(
+            'ERROR: Interrupted during waiting for WAL streaming',
+            gdb.output)
+
+        self.assertIn(
+            'WARNING: backup in progress, stop backup',
+            gdb.output)
+        
+        # TODO: check the same for PAGE backup
+
+        # Clean after yourself
+        self.del_test_dir(module_name, fname)

tests/helpers/ptrack_helpers.py

@@ -1757,6 +1757,7 @@ def __str__(self):
 class GDBobj(ProbackupTest):
     def __init__(self, cmd, verbose, attach=False):
         self.verbose = verbose
+        self.output = ''
 
         # Check gdb presense
         try:
@@ -1798,10 +1799,8 @@ def __init__(self, cmd, verbose, attach=False):
         )
         self.gdb_pid = self.proc.pid
 
-        # discard data from pipe,
-        # is there a way to do it a less derpy way?
         while True:
-            line = self.proc.stdout.readline()
+            line = self.get_line()
 
             if 'No such process' in line:
                 raise GdbException(line)
@@ -1811,6 +1810,11 @@ def __init__(self, cmd, verbose, attach=False):
             else:
                 break
 
+    def get_line(self):
+        line = self.proc.stdout.readline()
+        self.output += line
+        return line
+
     def kill(self):
         self.proc.kill()
         self.proc.wait()
@@ -1932,10 +1936,8 @@ def continue_execution_until_break(self, ignore_count=0):
             'Failed to continue execution until break.\n')
 
     def stopped_in_breakpoint(self):
-        output = []
         while True:
-            line = self.proc.stdout.readline()
-            output += [line]
+            line = self.get_line()
             if self.verbose:
                 print(line)
             if line.startswith('*stopped,reason="breakpoint-hit"'):
@@ -1952,7 +1954,7 @@ def _execute(self, cmd, running=True):
 
         # look for command we just send
         while True:
-            line = self.proc.stdout.readline()
+            line = self.get_line()
             if self.verbose:
                 print(repr(line))
 
@@ -1962,7 +1964,7 @@ def _execute(self, cmd, running=True):
                 break
 
         while True:
-            line = self.proc.stdout.readline()
+            line = self.get_line()
             output += [line]
             if self.verbose:
                 print(repr(line))