Use wait_replica_wal_lsn() in pg_start_backup() and pg_stop_backup()

Author: Arthur Zakirov

src/backup.c

@@ -640,6 +640,18 @@ pg_start_backup(const char *label, bool smooth, pgBackup *backup)
 
 	PQclear(res);
 
+	/*
+	 * Switch to a new WAL segment. It is necessary to get archived WAL
+	 * segment, which includes start LSN of current backup.
+	 *
+	 * Do not switch for standby node and if backup is stream.
+	 */
+	if (!stream_wal)
+		pg_switch_wal(conn);
+	/* Wait for start_lsn to be received by replica */
+	if (from_replica)
+		wait_replica_wal_lsn(backup->start_lsn, true);
+
 	if (!stream_wal)
 		/*
 		 * Do not wait start_lsn for stream backup.
@@ -658,16 +670,15 @@ pg_switch_wal(PGconn *conn)
 	PGresult   *res;
 
 	/* Remove annoying NOTICE messages generated by backend */
-	res = pgut_execute(conn, "SET client_min_messages = warning;", 0,
-					   NULL);
+	res = pgut_execute(conn, "SET client_min_messages = warning;", 0, NULL);
 	PQclear(res);
 
 	if (server_version >= 100000)
 		res = pgut_execute(conn, "SELECT * FROM pg_switch_wal()", 0,
-						   NULL);
+											  NULL);
 	else
 		res = pgut_execute(conn, "SELECT * FROM pg_switch_xlog()", 0,
-						   NULL);
+											  NULL);
 
 	PQclear(res);
 }
@@ -908,7 +919,7 @@ wait_wal_lsn(XLogRecPtr lsn)
 }
 
 /*
- * Wait for target 'lsn' on replica instance.
+ * Wait for target 'lsn' on replica instance from master.
  */
 static void
 wait_replica_wal_lsn(XLogRecPtr lsn, bool is_start_backup)
@@ -973,6 +984,7 @@ wait_replica_wal_lsn(XLogRecPtr lsn, bool is_start_backup)
 static void
 pg_stop_backup(pgBackup *backup)
 {
+	PGconn	   *conn;
 	PGresult   *res;
 	uint32		xlogid;
 	uint32		xrecoff;
@@ -990,8 +1002,11 @@ pg_stop_backup(pgBackup *backup)
 	if (!backup_in_progress)
 		elog(FATAL, "backup is not in progress");
 
+	/* For replica we call pg_stop_backup() on master */
+	conn = (from_replica) ? master_conn : backup_conn;
+
 	/* Remove annoying NOTICE messages generated by backend */
-	res = pgut_execute(backup_conn, "SET client_min_messages = warning;",
+	res = pgut_execute(conn, "SET client_min_messages = warning;",
 					   0, NULL);
 	PQclear(res);
 
@@ -1005,69 +1020,16 @@ pg_stop_backup(pgBackup *backup)
 		backup_id = base36enc(backup->start_time);
 
 		if (!from_replica)
-		{
 			snprintf(name, lengthof(name), "pg_probackup, backup_id %s",
 					 backup_id);
-			params[0] = name;
-
-			res = pgut_execute(backup_conn, "SELECT pg_create_restore_point($1)",
-							   1, params);
-			PQclear(res);
-		}
 		else
-		{
-			uint32		try_count = 0;
-
 			snprintf(name, lengthof(name), "pg_probackup, backup_id %s. Replica Backup",
 					 backup_id);
-			params[0] = name;
+		params[0] = name;
 
-			res = pgut_execute(master_conn, "SELECT pg_create_restore_point($1)",
-							   1, params);
-			/* Extract timeline and LSN from result */
-			XLogDataFromLSN(PQgetvalue(res, 0, 0), &xlogid, &xrecoff);
-			/* Calculate LSN */
-			restore_lsn = (XLogRecPtr) ((uint64) xlogid << 32) | xrecoff;
-			PQclear(res);
-
-			/* Switch WAL on master to retreive restore_lsn */
-			pg_switch_wal(master_conn);
-
-			/* Wait for restore_lsn from master */
-			while (true)
-			{
-				XLogRecPtr	min_recovery_lsn;
-
-				res = pgut_execute(backup_conn, "SELECT min_recovery_end_location from pg_control_recovery()",
-								   0, NULL);
-				/* Extract timeline and LSN from result */
-				XLogDataFromLSN(PQgetvalue(res, 0, 0), &xlogid, &xrecoff);
-				/* Calculate LSN */
-				min_recovery_lsn = (XLogRecPtr) ((uint64) xlogid << 32) | xrecoff;
-				PQclear(res);
-
-				/* restore_lsn was streamed and applied to the replica */
-				if (min_recovery_lsn >= restore_lsn)
-					break;
-
-				sleep(1);
-				if (interrupted)
-					elog(ERROR, "Interrupted during waiting for restore point LSN");
-				try_count++;
-
-				/* Inform user if restore_lsn is absent in first attempt */
-				if (try_count == 1)
-					elog(INFO, "Wait for restore point LSN %X/%X to be streamed "
-						 "to replica",
-						 (uint32) (restore_lsn >> 32), (uint32) restore_lsn);
-
-				if (replica_timeout > 0 && try_count > replica_timeout)
-					elog(ERROR, "Restore point LSN %X/%X could not be "
-						 "streamed to replica in %d seconds",
-						 (uint32) (restore_lsn >> 32), (uint32) restore_lsn,
-						 replica_timeout);
-			}
-		}
+		res = pgut_execute(conn, "SELECT pg_create_restore_point($1)",
+						   1, params);
+		PQclear(res);
 
 		pfree(backup_id);
 	}
@@ -1084,13 +1046,13 @@ pg_stop_backup(pgBackup *backup)
 		 * pg_stop_backup(false) copy of the backup label and tablespace map
 		 * so they can be written to disk by the caller.
 		 */
-		sent = pgut_send(backup_conn,
+		sent = pgut_send(conn,
 						   "SELECT *, txid_snapshot_xmax(txid_current_snapshot()),"
 						   " current_timestamp(0)::timestamp"
 						   " FROM pg_stop_backup(false)",
 						   0, NULL, WARNING);
 	else
-		sent = pgut_send(backup_conn,
+		sent = pgut_send(conn,
 						   "SELECT *, txid_snapshot_xmax(txid_current_snapshot()),"
 						   " current_timestamp(0)::timestamp"
 						   " FROM pg_stop_backup()",
@@ -1108,30 +1070,30 @@ pg_stop_backup(pgBackup *backup)
 
 	while (1)
 	{
-		if (!PQconsumeInput(backup_conn) || PQisBusy(backup_conn))
+		if (!PQconsumeInput(conn) || PQisBusy(conn))
 		{
-				pg_stop_backup_timeout++;
-				sleep(1);
+			pg_stop_backup_timeout++;
+			sleep(1);
 
-				if (interrupted)
-				{
-					pgut_cancel(backup_conn);
-					elog(ERROR, "interrupted during waiting for pg_stop_backup");
-				}
-				/*
-				 * If postgres haven't answered in PG_STOP_BACKUP_TIMEOUT seconds,
-				 * send an interrupt.
-				 */
-				if (pg_stop_backup_timeout > PG_STOP_BACKUP_TIMEOUT)
-				{
-					pgut_cancel(backup_conn);
-					elog(ERROR, "pg_stop_backup doesn't answer in %d seconds, cancel it",
-						 PG_STOP_BACKUP_TIMEOUT);
-				}
+			if (interrupted)
+			{
+				pgut_cancel(conn);
+				elog(ERROR, "interrupted during waiting for pg_stop_backup");
+			}
+			/*
+			 * If postgres haven't answered in PG_STOP_BACKUP_TIMEOUT seconds,
+			 * send an interrupt.
+			 */
+			if (pg_stop_backup_timeout > PG_STOP_BACKUP_TIMEOUT)
+			{
+				pgut_cancel(conn);
+				elog(ERROR, "pg_stop_backup doesn't answer in %d seconds, cancel it",
+					 PG_STOP_BACKUP_TIMEOUT);
+			}
 		}
 		else
 		{
-			res = PQgetResult(backup_conn);
+			res = PQgetResult(conn);
 			break;
 		}
 	}
@@ -1228,22 +1190,22 @@ pg_stop_backup(pgBackup *backup)
 		if (sscanf(PQgetvalue(res, 0, 3), XID_FMT, &recovery_xid) != 1)
 			elog(ERROR,
 				 "result of txid_snapshot_xmax() is invalid: %s",
-				 PQerrorMessage(backup_conn));
+				 PQerrorMessage(conn));
 		if (!parse_time(PQgetvalue(res, 0, 4), &recovery_time))
 			elog(ERROR,
 				 "result of current_timestamp is invalid: %s",
-				 PQerrorMessage(backup_conn));
+				 PQerrorMessage(conn));
 	}
 	else
 	{
 		if (sscanf(PQgetvalue(res, 0, 1), XID_FMT, &recovery_xid) != 1)
 			elog(ERROR,
 				 "result of txid_snapshot_xmax() is invalid: %s",
-				 PQerrorMessage(backup_conn));
+				 PQerrorMessage(conn));
 		if (!parse_time(PQgetvalue(res, 0, 2), &recovery_time))
 			elog(ERROR,
 				 "result of current_timestamp is invalid: %s",
-				 PQerrorMessage(backup_conn));
+				 PQerrorMessage(conn));
 	}
 
 	PQclear(res);
@@ -1258,6 +1220,9 @@ pg_stop_backup(pgBackup *backup)
 		char	   *xlog_path,
 					stream_xlog_path[MAXPGPATH];
 
+		/* Wait for stop_lsn to be received by replica */
+		if (from_replica)
+			wait_replica_wal_lsn(stop_backup_lsn, false);
 		/*
 		 * Wait for stop_lsn to be archived or streamed.
 		 * We wait for stop_lsn in stream mode just in case.