test.patch

application/octet-stream

Download

Patch

Same data as JSON: GET /api/v1/attachments/:id/patch the parsed metadata as JSON — format, series position, per-file stats; never the diff bytes. API reference →

Format: unified

File	+	−
src/backend/postmaster/bgwriter.c	1	1
src/backend/replication/logical/logical.c	11	0
src/backend/replication/logical/worker.c	7	0
src/backend/replication/slot.c	5	0
src/backend/replication/walsender.c	1	0

diff --git a/src/backend/postmaster/bgwriter.c b/src/backend/postmaster/bgwriter.c
index 91e6f6ea18..627a34b9d5 100644
--- a/src/backend/postmaster/bgwriter.c
+++ b/src/backend/postmaster/bgwriter.c
@@ -70,7 +70,7 @@ int			BgWriterDelay = 200;
  * Interval in which standby snapshots are logged into the WAL stream, in
  * milliseconds.
  */
-#define LOG_SNAPSHOT_INTERVAL_MS 15000
+#define LOG_SNAPSHOT_INTERVAL_MS 15000000
 
 /*
  * LSN and timestamp at which we last issued a LogStandbySnapshot(), to avoid
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 625a7f4273..650af046b8 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -1735,6 +1735,8 @@ LogicalIncreaseRestartDecodingForSlot(XLogRecPtr current_lsn, XLogRecPtr restart
 void
 LogicalConfirmReceivedLocation(XLogRecPtr lsn)
 {
+	bool backward = false;
+
 	Assert(lsn != InvalidXLogRecPtr);
 
 	/* Do an unlocked check for candidate_lsn first. */
@@ -1776,6 +1778,9 @@ LogicalConfirmReceivedLocation(XLogRecPtr lsn)
 		{
 			Assert(MyReplicationSlot->candidate_restart_lsn != InvalidXLogRecPtr);
 
+			if (MyReplicationSlot->data.restart_lsn > MyReplicationSlot->candidate_restart_lsn)
+				backward = true;
+
 			MyReplicationSlot->data.restart_lsn = MyReplicationSlot->candidate_restart_lsn;
 			MyReplicationSlot->candidate_restart_lsn = InvalidXLogRecPtr;
 			MyReplicationSlot->candidate_restart_valid = InvalidXLogRecPtr;
@@ -1814,6 +1819,12 @@ LogicalConfirmReceivedLocation(XLogRecPtr lsn)
 		MyReplicationSlot->data.confirmed_flush = lsn;
 		SpinLockRelease(&MyReplicationSlot->mutex);
 	}
+
+	if (backward)
+	{
+		pg_usleep(5 * 1000000L);
+		elog(ERROR, "XXX restart_lsn went backward, will restart (and fail)");
+	}
 }
 
 /*
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index e48a3f589a..daabc0625a 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -872,6 +872,13 @@ apply_handle_commit(StringInfo s)
 	/* Process any tables that are being synchronized in parallel. */
 	process_syncing_tables(commit_data.end_lsn);
 
+	if (apply_error_callback_arg.remote_xid == 744)
+	{
+		elog(LOG, "XXX committed xid 744 origin lsn %X/%X, sleep 60s",
+			 LSN_FORMAT_ARGS(replorigin_session_get_progress(false)));
+		pg_usleep(60 * 1000000);
+	}
+
 	pgstat_report_activity(STATE_IDLE, NULL);
 	reset_apply_error_context_info();
 }
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 899acfd912..bd29969f92 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -528,6 +528,11 @@ retry:
 	/* We made this slot active, so it's ours now. */
 	MyReplicationSlot = s;
 
+	elog(LOG, "XXX acquired slot restart_lsn %X/%X candidate_restart_lsn %X/%X candidate_restart_valid %X/%X",
+		 LSN_FORMAT_ARGS(s->data.restart_lsn),
+		 LSN_FORMAT_ARGS(s->candidate_restart_lsn),
+		 LSN_FORMAT_ARGS(s->candidate_restart_valid));
+
 	/*
 	 * The call to pgstat_acquire_replslot() protects against stats for a
 	 * different slot, from before a restart or such, being present during
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index c11bb3716f..35002fe469 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -3043,6 +3043,7 @@ XLogSendLogical(void)
 	 */
 	WalSndCaughtUp = false;
 
+	pg_usleep(400* 1000);
 	record = XLogReadRecord(logical_decoding_ctx->reader, &errm);
 
 	/* xlog record was invalid */