From 7d0661b4ea82c441071a9fba7873efe29e4083cf Mon Sep 17 00:00:00 2001
From: Ayush Tiwari <ayushtiwari.slg01@gmail.com>
Date: Wed, 2 Sep 2026 16:00:15 +0000
Subject: [PATCH v3 2/2] WIP: Batch fsyncs when recycling WAL segments

The previous commit batches durable removal of obsolete WAL files.
Extend that checkpoint pass to recycled files as well: rename each old
segment into its future name, fsync the destination files, and fsync
pg_wal once for the whole batch.  Archive-status cleanup still happens
only after the batch is durable.

A recycled segment becomes usable by the WAL write path as soon as it
is renamed, before the batched fsync.  Track a timeline-qualified
frontier of segments whose pg_wal entry is known durable, and have
XLogFileInit() fsync the segment and pg_wal before writing one that the
frontier does not yet cover.

The frontier alone is not enough.  A foreground installer can skip the
batch's pending names and durably install a higher segment, advancing
the frontier past renames that are not yet durable.  Mark a recycle
batch active before its first rename and clear it only after the
per-file and pg_wal fsyncs complete; the write path takes the fallback
whenever the marker is set.  A failed batch leaves the marker set,
keeping the conservative path enabled until restart.

Add an injection-point TAP test that drives a foreground installation
past the pending recycled segments, verifies the write-path fallback
runs, and checks that committed data survives a crash taken while the
batch is still pending.
---
 src/backend/access/transam/xlog.c             | 284 ++++++++++++++++--
 src/test/recovery/meson.build                 |   1 +
 .../recovery/t/057_wal_recycle_durability.pl  | 204 +++++++++++++
 3 files changed, 462 insertions(+), 27 deletions(-)
 create mode 100644 src/test/recovery/t/057_wal_recycle_durability.pl

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 204c36a18c8..cd0e0e4b584 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -466,6 +466,19 @@ typedef struct XLogCtlData
 
 	XLogSegNo	lastRemovedSegNo;	/* latest removed/recycled XLOG segment */
 
+	/*
+	 * Directory-durability frontier: the highest WAL segment whose pg_wal
+	 * entry is known durable, on timeline InstalledDurableTLI.  Lets batched
+	 * recycling defer fsyncs without the write path entering a segment whose
+	 * rename is not yet flushed.  WalRecycleBatchActive overrides the frontier
+	 * while a batch is pending, because an unrelated installer could otherwise
+	 * advance it past pending renames.  Timeline-qualified so a stale value
+	 * cannot carry across a promotion.  Protected by info_lck.
+	 */
+	TimeLineID	InstalledDurableTLI;
+	XLogSegNo	InstalledDurableSeg;
+	bool		WalRecycleBatchActive;
+
 	/* Fake LSN counter, for unlogged relations. */
 	pg_atomic_uint64 unloggedLSN;
 
@@ -716,15 +729,18 @@ static void AdvanceXLInsertBuffer(XLogRecPtr upto, TimeLineID tli,
 static void XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible);
 static bool InstallXLogFileSegment(XLogSegNo *segno, char *tmppath,
 								   bool find_free, XLogSegNo max_segno,
-								   TimeLineID tli);
+								   TimeLineID tli, bool batch);
 static void XLogFileClose(void);
+static void AdvanceInstalledDurableSeg(TimeLineID tli, XLogSegNo segno);
+static void EnsureXLogSegDirDurable(int fd, XLogSegNo segno, TimeLineID tli);
 static void PreallocXlogFiles(XLogRecPtr endptr, TimeLineID tli);
 static void RemoveTempXlogFiles(void);
 static void RemoveOldXlogFiles(XLogSegNo segno, XLogRecPtr lastredoptr,
 							   XLogRecPtr endptr, TimeLineID insertTLI);
 static void RemoveXlogFile(const struct dirent *segment_de,
 						   XLogSegNo recycleSegNo, XLogSegNo *endlogSegNo,
-						   TimeLineID insertTLI, List **cleanup_names);
+						   TimeLineID insertTLI, bool batch,
+						   List **recycled_paths, List **cleanup_names);
 static void UpdateLastRemovedPtr(char *filename);
 static void ValidateXLOGDirectoryStructure(void);
 static void CleanupBackupHistory(void);
@@ -3392,6 +3408,8 @@ XLogFileInitInternal(XLogSegNo logsegno, TimeLineID logtli,
 	 */
 	installed_segno = logsegno;
 
+	INJECTION_POINT_CACHED("wal-recycle-before-segment-install", NULL);
+
 	/*
 	 * XXX: What should we use as max_segno? We used to use XLOGfileslop when
 	 * that was a constant, but that was always a bit dubious: normally, at a
@@ -3403,7 +3421,7 @@ XLogFileInitInternal(XLogSegNo logsegno, TimeLineID logtli,
 	 */
 	max_segno = logsegno + CheckPointSegments;
 	if (InstallXLogFileSegment(&installed_segno, tmppath, true, max_segno,
-							   logtli))
+							   logtli, false))
 	{
 		*added = true;
 		elog(DEBUG2, "done creating and filling new WAL file");
@@ -3444,16 +3462,24 @@ XLogFileInit(XLogSegNo logsegno, TimeLineID logtli)
 	Assert(logtli != 0);
 
 	fd = XLogFileInitInternal(logsegno, logtli, &ignore_added, path);
-	if (fd >= 0)
-		return fd;
-
-	/* Now open original target segment (might not be file I just made) */
-	fd = BasicOpenFile(path, O_RDWR | PG_BINARY | O_CLOEXEC |
-					   get_sync_bit(wal_sync_method));
 	if (fd < 0)
-		ereport(ERROR,
-				(errcode_for_file_access(),
-				 errmsg("could not open file \"%s\": %m", path)));
+	{
+		/* Now open original target segment (might not be file I just made) */
+		fd = BasicOpenFile(path, O_RDWR | PG_BINARY | O_CLOEXEC |
+						   get_sync_bit(wal_sync_method));
+		if (fd < 0)
+			ereport(ERROR,
+					(errcode_for_file_access(),
+					 errmsg("could not open file \"%s\": %m", path)));
+	}
+
+	/*
+	 * About to write WAL here.  If the checkpointer recycled this segment with
+	 * a deferred fsync, make its rename durable first.  Covers both the reuse
+	 * and reopen paths above; PreallocXlogFiles() only pre-creates and calls
+	 * XLogFileInitInternal() directly, so it does not reach here.
+	 */
+	EnsureXLogSegDirDurable(fd, logsegno, logtli);
 	return fd;
 }
 
@@ -3587,7 +3613,7 @@ XLogFileCopy(TimeLineID destTLI, XLogSegNo destsegno,
 	/*
 	 * Now move the segment into place with its final name.
 	 */
-	if (!InstallXLogFileSegment(&destsegno, tmppath, false, 0, destTLI))
+	if (!InstallXLogFileSegment(&destsegno, tmppath, false, 0, destTLI, false))
 		elog(ERROR, "InstallXLogFileSegment should not have failed");
 }
 
@@ -3619,7 +3645,8 @@ XLogFileCopy(TimeLineID destTLI, XLogSegNo destsegno,
  */
 static bool
 InstallXLogFileSegment(XLogSegNo *segno, char *tmppath,
-					   bool find_free, XLogSegNo max_segno, TimeLineID tli)
+					   bool find_free, XLogSegNo max_segno, TimeLineID tli,
+					   bool batch)
 {
 	char		path[MAXPGPATH];
 	struct stat stat_buf;
@@ -3628,6 +3655,14 @@ InstallXLogFileSegment(XLogSegNo *segno, char *tmppath,
 
 	XLogFilePath(path, tli, *segno, wal_segment_size);
 
+	if (batch)
+	{
+		/* Publish the pending batch before its first rename is visible. */
+		SpinLockAcquire(&XLogCtl->info_lck);
+		XLogCtl->WalRecycleBatchActive = true;
+		SpinLockRelease(&XLogCtl->info_lck);
+	}
+
 	LWLockAcquire(ControlFileLock, LW_EXCLUSIVE);
 	if (!XLogCtl->InstallXLogFileSegmentActive)
 	{
@@ -3657,7 +3692,27 @@ InstallXLogFileSegment(XLogSegNo *segno, char *tmppath,
 	}
 
 	Assert(access(path, F_OK) != 0 && errno == ENOENT);
-	if (durable_rename(tmppath, path, LOG) != 0)
+
+	if (batch)
+	{
+		/*
+		 * Plain rename; the caller fsyncs the renamed files and pg_wal
+		 * once the whole pass is done, so the filesystem can coalesce the
+		 * flushes.  The pre-rename source fsync that durable_rename() would
+		 * do is skipped: a WAL segment is already durable by the time it is
+		 * recycled.
+		 */
+		if (rename(tmppath, path) < 0)
+		{
+			LWLockRelease(ControlFileLock);
+			ereport(LOG,
+					(errcode_for_file_access(),
+					 errmsg("could not rename file \"%s\" to \"%s\": %m",
+							tmppath, path)));
+			return false;
+		}
+	}
+	else if (durable_rename(tmppath, path, LOG) != 0)
 	{
 		LWLockRelease(ControlFileLock);
 		/* durable_rename already emitted log message */
@@ -3666,6 +3721,17 @@ InstallXLogFileSegment(XLogSegNo *segno, char *tmppath,
 
 	LWLockRelease(ControlFileLock);
 
+	/*
+	 * The non-batched path used durable_rename(), so the entry is durable now;
+	 * advance the frontier.  In batched mode the caller advances it after its
+	 * bulk pg_wal fsync.
+	 */
+	if (!batch)
+	{
+		AdvanceInstalledDurableSeg(tli, *segno);
+		INJECTION_POINT_CACHED("wal-recycle-after-segment-install", NULL);
+	}
+
 	return true;
 }
 
@@ -3880,6 +3946,108 @@ UpdateLastRemovedPtr(char *filename)
 	SpinLockRelease(&XLogCtl->info_lck);
 }
 
+/*
+ * Advance the directory-durability frontier to at least (tli, segno).
+ * Called once a segment's rename into pg_wal has been made durable (by
+ * durable_rename(), or by the checkpointer's batched fsync of pg_wal).
+ *
+ * The frontier is timeline-qualified: a higher timeline always supersedes the
+ * previous one (its segments are different files needing their own fsync), and
+ * within a timeline the highest segment number wins.
+ */
+static void
+AdvanceInstalledDurableSeg(TimeLineID tli, XLogSegNo segno)
+{
+	SpinLockAcquire(&XLogCtl->info_lck);
+	if (tli > XLogCtl->InstalledDurableTLI)
+	{
+		XLogCtl->InstalledDurableTLI = tli;
+		XLogCtl->InstalledDurableSeg = segno;
+	}
+	else if (tli == XLogCtl->InstalledDurableTLI &&
+			 segno > XLogCtl->InstalledDurableSeg)
+		XLogCtl->InstalledDurableSeg = segno;
+	SpinLockRelease(&XLogCtl->info_lck);
+}
+
+/*
+ * Ensure segment (segno, tli)'s pg_wal entry is durable before it is written.
+ *
+ * Batched recycling renames segments but defers the pg_wal fsync, so if the
+ * write path reaches a just-renamed segment first, its rename is not yet
+ * durable (issue_xlog_fsync() fsyncs only the file, never pg_wal).  If the
+ * frontier does not already cover it, or a recycle batch is active, fsync the
+ * file and pg_wal here.  The active marker prevents a concurrent durable
+ * installation at a higher segment from hiding lower pending renames.  The
+ * timeline-qualified check keeps a stale frontier from carrying across a
+ * promotion.  Normally the checkpointer stays ahead and this is just a
+ * spinlock-protected compare.
+ */
+static void
+EnsureXLogSegDirDurable(int fd, XLogSegNo segno, TimeLineID tli)
+{
+	bool		durable;
+
+	SpinLockAcquire(&XLogCtl->info_lck);
+	durable = (!XLogCtl->WalRecycleBatchActive &&
+			   tli == XLogCtl->InstalledDurableTLI &&
+			   segno <= XLogCtl->InstalledDurableSeg);
+	SpinLockRelease(&XLogCtl->info_lck);
+
+	if (durable)
+		return;
+
+	INJECTION_POINT_CACHED("wal-recycle-before-write-path-fsync", NULL);
+
+	/*
+	 * fsync the file (carries the rename where we cannot fsync a directory,
+	 * e.g. Windows) and fsync pg_wal (carries it on most Unix filesystems).
+	 */
+	if (pg_fsync(fd) != 0)
+	{
+		char		path[MAXPGPATH];
+
+		XLogFilePath(path, tli, segno, wal_segment_size);
+		ereport(data_sync_elevel(ERROR),
+				(errcode_for_file_access(),
+				 errmsg("could not fsync file \"%s\": %m", path)));
+	}
+	fsync_fname(XLOGDIR, true);
+
+	AdvanceInstalledDurableSeg(tli, segno);
+}
+
+/*
+ * fsync a recycled segment file by name, tolerating concurrent removal.
+ * Unlike fsync_fname(), ENOENT is not an error: if another process removed
+ * the segment after we renamed it, its durability is no longer our concern.
+ */
+static void
+fsync_fname_recycled(const char *fname)
+{
+	int			fd;
+
+	fd = BasicOpenFile(fname, O_RDWR | PG_BINARY | O_CLOEXEC);
+	if (fd < 0)
+	{
+		if (errno == ENOENT)
+			return;
+		ereport(data_sync_elevel(ERROR),
+				(errcode_for_file_access(),
+				 errmsg("could not open file \"%s\": %m", fname)));
+	}
+
+	if (pg_fsync(fd) != 0)
+		ereport(data_sync_elevel(ERROR),
+				(errcode_for_file_access(),
+				 errmsg("could not fsync file \"%s\": %m", fname)));
+
+	if (close(fd) != 0)
+		ereport(data_sync_elevel(ERROR),
+				(errcode_for_file_access(),
+				 errmsg("could not close file \"%s\": %m", fname)));
+}
+
 /*
  * Remove all temporary log files in pg_wal
  *
@@ -3928,8 +4096,15 @@ RemoveOldXlogFiles(XLogSegNo segno, XLogRecPtr lastredoptr, XLogRecPtr endptr,
 	char		lastoff[MAXFNAMELEN];
 	XLogSegNo	endlogSegNo;
 	XLogSegNo	recycleSegNo;
+	List	   *recycled_paths = NIL;
 	List	   *cleanup_names = NIL;
 	ListCell   *lc;
+	bool		batch_was_active;
+
+	/* Preserve a conservative marker left by an earlier failed batch. */
+	SpinLockAcquire(&XLogCtl->info_lck);
+	batch_was_active = XLogCtl->WalRecycleBatchActive;
+	SpinLockRelease(&XLogCtl->info_lck);
 
 	/* Initialize info about where to try to recycle to */
 	XLByteToSeg(endptr, endlogSegNo, wal_segment_size);
@@ -3973,21 +4148,54 @@ RemoveOldXlogFiles(XLogSegNo segno, XLogRecPtr lastredoptr, XLogRecPtr endptr,
 				UpdateLastRemovedPtr(xlde->d_name);
 
 				RemoveXlogFile(xlde, recycleSegNo, &endlogSegNo, insertTLI,
-							   &cleanup_names);
+							   true, &recycled_paths, &cleanup_names);
 			}
 		}
 	}
 
 	FreeDir(xldir);
 
-	/* Make all removals durable with one directory fsync. */
-	if (cleanup_names != NIL)
+	/*
+	 * Test hook: pause here, after the renames but before they are made
+	 * durable, so a test can drive the write frontier into a just-recycled
+	 * segment and crash, exercising the write-path barrier.
+	 */
+	INJECTION_POINT("wal-recycle-before-batch-fsync", NULL);
+
+	/*
+	 * Make the whole pass durable at once: fsync each recycled file, then
+	 * fsync pg_wal a single time.  Renaming first and fsyncing afterwards lets
+	 * the filesystem coalesce what used to be one journal flush per segment.
+	 * The per-file fsyncs usually do nothing (contents are already durable)
+	 * but persist the rename where we cannot fsync a directory, e.g. Windows.
+	 * A segment concurrently removed (e.g. by promotion cleanup) is skipped.
+	 */
+	foreach(lc, recycled_paths)
+		fsync_fname_recycled((char *) lfirst(lc));
+
+	if (recycled_paths != NIL || cleanup_names != NIL)
 		fsync_fname(XLOGDIR, true);
 
-	/* Remove archive status only after the WAL-file removals are durable. */
+	if (recycled_paths != NIL)
+		AdvanceInstalledDurableSeg(insertTLI, endlogSegNo - 1);
+
+	/* A failed earlier batch keeps the conservative slow path enabled. */
+	if (!batch_was_active)
+	{
+		SpinLockAcquire(&XLogCtl->info_lck);
+		XLogCtl->WalRecycleBatchActive = false;
+		SpinLockRelease(&XLogCtl->info_lck);
+	}
+
+	/*
+	 * Now that the batch is durable, drop the old segments' archive-status
+	 * files.  Doing it earlier could, after a crash that lost a rename, leave
+	 * an old segment whose .done marker was already gone and re-archive it.
+	 */
 	foreach(lc, cleanup_names)
 		XLogArchiveCleanup((char *) lfirst(lc));
 
+	list_free_deep(recycled_paths);
 	list_free_deep(cleanup_names);
 }
 
@@ -4056,7 +4264,8 @@ RemoveNonParentXlogFiles(XLogRecPtr switchpoint, TimeLineID newTLI)
 			 * - but seems safer to let them be archived and removed later.
 			 */
 			if (!XLogArchiveIsReady(xlde->d_name))
-				RemoveXlogFile(xlde, recycleSegNo, &endLogSegNo, newTLI, NULL);
+				RemoveXlogFile(xlde, recycleSegNo, &endLogSegNo, newTLI,
+						   false, NULL, NULL);
 		}
 	}
 
@@ -4079,10 +4288,10 @@ RemoveNonParentXlogFiles(XLogRecPtr switchpoint, TimeLineID newTLI)
 static void
 RemoveXlogFile(const struct dirent *segment_de,
 			   XLogSegNo recycleSegNo, XLogSegNo *endlogSegNo,
-			   TimeLineID insertTLI, List **cleanup_names)
+			   TimeLineID insertTLI, bool batch,
+			   List **recycled_paths, List **cleanup_names)
 {
 	char		path[MAXPGPATH];
-	bool		removed = false;
 	const char *unlinkpath;
 #ifdef WIN32
 	char		newpath[MAXPGPATH];
@@ -4102,12 +4311,25 @@ RemoveXlogFile(const struct dirent *segment_de,
 		XLogCtl->InstallXLogFileSegmentActive &&	/* callee rechecks this */
 		get_dirent_type(path, segment_de, false, DEBUG2) == PGFILETYPE_REG &&
 		InstallXLogFileSegment(endlogSegNo, path,
-							   true, recycleSegNo, insertTLI))
+							   true, recycleSegNo, insertTLI, batch))
 	{
 		ereport(DEBUG2,
 				(errmsg_internal("recycled write-ahead log file \"%s\"",
 								 segname)));
 		CheckpointStats.ckpt_segs_recycled++;
+
+		/*
+		 * Batched mode only renamed the segment; remember its new path so the
+		 * caller can fsync it once all renames are done.
+		 */
+		if (recycled_paths != NULL)
+		{
+			char		dstpath[MAXPGPATH];
+
+			XLogFilePath(dstpath, insertTLI, *endlogSegNo, wal_segment_size);
+			*recycled_paths = lappend(*recycled_paths, pstrdup(dstpath));
+		}
+
 		/* Needn't recheck that slot on future iterations */
 		(*endlogSegNo)++;
 	}
@@ -4144,13 +4366,18 @@ RemoveXlogFile(const struct dirent *segment_de,
 		unlinkpath = newpath;
 #endif
 
-		if (cleanup_names != NULL)
+		if (batch)
 			rc = unlink(unlinkpath);
 		else
 			rc = durable_unlink(unlinkpath, LOG);
 		if (rc != 0)
 		{
-			if (cleanup_names != NULL)
+			/*
+			 * In batched mode the caller's pg_wal fsync makes the unlink
+			 * durable; plain unlink() only sets errno, so report failures
+			 * here.
+			 */
+			if (batch)
 				ereport(LOG,
 						(errcode_for_file_access(),
 						 errmsg("could not remove file \"%s\": %m",
@@ -4158,10 +4385,13 @@ RemoveXlogFile(const struct dirent *segment_de,
 			return;
 		}
 		CheckpointStats.ckpt_segs_removed++;
-		removed = true;
 	}
 
-	if (removed && cleanup_names != NULL)
+	/*
+	 * Batched mode defers archive-status cleanup to the caller (after the
+	 * batch fsync); unbatched mode is already durable, so clean up now.
+	 */
+	if (cleanup_names != NULL)
 		*cleanup_names = lappend(*cleanup_names, pstrdup(segname));
 	else
 		XLogArchiveCleanup(segname);
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build
index 72113c5ac6e..9d07a337e09 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -65,6 +65,7 @@ tests += {
       't/054_unlogged_sequence_promotion.pl',
       't/055_cascade_reconnect.pl',
       't/056_standby_snapshot_export.pl',
+      't/057_wal_recycle_durability.pl',
     ],
   },
 }
diff --git a/src/test/recovery/t/057_wal_recycle_durability.pl b/src/test/recovery/t/057_wal_recycle_durability.pl
new file mode 100644
index 00000000000..1a5cd79bf84
--- /dev/null
+++ b/src/test/recovery/t/057_wal_recycle_durability.pl
@@ -0,0 +1,204 @@
+# Copyright (c) 2025-2026, PostgreSQL Global Development Group
+#
+# Test the write-path durability barrier for batched WAL-segment recycling.
+#
+# When the checkpointer recycles old WAL segments into future ones it
+# renames them and defers making the renames durable to a single fsync of
+# pg_wal at the end of the pass (plus a per-file fsync of each recycled
+# segment).  A recycled segment becomes usable by the WAL write path as
+# soon as it is renamed, so if the write frontier reaches such a segment
+# before the checkpointer's batched fsync, XLogFileInit() must make the
+# rename durable itself, the write-path "durability barrier"
+# (EnsureXLogSegDirDurable()).
+#
+# This test also makes a foreground WAL segment installation leap past the
+# pending recycled segments.  That advances the durability frontier beyond
+# them, so the active-batch marker must override the frontier and force the
+# write-path barrier.  It then crashes the server with the batch still pending
+# and checks that committed data survives crash recovery.
+use strict;
+use warnings FATAL => 'all';
+
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+if ($ENV{enable_injection_points} ne 'yes')
+{
+	plan skip_all => 'Injection points not supported by this build';
+}
+
+my $node = PostgreSQL::Test::Cluster->new('primary');
+$node->init;
+$node->append_conf(
+	'postgresql.conf', q{
+wal_recycle = on
+min_wal_size = 32MB
+max_wal_size = 1GB
+checkpoint_timeout = 1h
+log_checkpoints = on
+});
+$node->start;
+
+# Skip if the injection_points extension is not installed, e.g. under
+# installcheck where the module may not be present.
+if (!$node->check_extension('injection_points'))
+{
+	plan skip_all => 'Extension injection_points not installed';
+}
+
+$node->safe_psql('postgres', q(CREATE EXTENSION injection_points));
+
+$node->safe_psql(
+	'postgres', q{
+	CREATE TABLE t (id int primary key, v text);
+	INSERT INTO t VALUES (0, 'baseline');
+	CREATE TABLE filler (id int, pad text);
+});
+
+# Generate old segments for the checkpoint to recycle, but do not checkpoint
+# yet.  This keeps the known-durable frontier at the current segment.
+for (1 .. 8)
+{
+	$node->safe_psql('postgres',
+		q{INSERT INTO filler SELECT g, repeat('x', 900) FROM generate_series(1, 20000) g}
+	);
+	$node->safe_psql('postgres', q{SELECT pg_switch_wal()});
+}
+
+# Start a checkpoint in the background.  First pause immediately before old
+# WAL removal so a writer can prepare a temporary segment for installation.
+# The second pause keeps the resulting recycle batch pending.
+my $checkpoint = $node->background_psql('postgres');
+$checkpoint->query_safe(
+	q{
+	select injection_points_attach('checkpoint-before-old-wal-removal', 'wait');
+	select injection_points_attach('wal-recycle-before-batch-fsync', 'wait');
+});
+$checkpoint->query_until(
+	qr/starting_checkpoint/, q(\echo starting_checkpoint
+checkpoint;
+\q
+));
+
+# The checkpoint has fixed its WAL-removal horizon but has not scanned pg_wal.
+$node->wait_for_event('checkpointer', 'checkpoint-before-old-wal-removal');
+
+# Make the next segment slot empty.  The writer will prepare that segment while
+# the checkpoint is paused, and the checkpoint will recycle an old segment into
+# the slot before the writer can install its temporary file.
+my $target_wal = $node->safe_psql(
+	'postgres', q{
+	SELECT pg_walfile_name(pg_current_wal_insert_lsn() +
+		pg_size_bytes(current_setting('wal_segment_size')))
+});
+my $target_path = $node->data_dir . "/pg_wal/$target_wal";
+unlink($target_path) if -e $target_path;
+ok(!-e $target_path, 'next WAL segment slot is empty');
+
+# Prepare the writer's two cached wait callbacks outside its WAL critical
+# section: one before installing its temporary segment and one in the
+# write-path durability barrier.
+my $writer = $node->background_psql('postgres');
+$writer->query_safe(
+	q{
+	select injection_points_attach('wal-recycle-before-segment-install', 'wait');
+	select injection_points_attach('wal-recycle-after-segment-install', 'wait');
+	select injection_points_attach('wal-recycle-before-write-path-fsync', 'wait');
+});
+
+for my $point (
+	'wal-recycle-before-segment-install',
+	'wal-recycle-after-segment-install',
+	'wal-recycle-before-write-path-fsync')
+{
+	$writer->query_until(
+		qr/priming_writer/,
+		"\\echo priming_writer\nselect injection_points_run('$point');\n"
+		  . "\\echo writer_primed\n");
+	$node->wait_for_event('client backend', $point);
+	$node->safe_psql('postgres',
+		"select injection_points_wakeup('$point')");
+	$writer->query_until(qr/writer_primed/, '');
+}
+
+$writer->query_safe(
+	q{
+	select injection_points_load('wal-recycle-before-segment-install');
+	select injection_points_load('wal-recycle-after-segment-install');
+	select injection_points_load('wal-recycle-before-write-path-fsync');
+});
+
+# Switch into the missing target.  The writer creates and fsyncs a temporary
+# segment, then pauses immediately before attempting to install it.
+$writer->query_until(
+	qr/starting_writer/, q(\echo starting_writer
+SELECT pg_switch_wal();
+INSERT INTO t VALUES (1, 'committed-in-window');
+\echo writer_finished
+));
+$node->wait_for_event('client backend',
+	'wal-recycle-before-segment-install');
+
+# Let old-WAL cleanup fill the missing target and pause before its batch fsync.
+$node->safe_psql(
+	'postgres', q{
+	select injection_points_detach('checkpoint-before-old-wal-removal');
+	select injection_points_wakeup('checkpoint-before-old-wal-removal');
+});
+$node->wait_for_event('checkpointer', 'wal-recycle-before-batch-fsync');
+ok(-e $target_path, 'checkpoint recycled a segment into the writer target');
+
+# The writer now finds its target occupied, skips the batch's pending names,
+# and durably installs its temporary file at a higher segment.  That advances
+# the high watermark beyond the target while the recycle batch remains active.
+$node->safe_psql(
+	'postgres', q{
+	select injection_points_detach('wal-recycle-before-segment-install');
+	select injection_points_wakeup('wal-recycle-before-segment-install');
+});
+$node->wait_for_event('client backend',
+	'wal-recycle-after-segment-install');
+pass('foreground WAL installation advanced the durability frontier');
+$node->safe_psql(
+	'postgres', q{
+	select injection_points_detach('wal-recycle-after-segment-install');
+	select injection_points_wakeup('wal-recycle-after-segment-install');
+});
+
+# Despite that higher watermark, the active batch must force the durability
+# barrier before the writer can use the pending recycled target.
+$node->wait_for_event('client backend',
+	'wal-recycle-before-write-path-fsync');
+pass('active recycle batch overrode the advanced durability frontier');
+
+# Detach first so later segment switches do not wait again, then let the first
+# call make the recycled segment and pg_wal durable.
+$node->safe_psql(
+	'postgres', q{
+	select injection_points_detach('wal-recycle-before-write-path-fsync');
+	select injection_points_wakeup('wal-recycle-before-write-path-fsync');
+});
+$writer->query_until(qr/writer_finished/, '');
+$writer->quit;
+
+is( $node->safe_psql('postgres', q{SELECT count(*) FROM t}),
+	'2', 'row committed before crash');
+
+# Crash with the window still open: the checkpointer never ran its batched
+# fsync, so durability of the recycled segments rests entirely on the barrier.
+$node->stop('immediate');
+
+# The checkpoint session's connection died with the crash; reap it quietly.
+eval { $checkpoint->quit; };
+
+# Crash recovery.
+$node->start;
+
+# Every committed row must still be present.
+is( $node->safe_psql(
+		'postgres', q{SELECT string_agg(v, ',' ORDER BY id) FROM t}),
+	'baseline,committed-in-window',
+	'committed row survived crash with the recycle-durability window open');
+
+done_testing();
-- 
2.43.0

