[CF 5627] Fix oldest LSN calculation for WAL segments removal when slots are advancing during checkpoint

This branch was automatically generated by a robot using patches from an
email thread registered at:

https://commitfest.postgresql.org/patch/5627

The branch will be overwritten each time a new patch version is posted to
the thread, and also periodically to check for bitrot caused by changes
on the master branch.

Patch(es): https://www.postgresql.org/message-id/CALDaNm1u=QV1a7w48BWgkg6GnK20eE_y+raL43D46_R9Wnt8wg@mail.gmail.com
Author(s): Vitaly Davydov

Author: Commitfest Bot

src/backend/access/transam/xlog.c

@@ -6836,6 +6836,7 @@ ShutdownXLOG(int code, Datum arg)
 static void
 LogCheckpointStart(int flags, bool restartpoint)
 {
+	#if 0
 	if (restartpoint)
 		ereport(LOG,
 		/* translator: the placeholders show checkpoint options */
@@ -6860,6 +6861,7 @@ LogCheckpointStart(int flags, bool restartpoint)
 						(flags & CHECKPOINT_CAUSE_XLOG) ? " wal" : "",
 						(flags & CHECKPOINT_CAUSE_TIME) ? " time" : "",
 						(flags & CHECKPOINT_FLUSH_ALL) ? " flush-all" : "")));
+	#endif
 }
 
 /*
@@ -6909,6 +6911,7 @@ LogCheckpointEnd(bool restartpoint)
 			CheckpointStats.ckpt_sync_rels;
 	average_msecs = (long) ((average_sync_time + 999) / 1000);
 
+	#if 0
 	/*
 	 * ControlFileLock is not required to see ControlFile->checkPoint and
 	 * ->checkPointCopy here as we are the only updator of those variables at
@@ -6962,6 +6965,7 @@ LogCheckpointEnd(bool restartpoint)
 						(int) (CheckPointDistanceEstimate / 1024.0),
 						LSN_FORMAT_ARGS(ControlFile->checkPoint),
 						LSN_FORMAT_ARGS(ControlFile->checkPointCopy.redo))));
+	#endif
 }
 
 /*
@@ -7160,13 +7164,15 @@ CreateCheckPoint(int flags)
 	if ((flags & (CHECKPOINT_IS_SHUTDOWN | CHECKPOINT_END_OF_RECOVERY |
 				  CHECKPOINT_FORCE)) == 0)
 	{
+		#if 0
 		if (last_important_lsn == ControlFile->checkPoint)
 		{
 			END_CRIT_SECTION();
 			ereport(DEBUG1,
 					(errmsg_internal("checkpoint skipped because system is idle")));
 			return false;
 		}
+		#endif
 	}
 
 	/*

src/backend/access/transam/xlogrecovery.c

@@ -1918,6 +1918,8 @@ ApplyWalRecord(XLogReaderState *xlogreader, XLogRecord *record, TimeLineID *repl
 {
 	ErrorContextCallback errcallback;
 	bool		switchedTLI = false;
+	static int count = 0;
+	count++;
 
 	/* Setup error traceback support for ereport() */
 	errcallback.callback = rm_redo_error_callback;
@@ -2009,7 +2011,13 @@ ApplyWalRecord(XLogReaderState *xlogreader, XLogRecord *record, TimeLineID *repl
 
 	/* Pop the error context stack */
 	error_context_stack = errcallback.previous;
-
+	pg_usleep(1000L);
+	if (count > 1000)
+	pg_usleep(1000L);
+		
+	elog(LOG, "Applied wal record end rec %X/%X, read rec %X/%X",
+			LSN_FORMAT_ARGS(xlogreader->EndRecPtr),
+			LSN_FORMAT_ARGS(xlogreader->ReadRecPtr));
 	/*
 	 * Update lastReplayedEndRecPtr after this record has been successfully
 	 * replayed.
@@ -3640,6 +3648,7 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
 					if (StandbyMode && CheckForStandbyTrigger())
 					{
 						XLogShutdownWalRcv();
+						elog(LOG, "returning fail vignesh1");
 						return XLREAD_FAIL;
 					}
 
@@ -3648,7 +3657,10 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
 					 * and pg_wal.
 					 */
 					if (!StandbyMode)
+					{
+						elog(LOG, "returning fail vignesh2");
 						return XLREAD_FAIL;
+					}
 
 					/*
 					 * Move to XLOG_FROM_STREAM state, and set to start a
@@ -3870,6 +3882,8 @@ WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess,
 						}
 						curFileTLI = tli;
 						SetInstallXLogFileSegmentActive();
+						elog(LOG, "vignesh request to start streaming from primary at %X/%X on timeline %u",
+							 LSN_FORMAT_ARGS(ptr), curFileTLI);
 						RequestXLogStreaming(tli, ptr, PrimaryConnInfo,
 											 PrimarySlotName,
 											 wal_receiver_create_temp_slot);

src/backend/postmaster/checkpointer.c

@@ -345,7 +345,7 @@ CheckpointerMain(const void *startup_data, size_t startup_data_len)
 	 */
 	for (;;)
 	{
-		bool		do_checkpoint = false;
+		bool		do_checkpoint = true;
 		int			flags = 0;
 		pg_time_t	now;
 		int			elapsed_secs;
@@ -573,11 +573,14 @@ CheckpointerMain(const void *startup_data, size_t startup_data_len)
 				continue;		/* no sleep for us ... */
 			cur_timeout = Min(cur_timeout, XLogArchiveTimeout - elapsed_secs);
 		}
+		pg_usleep(1000L);
 
+		#if 0
 		(void) WaitLatch(MyLatch,
 						 WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
 						 cur_timeout * 1000L /* convert to ms */ ,
 						 WAIT_EVENT_CHECKPOINTER_MAIN);
+		#endif
 	}
 
 	/*

src/backend/replication/walreceiver.c

@@ -396,6 +396,9 @@ WalReceiverMain(const void *startup_data, size_t startup_data_len)
 
 			/* Initialize LogstreamResult and buffers for processing messages */
 			LogstreamResult.Write = LogstreamResult.Flush = GetXLogReplayRecPtr(NULL);
+			elog(LOG, "WalReceiverMain LogstreamResult.Flush initialized to %X/%X",
+				 LSN_FORMAT_ARGS(LogstreamResult.Flush));
+
 			initStringInfo(&reply_message);
 
 			/* Initialize nap wakeup times. */
@@ -848,6 +851,8 @@ XLogWalRcvProcessMsg(unsigned char type, char *buf, Size len, TimeLineID tli)
 				buf += hdrlen;
 				len -= hdrlen;
 				XLogWalRcvWrite(buf, len, dataStart, tli);
+				elog(LOG, "XLogWalRcvProcessMsg: wrote %zu bytes of WAL at %X/%X",
+					 len, LSN_FORMAT_ARGS(dataStart));
 				break;
 			}
 		case 'k':				/* Keepalive */
@@ -960,6 +965,8 @@ XLogWalRcvWrite(char *buf, Size nbytes, XLogRecPtr recptr, TimeLineID tli)
 		buf += byteswritten;
 
 		LogstreamResult.Write = recptr;
+		elog(LOG, "XLogWalRcvFlush LogstreamResult.Write set to %X/%X",
+			 LSN_FORMAT_ARGS(LogstreamResult.Write));
 	}
 
 	/* Update shared-memory status */
@@ -994,6 +1001,9 @@ XLogWalRcvFlush(bool dying, TimeLineID tli)
 
 		LogstreamResult.Flush = LogstreamResult.Write;
 
+		elog(LOG, "XLogWalRcvFlush LogstreamResult.Flush initialized to %X/%X",
+			 LSN_FORMAT_ARGS(LogstreamResult.Flush));
+
 		/* Update shared-memory status */
 		SpinLockAcquire(&walrcv->mutex);
 		if (walrcv->flushedUpto < LogstreamResult.Flush)
@@ -1138,7 +1148,7 @@ XLogWalRcvSendReply(bool force, bool requestReply)
 	pq_sendbyte(&reply_message, requestReply ? 1 : 0);
 
 	/* Send it */
-	elog(DEBUG2, "sending write %X/%X flush %X/%X apply %X/%X%s",
+	elog(LOG, "sending write %X/%X flush %X/%X apply %X/%X%s",
 		 LSN_FORMAT_ARGS(writePtr),
 		 LSN_FORMAT_ARGS(flushPtr),
 		 LSN_FORMAT_ARGS(applyPtr),

src/backend/replication/walsender.c

@@ -2377,7 +2377,14 @@ PhysicalConfirmReceivedLocation(XLogRecPtr lsn)
 	if (slot->data.restart_lsn != lsn)
 	{
 		changed = true;
+
+		if (lsn < slot->data.restart_lsn)
+			elog(LOG, "crash scenario - slot %s, cannot confirm a restart LSN (%X/%X) that is older than the current one (%X/%X)",
+				 NameStr(slot->data.name), LSN_FORMAT_ARGS(lsn), LSN_FORMAT_ARGS(slot->data.restart_lsn));
+
 		slot->data.restart_lsn = lsn;
+		elog(LOG, "PhysicalConfirmReceivedLocation replication slot \"%s\" set restart_lsn to %X/%X",
+			 NameStr(slot->data.name), LSN_FORMAT_ARGS(slot->data.restart_lsn));
 	}
 	SpinLockRelease(&slot->mutex);