v4-0002-Add-TAP-tests-to-check-replication-slot-advance-d.patch

From 6f2afd8239cdefc62a519825a670db2eb8a4e111 Mon Sep 17 00:00:00 2001
From: Tomas Vondra <tomas@vondra.me>
Date: Thu, 21 Nov 2024 20:37:00 +0100
Subject: [PATCH v4 2/3] Add TAP tests to check replication slot advance during
 the checkpoint

The new tests verify that logical and physical replication slots are still
valid after an immediate restart on checkpoint completion when the slot was
advanced during the checkpoint.

This commit introduces two new injection points to make these tests possible:

* checkpoint-before-old-wal-removal - triggered in the checkpointer process
  just before old WAL segments cleanup;
* logical-replication-slot-advance-segment - triggered in
  LogicalConfirmReceivedLocation() when restart_lsn was changed enough to
  point to the next WAL segment.

Discussion: https://postgr.es/m/flat/1d12d2-67235980-35-19a406a0%4063439497
Author: Vitaly Davydov <v.davydov@postgrespro.ru>
Author: Tomas Vondra <tomas@vondra.me>
Reviewed-by: Alexander Korotkov <aekorotkov@gmail.com>
Backpatch-through: 17
---
 src/backend/access/transam/xlog.c             |   4 +
 src/backend/replication/logical/logical.c     |  18 +++
 .../test_replslot_required_lsn/Makefile       |  18 +++
 .../test_replslot_required_lsn/meson.build    |  16 ++
 src/test/recovery/meson.build                 |   2 +
 src/test/recovery/t/046_logical_slot.pl       | 139 ++++++++++++++++++
 src/test/recovery/t/047_physical_slot.pl      | 136 +++++++++++++++++
 7 files changed, 333 insertions(+)
 create mode 100644 src/test/modules/test_replslot_required_lsn/Makefile
 create mode 100644 src/test/modules/test_replslot_required_lsn/meson.build
 create mode 100644 src/test/recovery/t/046_logical_slot.pl
 create mode 100644 src/test/recovery/t/047_physical_slot.pl

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index 30ae65fce53..9c0f9a0af28 100644
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -7505,6 +7505,10 @@ CreateCheckPoint(int flags)
 	if (PriorRedoPtr != InvalidXLogRecPtr)
 		UpdateCheckPointDistanceEstimate(RedoRecPtr - PriorRedoPtr);
 
+#ifdef USE_INJECTION_POINTS
+	INJECTION_POINT("checkpoint-before-old-wal-removal", NULL);
+#endif
+
 	/*
 	 * Delete old log files, those no longer needed for last checkpoint to
 	 * prevent the disk holding the xlog from growing full.
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 1d56d0c4ef3..f1eb798f3e9 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -29,6 +29,7 @@
 #include "postgres.h"
 
 #include "access/xact.h"
+#include "access/xlog_internal.h"
 #include "access/xlogutils.h"
 #include "fmgr.h"
 #include "miscadmin.h"
@@ -41,6 +42,7 @@
 #include "storage/proc.h"
 #include "storage/procarray.h"
 #include "utils/builtins.h"
+#include "utils/injection_point.h"
 #include "utils/inval.h"
 #include "utils/memutils.h"
 
@@ -1825,9 +1827,13 @@ LogicalConfirmReceivedLocation(XLogRecPtr lsn)
 	{
 		bool		updated_xmin = false;
 		bool		updated_restart = false;
+		XLogRecPtr	restart_lsn pg_attribute_unused();
 
 		SpinLockAcquire(&MyReplicationSlot->mutex);
 
+		/* remember the old restart lsn */
+		restart_lsn = MyReplicationSlot->data.restart_lsn;
+
 		/*
 		 * Prevent moving the confirmed_flush backwards, as this could lead to
 		 * data duplication issues caused by replicating already replicated
@@ -1881,6 +1887,18 @@ LogicalConfirmReceivedLocation(XLogRecPtr lsn)
 		/* first write new xmin to disk, so we know what's up after a crash */
 		if (updated_xmin || updated_restart)
 		{
+#ifdef USE_INJECTION_POINTS
+			XLogSegNo	seg1,
+						seg2;
+
+			XLByteToSeg(restart_lsn, seg1, wal_segment_size);
+			XLByteToSeg(MyReplicationSlot->data.restart_lsn, seg2, wal_segment_size);
+
+			/* trigger injection point, but only if segment changes */
+			if (seg1 != seg2)
+				INJECTION_POINT("logical-replication-slot-advance-segment", NULL);
+#endif
+
 			ReplicationSlotMarkDirty();
 			ReplicationSlotSave();
 			elog(DEBUG1, "updated xmin: %u restart: %u", updated_xmin, updated_restart);
diff --git a/src/test/modules/test_replslot_required_lsn/Makefile b/src/test/modules/test_replslot_required_lsn/Makefile
new file mode 100644
index 00000000000..e5ff8af255b
--- /dev/null
+++ b/src/test/modules/test_replslot_required_lsn/Makefile
@@ -0,0 +1,18 @@
+# src/test/modules/test_replslot_required_lsn/Makefile
+
+EXTRA_INSTALL=src/test/modules/injection_points \
+	contrib/test_decoding
+
+export enable_injection_points
+TAP_TESTS = 1
+
+ifdef USE_PGXS
+PG_CONFIG = pg_config
+PGXS := $(shell $(PG_CONFIG) --pgxs)
+include $(PGXS)
+else
+subdir = src/test/modules/test_replslot_required_lsn
+top_builddir = ../../../..
+include $(top_builddir)/src/Makefile.global
+include $(top_srcdir)/contrib/contrib-global.mk
+endif
diff --git a/src/test/modules/test_replslot_required_lsn/meson.build b/src/test/modules/test_replslot_required_lsn/meson.build
new file mode 100644
index 00000000000..44d2546632b
--- /dev/null
+++ b/src/test/modules/test_replslot_required_lsn/meson.build
@@ -0,0 +1,16 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+
+tests += {
+  'name': 'test_replslot_required_lsn',
+  'sd': meson.current_source_dir(),
+  'bd': meson.current_build_dir(),
+  'tap': {
+    'env': {
+       'enable_injection_points': get_option('injection_points') ? 'yes' : 'no',
+    },
+    'tests': [
+      't/001_logical_slot.pl',
+      't/002_physical_slot.pl'
+    ],
+  },
+}
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build
index cb983766c67..5ee41c3cd4d 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -54,6 +54,8 @@ tests += {
       't/043_no_contrecord_switch.pl',
       't/044_invalidate_inactive_slots.pl',
       't/045_archive_restartpoint.pl',
+      't/046_logical_slot.pl',
+      't/047_physical_slot.pl'
     ],
   },
 }
diff --git a/src/test/recovery/t/046_logical_slot.pl b/src/test/recovery/t/046_logical_slot.pl
new file mode 100644
index 00000000000..e78375178aa
--- /dev/null
+++ b/src/test/recovery/t/046_logical_slot.pl
@@ -0,0 +1,139 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+#
+# This test verifies the case when the logical slot is advanced during
+# checkpoint. The test checks that the logical slot's restart_lsn still refers
+# to an existed WAL segment after immediate restart.
+#
+# Discussion:
+# https://www.postgresql.org/message-id/flat/1d12d2-67235980-35-19a406a0%4063439497
+#
+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, $result);
+
+$node = PostgreSQL::Test::Cluster->new('mike');
+$node->init;
+$node->append_conf('postgresql.conf',
+	"shared_preload_libraries = 'injection_points'");
+$node->append_conf('postgresql.conf', "wal_level = 'logical'");
+$node->start;
+$node->safe_psql('postgres', q(CREATE EXTENSION injection_points));
+
+# create a simple table to generate data into
+$node->safe_psql('postgres',
+	q{create table t (id serial primary key, b text)});
+
+# create the two slots we'll need
+$node->safe_psql('postgres',
+	q{select pg_create_logical_replication_slot('slot_logical', 'test_decoding')}
+);
+$node->safe_psql('postgres',
+	q{select pg_create_physical_replication_slot('slot_physical', true)});
+
+# advance both to current position, just to have everything "valid"
+$node->safe_psql('postgres',
+	q{select count(*) from pg_logical_slot_get_changes('slot_logical', null, null)}
+);
+$node->safe_psql('postgres',
+	q{select pg_replication_slot_advance('slot_physical', pg_current_wal_lsn())}
+);
+
+# run checkpoint, to flush current state to disk and set a baseline
+$node->safe_psql('postgres', q{checkpoint});
+
+# generate transactions to get RUNNING_XACTS
+my $xacts = $node->background_psql('postgres');
+$xacts->query_until(
+	qr/run_xacts/,
+	q(\echo run_xacts
+SELECT 1 \watch 0.1
+\q
+));
+
+# insert 2M rows, that's about 260MB (~20 segments) worth of WAL
+$node->safe_psql('postgres',
+	q{insert into t (b) select md5(i::text) from generate_series(1,1000000) s(i)}
+);
+
+# run another checkpoint, to set a new restore LSN
+$node->safe_psql('postgres', q{checkpoint});
+
+# another 2M rows, that's about 260MB (~20 segments) worth of WAL
+$node->safe_psql('postgres',
+	q{insert into t (b) select md5(i::text) from generate_series(1,1000000) s(i)}
+);
+
+# run another checkpoint, this time in the background, and make it wait
+# on the injection point), so that the checkpoint stops right before
+# removing old WAL segments
+print('starting checkpoint\n');
+
+my $checkpoint = $node->background_psql('postgres');
+$checkpoint->query_safe(
+	q(select injection_points_attach('checkpoint-before-old-wal-removal','wait'))
+);
+$checkpoint->query_until(
+	qr/starting_checkpoint/,
+	q(\echo starting_checkpoint
+checkpoint;
+\q
+));
+
+print('waiting for injection_point\n');
+# wait until the checkpoint stops right before removing WAL segments
+$node->wait_for_event('checkpointer', 'checkpoint-before-old-wal-removal');
+
+
+# try to advance the logical slot, but make it stop when it moves to the
+# next WAL segment (has to happen in the background too)
+my $logical = $node->background_psql('postgres');
+$logical->query_safe(
+	q{select injection_points_attach('logical-replication-slot-advance-segment','wait');}
+);
+$logical->query_until(
+	qr/get_changes/,
+	q(
+\echo get_changes
+select count(*) from pg_logical_slot_get_changes('slot_logical', null, null) \watch 1
+\q
+));
+
+# wait until the checkpoint stops right before removing WAL segments
+$node->wait_for_event('client backend',
+	'logical-replication-slot-advance-segment');
+
+# OK, we're in the right situation,  time to advance the physical slot,
+# which recalculates the required LSN, and then unblock the checkpoint,
+# which removes the WAL still needed by the logical slot
+$node->safe_psql('postgres',
+	q{select pg_replication_slot_advance('slot_physical', pg_current_wal_lsn())}
+);
+
+$node->safe_psql('postgres',
+	q{select injection_points_wakeup('checkpoint-before-old-wal-removal')});
+
+# abruptly stop the server (1 second should be enough for the checkpoint
+# to finish, would be better )
+$node->stop('immediate');
+
+$node->start;
+
+eval {
+	$node->safe_psql('postgres',
+		q{select count(*) from pg_logical_slot_get_changes('slot_logical', null, null);}
+	);
+};
+is($@, '', "Logical slot still valid");
+
+done_testing();
diff --git a/src/test/recovery/t/047_physical_slot.pl b/src/test/recovery/t/047_physical_slot.pl
new file mode 100644
index 00000000000..f2cf096b308
--- /dev/null
+++ b/src/test/recovery/t/047_physical_slot.pl
@@ -0,0 +1,136 @@
+# Copyright (c) 2025, PostgreSQL Global Development Group
+#
+# This test verifies the case when the physical slot is advanced during
+# checkpoint. The test checks that the physical slot's restart_lsn still refers
+# to an existed WAL segment after immediate restart.
+#
+# Discussion:
+# https://www.postgresql.org/message-id/flat/1d12d2-67235980-35-19a406a0%4063439497
+#
+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, $result);
+
+$node = PostgreSQL::Test::Cluster->new('mike');
+$node->init();
+$node->append_conf('postgresql.conf',
+	"shared_preload_libraries = 'injection_points'");
+$node->append_conf('postgresql.conf', "wal_level = 'replica'");
+$node->start();
+$node->safe_psql('postgres', q(CREATE EXTENSION injection_points));
+
+# create a simple table to generate data into
+$node->safe_psql('postgres',
+	q{create table t (id serial primary key, b text)});
+
+# create a physical replication slot
+$node->safe_psql('postgres',
+	q{select pg_create_physical_replication_slot('slot_physical', true)});
+
+# advance slot to current position, just to have everything "valid"
+$node->safe_psql('postgres',
+	q{select pg_replication_slot_advance('slot_physical', pg_current_wal_lsn())}
+);
+
+# run checkpoint, to flush current state to disk and set a baseline
+$node->safe_psql('postgres', q{checkpoint});
+
+# insert 2M rows, that's about 260MB (~20 segments) worth of WAL
+$node->safe_psql('postgres',
+	q{insert into t (b) select md5(i::text) from generate_series(1,100000) s(i)}
+);
+
+# advance slot to current position, just to have everything "valid"
+$node->safe_psql('postgres',
+	q{select pg_replication_slot_advance('slot_physical', pg_current_wal_lsn())}
+);
+
+# run another checkpoint, to set a new restore LSN
+$node->safe_psql('postgres', q{checkpoint});
+
+# another 2M rows, that's about 260MB (~20 segments) worth of WAL
+$node->safe_psql('postgres',
+	q{insert into t (b) select md5(i::text) from generate_series(1,1000000) s(i)}
+);
+
+my $restart_lsn_init = $node->safe_psql('postgres',
+	q{select restart_lsn from pg_replication_slots where slot_name = 'slot_physical'}
+);
+chomp($restart_lsn_init);
+note("restart lsn before checkpoint: $restart_lsn_init");
+
+# run another checkpoint, this time in the background, and make it wait
+# on the injection point), so that the checkpoint stops right before
+# removing old WAL segments
+note('starting checkpoint');
+
+my $checkpoint = $node->background_psql('postgres');
+$checkpoint->query_safe(
+	q{select injection_points_attach('checkpoint-before-old-wal-removal','wait')}
+);
+$checkpoint->query_until(
+	qr/starting_checkpoint/,
+	q(\echo starting_checkpoint
+checkpoint;
+\q
+));
+
+# wait until the checkpoint stops right before removing WAL segments
+note('waiting for injection_point');
+$node->wait_for_event('checkpointer', 'checkpoint-before-old-wal-removal');
+note('injection_point is reached');
+
+# OK, we're in the right situation,  time to advance the physical slot,
+# which recalculates the required LSN, and then unblock the checkpoint,
+# which removes the WAL still needed by the logical slot
+$node->safe_psql('postgres',
+	q{select pg_replication_slot_advance('slot_physical', pg_current_wal_lsn())}
+);
+
+# Continue checkpoint
+$node->safe_psql('postgres',
+	q{select injection_points_wakeup('checkpoint-before-old-wal-removal')});
+
+my $restart_lsn_old = $node->safe_psql('postgres',
+	q{select restart_lsn from pg_replication_slots where slot_name = 'slot_physical'}
+);
+chomp($restart_lsn_old);
+note("restart lsn before stop: $restart_lsn_old");
+
+# abruptly stop the server (1 second should be enough for the checkpoint
+# to finish, would be better )
+$node->stop('immediate');
+
+$node->start;
+
+# Get the restart_lsn of the slot right after restarting
+my $restart_lsn = $node->safe_psql('postgres',
+	q{select restart_lsn from pg_replication_slots where slot_name = 'slot_physical'}
+);
+chomp($restart_lsn);
+note("restart lsn: $restart_lsn");
+
+# Get wal segment name for slot's restart_lsn
+my $restart_lsn_segment = $node->safe_psql('postgres',
+	"SELECT pg_walfile_name('$restart_lsn'::pg_lsn)");
+chomp($restart_lsn_segment);
+
+# Check if the required wal segment exists
+note("required by slot segment name: $restart_lsn_segment");
+my $datadir = $node->data_dir;
+ok( -f "$datadir/pg_wal/$restart_lsn_segment",
+	"WAL segment $restart_lsn_segment for physical slot's restart_lsn $restart_lsn exists"
+);
+
+done_testing();
-- 
2.39.5 (Apple Git-154)