Wake primary_flush waiters after implicit WAL flushes

XLogWrite() can advance the flush position when it finishes a WAL
segment, even if its caller requested only a write. When this happens
while AdvanceXLInsertBuffer() recycles a buffer, primary_flush waiters
are not notified and can sleep until an unrelated flush or checkpoint.

Track such progress with a process-local pending flag and perform the
wakeup after releasing WAL write and insertion locks, keeping
WaitLSNLock acquisition outside the contended WAL path. Add a
deterministic injection-point test that forces this path and inspects
the registration directly in shared memory.

Author: Xuneng Zhou <xunengzhou@gmail.com>
Discussion: https://postgr.es/m/CABPTF7U0gW5%2B-4oL7-qdML-yerZxUb7ku4QXp7JxCYo0qyJ_Tw%40mail.gmail.com
Reviewed-by: Alexander Korotkov <aekorotkov@gmail.com>
Backpatch-through: 19

src/backend/access/transam/xlog.c

@@ -639,6 +639,12 @@ static int	UsableBytesInSegment;
  */
 static XLogwrtResult LogwrtResult = {0, 0};
 
+/*
+ * True if this process has published primary-flush progress that has not yet
+ * been reported to primary-flush waiters.
+ */
+static bool primaryFlushWakeupPending = false;
+
 /*
  * Update local copy of shared XLogCtl->log{Write,Flush}Result
  *
@@ -652,6 +658,23 @@ static XLogwrtResult LogwrtResult = {0, 0};
 		_target.Write = pg_atomic_read_u64(&XLogCtl->logWriteResult); \
 	} while (0)
 
+/*
+ * Process a primary-flush wakeup requested by XLogWrite().  The caller must
+ * not hold WALWriteLock or any WAL insertion lock.
+ */
+static void
+PrimaryFlushWakeupProcessRequests(void)
+{
+	if (unlikely(primaryFlushWakeupPending))
+	{
+		/* Clear the process-local request before satisfying it. */
+		primaryFlushWakeupPending = false;
+
+		/* XLogWrite() published this frontier before setting the request. */
+		WaitLSNWakeup(WAIT_LSN_TYPE_PRIMARY_FLUSH, LogwrtResult.Flush);
+	}
+}
+
 /*
  * openLogFile is -1 or a kernel FD for an open log file segment.
  * openLogSegNo identifies the segment, and openLogTLI the corresponding TLI.
@@ -1047,6 +1070,9 @@ XLogInsertRecord(XLogRecData *rdata,
 		}
 	}
 
+	/* Process any flush progress published while making room for the record. */
+	PrimaryFlushWakeupProcessRequests();
+
 #ifdef WAL_DEBUG
 	if (XLOG_DEBUG)
 	{
@@ -2334,6 +2360,7 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible)
 	bool		ispartialpage;
 	bool		last_iteration;
 	bool		finishing_seg;
+	XLogRecPtr	oldFlush;
 	int			curridx;
 	int			npages;
 	int			startidx;
@@ -2346,6 +2373,7 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible)
 	 * Update local LogwrtResult (caller probably did this already, but...)
 	 */
 	RefreshXLogWriteResult(LogwrtResult);
+	oldFlush = LogwrtResult.Flush;
 
 	/*
 	 * Since successive pages in the xlog cache are consecutively allocated,
@@ -2607,6 +2635,10 @@ XLogWrite(XLogwrtRqst WriteRqst, TimeLineID tli, bool flexible)
 	pg_write_barrier();
 	pg_atomic_write_u64(&XLogCtl->logFlushResult, LogwrtResult.Flush);
 
+	/* Defer notification until the caller has released its WAL locks. */
+	if (LogwrtResult.Flush > oldFlush)
+		primaryFlushWakeupPending = true;
+
 #ifdef USE_ASSERT_CHECKING
 	{
 		XLogRecPtr	Flush;
@@ -2944,6 +2976,7 @@ XLogFlush(XLogRecPtr record)
 	 * Wake up processes waiting for primary flush LSN to reach current flush
 	 * position.
 	 */
+	primaryFlushWakeupPending = false;
 	WaitLSNWakeup(WAIT_LSN_TYPE_PRIMARY_FLUSH, LogwrtResult.Flush);
 
 	/*
@@ -3132,6 +3165,7 @@ XLogBackgroundFlush(void)
 	 * Wake up processes waiting for primary flush LSN to reach current flush
 	 * position.
 	 */
+	primaryFlushWakeupPending = false;
 	WaitLSNWakeup(WAIT_LSN_TYPE_PRIMARY_FLUSH, LogwrtResult.Flush);
 
 	/*

src/backend/access/transam/xlogwait.c

@@ -60,6 +60,7 @@
 #include "storage/shmem.h"
 #include "storage/subsystems.h"
 #include "utils/fmgrprotos.h"
+#include "utils/injection_point.h"
 #include "utils/pg_lsn.h"
 #include "utils/snapmgr.h"
 #include "utils/wait_event.h"
@@ -239,6 +240,8 @@ addLSNWaiter(XLogRecPtr lsn, WaitLSNType lsnType)
 	updateMinWaitedLSN(lsnType);
 
 	LWLockRelease(WaitLSNLock);
+
+	INJECTION_POINT("wait-for-lsn-after-register", NULL);
 }
 
 /*

src/backend/postmaster/walwriter.c

@@ -60,6 +60,7 @@
 #include "storage/procsignal.h"
 #include "storage/smgr.h"
 #include "utils/hsearch.h"
+#include "utils/injection_point.h"
 #include "utils/memutils.h"
 #include "utils/resowner.h"
 #include "utils/wait_event.h"
@@ -240,6 +241,8 @@ WalWriterMain(const void *startup_data, size_t startup_data_len)
 		/* Process any signals received recently */
 		ProcessMainLoopInterrupts();
 
+		INJECTION_POINT("walwriter-before-background-flush", NULL);
+
 		/*
 		 * Do what we're here for; then, if XLogBackgroundFlush() found useful
 		 * work to do, reset hibernation counter.

src/test/recovery/t/049_wait_for_lsn.pl

@@ -1257,4 +1257,150 @@ $tl_standby2->stop;
 $tl_standby1->stop;
 $tl_primary->stop;
 
+# This final test needs an injection-point-enabled build.  Earlier test nodes
+# have already been stopped, so check the extensions on its dedicated node.
+if ($ENV{enable_injection_points} ne 'yes')
+{
+	done_testing();
+	exit;
+}
+
+# 13. AdvanceXLInsertBuffer() asks XLogWrite() only to write WAL, but completing
+# a segment also advances the flush position.  Verify that this implicit flush
+# wakes primary_flush waiters.
+my $flush_wake_primary = PostgreSQL::Test::Cluster->new('flush_wake_primary');
+$flush_wake_primary->init(extra => ['--wal-segsize=1']);
+$flush_wake_primary->append_conf(
+	'postgresql.conf', qq[
+wal_buffers = 4
+autovacuum = off
+checkpoint_timeout = '1h'
+max_wal_size = '1GB'
+bgwriter_lru_maxpages = 0
+]);
+$flush_wake_primary->start;
+
+# The modules may be absent under installcheck.
+if (   !$flush_wake_primary->check_extension('injection_points')
+	|| !$flush_wake_primary->check_extension('test_wait_lsn'))
+{
+	$flush_wake_primary->stop;
+	done_testing();
+	exit;
+}
+
+is( $flush_wake_primary->safe_psql(
+		'postgres',
+		"SELECT setting::int FROM pg_settings WHERE name = 'wal_segment_size'"
+	),
+	1024 * 1024,
+	'dedicated node runs with 1MB WAL segments');
+
+# Keep the WAL writer out of XLogBackgroundFlush(), whose ordinary flush
+# wakeup would mask the path under test.
+$flush_wake_primary->safe_psql(
+	'postgres', q[
+	CREATE EXTENSION injection_points;
+	CREATE EXTENSION test_wait_lsn;
+	SELECT injection_points_attach('walwriter-before-background-flush', 'wait');
+]);
+$flush_wake_primary->wait_for_event('walwriter',
+	'walwriter-before-background-flush');
+
+my $writer = $flush_wake_primary->background_psql('postgres');
+$writer->set_query_timer_restart;
+
+# Begin in a fresh segment, then round the insert pointer up to its end.
+$writer->query_safe('SELECT pg_switch_wal()');
+my $target_lsn = $writer->query_safe(
+	q[
+	WITH p AS
+	(
+		SELECT pg_current_wal_insert_lsn() AS lsn,
+			   setting::numeric AS segsz
+		FROM pg_settings
+		WHERE name = 'wal_segment_size'
+	)
+	SELECT lsn + (segsz - (lsn - '0/0'::pg_lsn) % segsz)
+	FROM p
+]);
+
+is( $writer->query_safe(
+		"SELECT pg_current_wal_flush_lsn() < '$target_lsn'::pg_lsn"),
+	't',
+	'wakeup target is not flushed initially');
+
+my $waiter = $flush_wake_primary->background_psql('postgres');
+$waiter->set_query_timer_restart;
+my $waiter_pid = $waiter->query_safe('SELECT pg_backend_pid()');
+
+$flush_wake_primary->safe_psql(
+	'postgres', q[
+	SELECT injection_points_attach('wait-for-lsn-after-register', 'wait');
+]);
+
+$waiter->query_until(
+	qr/^wait_started\r?$/m, qq[
+	\\echo wait_started
+	WAIT FOR LSN '$target_lsn'
+		WITH (MODE 'primary_flush', no_throw);
+	\\echo wait_finished
+]);
+$flush_wake_primary->wait_for_event('client backend',
+	'wait-for-lsn-after-register');
+
+is( $flush_wake_primary->safe_psql(
+		'postgres', qq[
+		SELECT test_wait_lsn_waiter_is_registered(
+			$waiter_pid, 'primary_flush', '$target_lsn')
+	]),
+	't',
+	'primary_flush waiter is registered before the implicit flush');
+
+# A 2MB nontransactional, non-flushing logical message assigns no XID and
+# dirties no relation buffers.  With four WAL buffers, it must recycle through
+# the target.  Keep its transaction open so implicit transaction completion
+# cannot call XLogSetAsyncXactLSN(); the xid-less rollback writes no WAL.
+$writer->query_safe('BEGIN');
+$writer->query_safe(
+	q[
+	SELECT pg_logical_emit_message(
+		false, 'wfl', repeat('x', 2 * 1024 * 1024), false)
+]);
+
+is( $writer->query_safe(
+		"SELECT pg_current_wal_flush_lsn() >= '$target_lsn'::pg_lsn"),
+	't',
+	'finishing a segment advanced flush through the target');
+
+# The waiter is still parked, so only a progress waker can remove it.
+is( $flush_wake_primary->safe_psql(
+		'postgres', qq[
+		SELECT test_wait_lsn_waiter_is_registered(
+			$waiter_pid, 'primary_flush', '$target_lsn')
+	]),
+	'f',
+	'implicit segment flush removes the primary_flush waiter');
+
+$flush_wake_primary->safe_psql(
+	'postgres', q[
+	SELECT injection_points_detach('wait-for-lsn-after-register');
+	SELECT injection_points_wakeup('wait-for-lsn-after-register');
+]);
+
+like($waiter->query_until(qr/^wait_finished\r?$/m, ''),
+	qr/^success\r?$/m, 'primary_flush waiter returns success after release');
+
+# Detach first so that the WAL writer cannot immediately re-enter the point.
+$flush_wake_primary->safe_psql(
+	'postgres', q[
+	SELECT injection_points_detach('walwriter-before-background-flush');
+	SELECT injection_points_wakeup('walwriter-before-background-flush');
+]);
+
+$writer->query_safe('ROLLBACK');
+$writer->quit;
+$waiter->quit;
+$flush_wake_primary->stop;
+
 done_testing();