Date: Thu, 26 Aug 2021 10:50:52 -0400
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
To: Kyotaro Horiguchi <horikyota.ntt@gmail.com>
Cc: bossartn@amazon.com, robertmhaas@gmail.com,
	pgsql-hackers@lists.postgresql.org, mengjuan.cmj@alibaba-inc.com,
	Jakub.Wartak@tomtom.com
Subject: Re: prevent immature WAL streaming
Message-ID: <202108261450.n33q3vuryu4b@alvherre.pgsql>
MIME-Version: 1.0
Content-Type: multipart/mixed; boundary="agszmhxcdeupegfm"
Content-Disposition: inline
Content-Transfer-Encoding: 8bit
In-Reply-To: <20210826.174834.1618367955913954817.horikyota.ntt@gmail.com>
Archived-At: 
 <https://www.postgresql.org/message-id/202108261450.n33q3vuryu4b%40alvherre.pgsql>
Precedence: bulk


--agszmhxcdeupegfm
Content-Type: text/plain; charset=utf-8
Content-Disposition: inline
Content-Transfer-Encoding: 8bit

So I'm again distracted by something else, so here's what will have to
pass for v3 for the time being.

-- 
Álvaro Herrera           39°49'30"S 73°17'W  —  https://www.EnterpriseDB.com/

--agszmhxcdeupegfm
Content-Type: text/x-diff; charset=utf-8
Content-Disposition: attachment;
	filename="v3-replication-segment-boundaries.patch"

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 24165ab03e..43495b8260 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -731,8 +731,10 @@ typedef struct XLogCtlData
 	 */
 	XLogSegNo	lastNotifiedSeg;
 	XLogSegNo	earliestSegBoundary;
+	XLogRecPtr	earliestSegBoundaryStartPtr;
 	XLogRecPtr	earliestSegBoundaryEndPtr;
 	XLogSegNo	latestSegBoundary;
+	XLogRecPtr	latestSegBoundaryStartPtr;
 	XLogRecPtr	latestSegBoundaryEndPtr;
 
 	slock_t		segtrack_lck;	/* locks shared variables shown above */
@@ -932,7 +934,7 @@ static void RemoveXlogFile(const char *segname, XLogSegNo recycleSegNo,
 						   XLogSegNo *endlogSegNo);
 static void UpdateLastRemovedPtr(char *filename);
 static void ValidateXLOGDirectoryStructure(void);
-static void RegisterSegmentBoundary(XLogSegNo seg, XLogRecPtr pos);
+static void RegisterSegmentBoundary(XLogSegNo seg, XLogRecPtr startpos, XLogRecPtr endpos);
 static void CleanupBackupHistory(void);
 static void UpdateMinRecoveryPoint(XLogRecPtr lsn, bool force);
 static XLogRecord *ReadRecord(XLogReaderState *xlogreader,
@@ -1022,6 +1024,8 @@ XLogInsertRecord(XLogRecData *rdata,
 							   info == XLOG_SWITCH);
 	XLogRecPtr	StartPos;
 	XLogRecPtr	EndPos;
+	XLogSegNo	StartSeg;
+	XLogSegNo	EndSeg;
 	bool		prevDoPageWrites = doPageWrites;
 
 	/* we assume that all of the record header is in the first chunk */
@@ -1157,6 +1161,31 @@ XLogInsertRecord(XLogRecData *rdata,
 		 */
 	}
 
+	/*
+	 * Before releasing our WAL insertion lock, register that we crossed the
+	 * segment boundary if that occurred.  We need to do it with the lock held
+	 * for GetSafeFlushRecPtr's sake: otherwise it could see the WAL flush
+	 * point advance but not see the registration, which would lead it to
+	 * wrongly conclude that our flush point is safe to use.
+	 */
+	if (StartPos / XLOG_BLCKSZ != EndPos / XLOG_BLCKSZ)
+	{
+		XLByteToSeg(StartPos, StartSeg, wal_segment_size);
+		XLByteToSeg(EndPos, EndSeg, wal_segment_size);
+
+		/*
+		 * Register our crossing the segment boundary if that occurred.
+		 *
+		 * Note that we did not use XLByteToPrevSeg() for determining the
+		 * ending segment.  This is so that a record that fits perfectly into
+		 * the end of the segment causes said segment to get marked ready for
+		 * archival immediately.
+		 */
+		if (StartSeg != EndSeg &&
+			(XLogArchivingActive() || XLogStandbyInfoActive()))
+			RegisterSegmentBoundary(EndSeg, StartPos, EndPos);
+	}
+
 	/*
 	 * Done! Let others know that we're finished.
 	 */
@@ -1168,27 +1197,10 @@ XLogInsertRecord(XLogRecData *rdata,
 
 	/*
 	 * If we crossed page boundary, update LogwrtRqst.Write; if we crossed
-	 * segment boundary, register that and wake up walwriter.
+	 * segment boundary, wake up walwriter.
 	 */
 	if (StartPos / XLOG_BLCKSZ != EndPos / XLOG_BLCKSZ)
 	{
-		XLogSegNo	StartSeg;
-		XLogSegNo	EndSeg;
-
-		XLByteToSeg(StartPos, StartSeg, wal_segment_size);
-		XLByteToSeg(EndPos, EndSeg, wal_segment_size);
-
-		/*
-		 * Register our crossing the segment boundary if that occurred.
-		 *
-		 * Note that we did not use XLByteToPrevSeg() for determining the
-		 * ending segment.  This is so that a record that fits perfectly into
-		 * the end of the segment causes the latter to get marked ready for
-		 * archival immediately.
-		 */
-		if (StartSeg != EndSeg && XLogArchivingActive())
-			RegisterSegmentBoundary(EndSeg, EndPos);
-
 		/*
 		 * Advance LogwrtRqst.Write so that it includes new block(s).
 		 *
@@ -2649,7 +2661,7 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible)
 
 				LogwrtResult.Flush = LogwrtResult.Write;	/* end of page */
 
-				if (XLogArchivingActive())
+				if (XLogArchivingActive() || XLogStandbyInfoActive())
 					NotifySegmentsReadyForArchive(LogwrtResult.Flush);
 
 				XLogCtl->lastSegSwitchTime = (pg_time_t) time(NULL);
@@ -2739,7 +2751,7 @@ XLogWrite(XLogwrtRqst WriteRqst, bool flexible)
 		SpinLockRelease(&XLogCtl->info_lck);
 	}
 
-	if (XLogArchivingActive())
+	if (XLogArchivingActive() || XLogStandbyInfoActive())
 		NotifySegmentsReadyForArchive(LogwrtResult.Flush);
 }
 
@@ -4398,7 +4410,7 @@ ValidateXLOGDirectoryStructure(void)
  * to delay until the end segment is known flushed.
  */
 static void
-RegisterSegmentBoundary(XLogSegNo seg, XLogRecPtr endpos)
+RegisterSegmentBoundary(XLogSegNo seg, XLogRecPtr startpos, XLogRecPtr endpos)
 {
 	XLogSegNo	segno PG_USED_FOR_ASSERTS_ONLY;
 
@@ -4415,6 +4427,7 @@ RegisterSegmentBoundary(XLogSegNo seg, XLogRecPtr endpos)
 	if (XLogCtl->earliestSegBoundary == MaxXLogSegNo)
 	{
 		XLogCtl->earliestSegBoundary = seg;
+		XLogCtl->earliestSegBoundaryStartPtr = startpos;
 		XLogCtl->earliestSegBoundaryEndPtr = endpos;
 	}
 	else if (seg > XLogCtl->earliestSegBoundary &&
@@ -4422,6 +4435,7 @@ RegisterSegmentBoundary(XLogSegNo seg, XLogRecPtr endpos)
 			  seg > XLogCtl->latestSegBoundary))
 	{
 		XLogCtl->latestSegBoundary = seg;
+		XLogCtl->latestSegBoundaryStartPtr = startpos;
 		XLogCtl->latestSegBoundaryEndPtr = endpos;
 	}
 
@@ -4438,10 +4452,8 @@ void
 NotifySegmentsReadyForArchive(XLogRecPtr flushRecPtr)
 {
 	XLogSegNo	latest_boundary_seg;
-	XLogSegNo	last_notified;
 	XLogSegNo	flushed_seg;
 	XLogSegNo	seg;
-	bool		keep_latest;
 
 	XLByteToSeg(flushRecPtr, flushed_seg, wal_segment_size);
 
@@ -4451,13 +4463,17 @@ NotifySegmentsReadyForArchive(XLogRecPtr flushRecPtr)
 		XLogCtl->latestSegBoundaryEndPtr <= flushRecPtr)
 	{
 		latest_boundary_seg = XLogCtl->latestSegBoundary;
-		keep_latest = false;
+		XLogCtl->earliestSegBoundary = MaxXLogSegNo;
+		XLogCtl->earliestSegBoundaryStartPtr = InvalidXLogRecPtr;
+		XLogCtl->earliestSegBoundaryEndPtr = InvalidXLogRecPtr;
 	}
 	else if (XLogCtl->earliestSegBoundary <= flushed_seg &&
 			 XLogCtl->earliestSegBoundaryEndPtr <= flushRecPtr)
 	{
 		latest_boundary_seg = XLogCtl->earliestSegBoundary;
-		keep_latest = true;
+		XLogCtl->earliestSegBoundary = XLogCtl->latestSegBoundary;
+		XLogCtl->earliestSegBoundaryStartPtr = XLogCtl->latestSegBoundaryStartPtr;
+		XLogCtl->earliestSegBoundaryEndPtr = XLogCtl->latestSegBoundaryEndPtr;
 	}
 	else
 	{
@@ -4465,41 +4481,38 @@ NotifySegmentsReadyForArchive(XLogRecPtr flushRecPtr)
 		return;
 	}
 
-	last_notified = XLogCtl->lastNotifiedSeg;
-
-	/*
-	 * Update shared memory and discard segment boundaries that are no longer
-	 * needed.
-	 *
-	 * It is safe to update shared memory before we attempt to create the
-	 * .ready files.  If our calls to XLogArchiveNotifySeg() fail,
-	 * RemoveOldXlogFiles() will retry it as needed.
-	 */
-	if (last_notified < latest_boundary_seg - 1)
-		XLogCtl->lastNotifiedSeg = latest_boundary_seg - 1;
-
-	if (keep_latest)
-	{
-		XLogCtl->earliestSegBoundary = XLogCtl->latestSegBoundary;
-		XLogCtl->earliestSegBoundaryEndPtr = XLogCtl->latestSegBoundaryEndPtr;
-	}
-	else
-	{
-		XLogCtl->earliestSegBoundary = MaxXLogSegNo;
-		XLogCtl->earliestSegBoundaryEndPtr = InvalidXLogRecPtr;
-	}
-
 	XLogCtl->latestSegBoundary = MaxXLogSegNo;
+	XLogCtl->latestSegBoundaryStartPtr = InvalidXLogRecPtr;
 	XLogCtl->latestSegBoundaryEndPtr = InvalidXLogRecPtr;
 
-	SpinLockRelease(&XLogCtl->segtrack_lck);
+	if (XLogArchivingActive())
+	{
+		XLogSegNo	last_notified;
 
-	/*
-	 * Notify archiver about segments that are ready for archival (by creating
-	 * the corresponding .ready files).
-	 */
-	for (seg = last_notified + 1; seg < latest_boundary_seg; seg++)
-		XLogArchiveNotifySeg(seg, false);
+		last_notified = XLogCtl->lastNotifiedSeg;
+
+		/*
+		 * Update shared memory and discard segment boundaries that are no
+		 * longer needed.
+		 *
+		 * It is safe to have updated shared memory before we attempt to
+		 * create the .ready files.  If our calls to XLogArchiveNotifySeg()
+		 * fail, RemoveOldXlogFiles() will retry it as needed.
+		 */
+		if (last_notified < latest_boundary_seg - 1)
+			XLogCtl->lastNotifiedSeg = latest_boundary_seg - 1;
+
+		SpinLockRelease(&XLogCtl->segtrack_lck);
+
+		/*
+		 * Notify archiver about segments that are ready for archival (by
+		 * creating the corresponding .ready files).
+		 */
+		for (seg = last_notified + 1; seg < latest_boundary_seg; seg++)
+			XLogArchiveNotifySeg(seg, false);
+	}
+	else
+		SpinLockRelease(&XLogCtl->segtrack_lck);
 
 	PgArchWakeup();
 }
@@ -8776,6 +8789,30 @@ GetFlushRecPtr(void)
 	return LogwrtResult.Flush;
 }
 
+/*
+ * GetSafeFlushRecPtr -- Returns a "safe" flush position.
+ *
+ * Similar to the above, except that avoid reporting a location that might be
+ * overwritten if there's a crash before syncing the next segment.
+ */
+XLogRecPtr
+GetSafeFlushRecPtr(void)
+{
+	XLogRecPtr		flush;
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	flush = XLogCtl->LogwrtResult.Flush;
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	SpinLockAcquire(&XLogCtl->segtrack_lck);
+	if (XLogCtl->earliestSegBoundary != MaxXLogSegNo &&
+		XLogCtl->earliestSegBoundaryStartPtr < flush)
+		flush = XLogCtl->earliestSegBoundaryStartPtr;
+	SpinLockRelease(&XLogCtl->segtrack_lck);
+
+	return flush;
+}
+
 /*
  * GetLastImportantRecPtr -- Returns the LSN of the last important record
  * inserted. All records not explicitly marked as unimportant are considered
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 3ca2a11389..aafb3e91de 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -621,7 +621,7 @@ StartReplication(StartReplicationCmd *cmd)
 		FlushPtr = GetStandbyFlushRecPtr();
 	}
 	else
-		FlushPtr = GetFlushRecPtr();
+		FlushPtr = GetSafeFlushRecPtr();
 
 	if (cmd->timeline != 0)
 	{
@@ -2650,7 +2650,7 @@ XLogSendPhysical(void)
 		 * primary: if the primary subsequently crashes and restarts, standbys
 		 * must not have applied any WAL that got lost on the primary.
 		 */
-		SendRqstPtr = GetFlushRecPtr();
+		SendRqstPtr = GetSafeFlushRecPtr();
 	}
 
 	/*
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.h
index 6b6ae81c2d..1af59c36d4 100644
--- a/src/include/access/xlog.h
+++ b/src/include/access/xlog.h
@@ -313,6 +313,7 @@ extern void GetFullPageWriteInfo(XLogRecPtr *RedoRecPtr_p, bool *doPageWrites_p)
 extern XLogRecPtr GetRedoRecPtr(void);
 extern XLogRecPtr GetInsertRecPtr(void);
 extern XLogRecPtr GetFlushRecPtr(void);
+extern XLogRecPtr GetSafeFlushRecPtr(void);
 extern XLogRecPtr GetLastImportantRecPtr(void);
 extern void RemovePromoteSignalFiles(void);
 extern void NotifySegmentsReadyForArchive(XLogRecPtr flushRecPtr);

--agszmhxcdeupegfm--