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 */