[PGPRO-6037] fix catchup timeline history checking (#462)

* [PGPRO-6037] fix catchup timeline history checking

Author: kulaginm

src/catchup.c

@@ -203,6 +203,8 @@ catchup_preflight_checks(PGNodeInfo *source_node_info, PGconn *source_conn,
 
 		/* fill dest_redo.lsn and dest_redo.tli */
 		get_redo(dest_pgdata, FIO_LOCAL_HOST, &dest_redo);
+		elog(VERBOSE, "source.tli = %X, dest_redo.lsn = %X/%X, dest_redo.tli = %X",
+			current.tli, (uint32) (dest_redo.lsn >> 32), (uint32) dest_redo.lsn, dest_redo.tli);
 
 		if (current.tli != 1)
 		{
@@ -285,11 +287,12 @@ catchup_check_tablespaces_existance_in_tbsmapping(PGconn *conn)
 static parray*
 catchup_get_tli_history(ConnectionOptions *conn_opt, TimeLineID tli)
 {
-	PGresult     *res;
-	PGconn	     *conn;
-	char         *history;
-	char          query[128];
-	parray	     *result = NULL;
+	PGresult             *res;
+	PGconn	             *conn;
+	char                 *history;
+	char                  query[128];
+	parray	             *result = NULL;
+	TimeLineHistoryEntry *entry = NULL;
 
 	snprintf(query, sizeof(query), "TIMELINE_HISTORY %u", tli);
 
@@ -336,6 +339,12 @@ catchup_get_tli_history(ConnectionOptions *conn_opt, TimeLineID tli)
 	pg_free(history);
 	PQclear(res);
 
+	/* append last timeline entry (as read_timeline_history() do) */
+	entry = pgut_new(TimeLineHistoryEntry);
+	entry->tli = tli;
+	entry->end = InvalidXLogRecPtr;
+	parray_insert(result, 0, entry);
+
 	return result;
 }
 

src/restore.c

@@ -1821,11 +1821,15 @@ satisfy_timeline(const parray *timelines, TimeLineID tli, XLogRecPtr lsn)
 {
 	int			i;
 
+	elog(VERBOSE, "satisfy_timeline() checking: tli = %X, lsn = %X/%X",
+		tli, (uint32) (lsn >> 32), (uint32) lsn);
 	for (i = 0; i < parray_num(timelines); i++)
 	{
 		TimeLineHistoryEntry *timeline;
 
 		timeline = (TimeLineHistoryEntry *) parray_get(timelines, i);
+		elog(VERBOSE, "satisfy_timeline() check %i entry: timeline->tli = %X, timeline->end = %X/%X",
+			i, timeline->tli, (uint32) (timeline->end >> 32), (uint32) timeline->end);
 		if (tli == timeline->tli &&
 			(XLogRecPtrIsInvalid(timeline->end) ||
 			 lsn <= timeline->end))

src/stream.c

@@ -615,6 +615,8 @@ parse_tli_history_buffer(char *history, TimeLineID tli)
 			if (!result)
 				result = parray_new();
 			parray_append(result, entry);
+			elog(VERBOSE, "parse_tli_history_buffer() found entry: tli = %X, end = %X/%X",
+				tli, switchpoint_hi, switchpoint_lo);
 
 			/* we ignore the remainder of each line */
 		}

tests/catchup.py

@@ -292,7 +292,7 @@ def test_tli_delta_catchup(self):
         src_pg.safe_psql("postgres", "CREATE TABLE ultimate_question AS SELECT 42 AS answer")
         src_query_result = src_pg.safe_psql("postgres", "SELECT * FROM ultimate_question")
 
-        # do catchup
+        # do catchup (src_tli = 2, dst_tli = 1)
         self.catchup_node(
             backup_mode = 'DELTA',
             source_pgdata = src_pg.data_dir,
@@ -310,15 +310,25 @@ def test_tli_delta_catchup(self):
         dst_options = {}
         dst_options['port'] = str(dst_pg.port)
         self.set_auto_conf(dst_pg, dst_options)
-        dst_pg.slow_start()
+        self.set_replica(master = src_pg, replica = dst_pg)
+        dst_pg.slow_start(replica = True)
 
         # 2nd check: run verification query
         dst_query_result = dst_pg.safe_psql("postgres", "SELECT * FROM ultimate_question")
         self.assertEqual(src_query_result, dst_query_result, 'Different answer from copy')
 
+        dst_pg.stop()
+
+        # do catchup (src_tli = 2, dst_tli = 2)
+        self.catchup_node(
+            backup_mode = 'DELTA',
+            source_pgdata = src_pg.data_dir,
+            destination_node = dst_pg,
+            options = ['-d', 'postgres', '-p', str(src_pg.port), '--stream']
+            )
+
         # Cleanup
         src_pg.stop()
-        dst_pg.stop()
         self.del_test_dir(module_name, self.fname)
 
     def test_tli_ptrack_catchup(self):
@@ -365,7 +375,7 @@ def test_tli_ptrack_catchup(self):
         src_pg.safe_psql("postgres", "CREATE TABLE ultimate_question AS SELECT 42 AS answer")
         src_query_result = src_pg.safe_psql("postgres", "SELECT * FROM ultimate_question")
 
-        # do catchup
+        # do catchup (src_tli = 2, dst_tli = 1)
         self.catchup_node(
             backup_mode = 'PTRACK',
             source_pgdata = src_pg.data_dir,
@@ -383,15 +393,25 @@ def test_tli_ptrack_catchup(self):
         dst_options = {}
         dst_options['port'] = str(dst_pg.port)
         self.set_auto_conf(dst_pg, dst_options)
-        dst_pg.slow_start()
+        self.set_replica(master = src_pg, replica = dst_pg)
+        dst_pg.slow_start(replica = True)
 
         # 2nd check: run verification query
         dst_query_result = dst_pg.safe_psql("postgres", "SELECT * FROM ultimate_question")
         self.assertEqual(src_query_result, dst_query_result, 'Different answer from copy')
 
+        dst_pg.stop()
+
+        # do catchup (src_tli = 2, dst_tli = 2)
+        self.catchup_node(
+            backup_mode = 'PTRACK',
+            source_pgdata = src_pg.data_dir,
+            destination_node = dst_pg,
+            options = ['-d', 'postgres', '-p', str(src_pg.port), '--stream']
+            )
+
         # Cleanup
         src_pg.stop()
-        dst_pg.stop()
         self.del_test_dir(module_name, self.fname)
 
 #########################################