master_v2-0001-Fix-possibility-of-logical-decoding-partial-trans.patch

From 6ce61c863e8a6aab530f56ac984d6adfc3c0442a Mon Sep 17 00:00:00 2001
From: Masahiko Sawada <sawada.mshk@gmail.com>
Date: Fri, 5 Jul 2024 10:29:09 +0900
Subject: [PATCH v2] Fix possibility of logical decoding partial transaction
 changes.

When creating and initializing a logical slot, the restart_lsn is set
to the latest WAL insertion point (or the latest replay point on
standbys). Subsequently, WAL records are decoded from that point to
find the start point for extracting changes in the
DecodingContextFindStartpoint() function. Since the initial
restart_lsn could be in the middle of a transaction, the start point
must be a consistent point where we won't see the data for partial
transactions.

Previously, when not building a full snapshot, serialized snapshots
were restored, and the SnapBuild jumps to the consistent state even
while finding the start point. Consequently, the slot's restart_lsn
and confirmed_flush could be set to the middle of a transaction. This
could lead to various unexpected consequences. Specifically, there
were reports of logical decoding decoding partial transactions, and
assertion failures occurred because only subtransactions were decoded
without decoding their top-level transaction until decoding the commit
record.

To resolve this issue, the changes prevent restoring the serialized
snapshot and jumping to the consistent state while finding the start
point.

On v17 and HEAD, a flag indicating whether snapshot restores should be
skipped has been added to the SnapBuild struct. This flag is set only
while finding the start point, and SNAPBUILD_VERSION has been bumpded
due to changes in the SnapBuild struct.

On backbranches, the flag is stored in the LogicalDecodingContext
instead, preserving on-disk compatibility. The flag is set only when
initializing the LogicalDecodingContext. Therefore, the context used
for finding the start point must be the same as the one used for
initializing the slot, which was not strictly requried before but is
not a practical issue.

Backpatch to all supported versions.

Reported-by: Drew Callahan
Reviewed-by: Amit Kapila, Hayato Kuroda
Discussion: https://postgr.es/m/2444AA15-D21B-4CCE-8052-52C7C2DAFE5C%40amazon.com
Backpatch-through: 12
---
 contrib/test_decoding/Makefile                |  3 +-
 .../expected/skip_snapshot_restore.out        | 45 ++++++++++++++++++
 contrib/test_decoding/meson.build             |  1 +
 .../specs/skip_snapshot_restore.spec          | 46 +++++++++++++++++++
 src/backend/replication/logical/logical.c     |  6 +++
 src/backend/replication/logical/snapbuild.c   | 32 ++++++++++---
 src/include/replication/snapbuild.h           |  1 +
 7 files changed, 127 insertions(+), 7 deletions(-)
 create mode 100644 contrib/test_decoding/expected/skip_snapshot_restore.out
 create mode 100644 contrib/test_decoding/specs/skip_snapshot_restore.spec

diff --git a/contrib/test_decoding/Makefile b/contrib/test_decoding/Makefile
index c7ce603706..a4ba1a509a 100644
--- a/contrib/test_decoding/Makefile
+++ b/contrib/test_decoding/Makefile
@@ -8,7 +8,8 @@ REGRESS = ddl xact rewrite toast permissions decoding_in_xact \
 	spill slot truncate stream stats twophase twophase_stream
 ISOLATION = mxact delayed_startup ondisk_startup concurrent_ddl_dml \
 	oldest_xmin snapshot_transfer subxact_without_top concurrent_stream \
-	twophase_snapshot slot_creation_error catalog_change_snapshot
+	twophase_snapshot slot_creation_error catalog_change_snapshot \
+	skip_snapshot_restore
 
 REGRESS_OPTS = --temp-config $(top_srcdir)/contrib/test_decoding/logical.conf
 ISOLATION_OPTS = --temp-config $(top_srcdir)/contrib/test_decoding/logical.conf
diff --git a/contrib/test_decoding/expected/skip_snapshot_restore.out b/contrib/test_decoding/expected/skip_snapshot_restore.out
new file mode 100644
index 0000000000..c64dbd9c4e
--- /dev/null
+++ b/contrib/test_decoding/expected/skip_snapshot_restore.out
@@ -0,0 +1,45 @@
+Parsed test spec with 3 sessions
+
+starting permutation: s0_init s0_begin s0_insert1 s1_init s2_checkpoint s2_get_changes_slot0 s0_insert2 s0_commit s1_get_changes_slot0 s1_get_changes_slot1
+step s0_init: SELECT 'init' FROM pg_create_logical_replication_slot('slot0', 'test_decoding');
+?column?
+--------
+init    
+(1 row)
+
+step s0_begin: BEGIN;
+step s0_insert1: INSERT INTO tbl VALUES (1);
+step s1_init: SELECT 'init' FROM pg_create_logical_replication_slot('slot1', 'test_decoding'); <waiting ...>
+step s2_checkpoint: CHECKPOINT;
+step s2_get_changes_slot0: SELECT data FROM pg_logical_slot_get_changes('slot0', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0');
+data
+----
+(0 rows)
+
+step s0_insert2: INSERT INTO tbl VALUES (2);
+step s0_commit: COMMIT;
+step s1_init: <... completed>
+?column?
+--------
+init    
+(1 row)
+
+step s1_get_changes_slot0: SELECT data FROM pg_logical_slot_get_changes('slot0', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0');
+data                                     
+-----------------------------------------
+BEGIN                                    
+table public.tbl: INSERT: val1[integer]:1
+table public.tbl: INSERT: val1[integer]:2
+COMMIT                                   
+(4 rows)
+
+step s1_get_changes_slot1: SELECT data FROM pg_logical_slot_get_changes('slot1', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0');
+data
+----
+(0 rows)
+
+?column?
+--------
+stop    
+(1 row)
+
diff --git a/contrib/test_decoding/meson.build b/contrib/test_decoding/meson.build
index f1548c0faf..f643dc81a2 100644
--- a/contrib/test_decoding/meson.build
+++ b/contrib/test_decoding/meson.build
@@ -62,6 +62,7 @@ tests += {
       'concurrent_stream',
       'twophase_snapshot',
       'slot_creation_error',
+      'skip_snapshot_restore',
     ],
     'regress_args': [
       '--temp-config', files('logical.conf'),
diff --git a/contrib/test_decoding/specs/skip_snapshot_restore.spec b/contrib/test_decoding/specs/skip_snapshot_restore.spec
new file mode 100644
index 0000000000..3f1fb6f02c
--- /dev/null
+++ b/contrib/test_decoding/specs/skip_snapshot_restore.spec
@@ -0,0 +1,46 @@
+# Test that a slot creation skips to restore serialized snapshot to reach
+# the consistent state.
+
+setup
+{
+    DROP TABLE IF EXISTS tbl;
+    CREATE TABLE tbl (val1 integer);
+}
+
+teardown
+{
+    DROP TABLE tbl;
+    SELECT 'stop' FROM pg_drop_replication_slot('slot0');
+    SELECT 'stop' FROM pg_drop_replication_slot('slot1');
+}
+
+session "s0"
+setup { SET synchronous_commit = on; }
+step "s0_init" { SELECT 'init' FROM pg_create_logical_replication_slot('slot0', 'test_decoding'); }
+step "s0_begin" { BEGIN; }
+step "s0_insert1" { INSERT INTO tbl VALUES (1); }
+step "s0_insert2" { INSERT INTO tbl VALUES (2); }
+step "s0_commit" { COMMIT; }
+
+session "s1"
+setup { SET synchronous_commit = on; }
+step "s1_init" { SELECT 'init' FROM pg_create_logical_replication_slot('slot1', 'test_decoding'); }
+step "s1_get_changes_slot0" { SELECT data FROM pg_logical_slot_get_changes('slot0', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0'); }
+step "s1_get_changes_slot1" { SELECT data FROM pg_logical_slot_get_changes('slot1', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0'); }
+
+session "s2"
+setup { SET synchronous_commit = on ;}
+step "s2_checkpoint" { CHECKPOINT; }
+step "s2_get_changes_slot0" { SELECT data FROM pg_logical_slot_get_changes('slot0', NULL, NULL, 'skip-empty-xacts', '1', 'include-xids', '0'); }
+
+
+# While 'slot1' creation by "s1_init" waits for s0-transaction to commit, the
+# RUNNING_XACTS record is written by "s2_checkpoint" and "s2_get_changes_slot1"
+# serializes consistent snapshots to the disk at LSNs where are before
+# s0-transaction's commit. After s0-transaction commits, "s1_init" resumes but
+# must not restore any serialized snapshots and will reach the consistent state
+# when decoding a RUNNING_XACT record generated after s0-transaction's commit.
+# We check if the get_changes on 'slot1' will not return any s0-transaction's
+# changes as its confirmed_flush_lsn will be after the s0-transaction's commit
+# record.
+permutation "s0_init" "s0_begin" "s0_insert1" "s1_init" "s2_checkpoint" "s2_get_changes_slot0" "s0_insert2" "s0_commit" "s1_get_changes_slot0" "s1_get_changes_slot1"
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 99f31849bb..eb32bc9e14 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -650,6 +650,9 @@ DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
 {
 	ReplicationSlot *slot = ctx->slot;
 
+	/* Let snapshot builder start to find the start point */
+	SnapBuildSetFindStartPoint(ctx->snapshot_builder, true);
+
 	/* Initialize from where to start reading WAL. */
 	XLogBeginRead(ctx->reader, slot->data.restart_lsn);
 
@@ -683,6 +686,9 @@ DecodingContextFindStartpoint(LogicalDecodingContext *ctx)
 	if (slot->data.two_phase)
 		slot->data.two_phase_at = ctx->reader->EndRecPtr;
 	SpinLockRelease(&slot->mutex);
+
+	/* Complete to find the start point */
+	SnapBuildSetFindStartPoint(ctx->snapshot_builder, false);
 }
 
 /*
diff --git a/src/backend/replication/logical/snapbuild.c b/src/backend/replication/logical/snapbuild.c
index e37e22f441..4591df9c89 100644
--- a/src/backend/replication/logical/snapbuild.c
+++ b/src/backend/replication/logical/snapbuild.c
@@ -189,6 +189,9 @@ struct SnapBuild
 	/* Indicates if we are building full snapshot or just catalog one. */
 	bool		building_full_snapshot;
 
+	/* Indicates if we are finding the start point to extract changes */
+	bool		finding_start_point;
+
 	/*
 	 * Snapshot that's valid to see the catalog state seen at this moment.
 	 */
@@ -435,6 +438,15 @@ SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr)
 	return ptr < builder->start_decoding_at;
 }
 
+/*
+ * Let the snapshot builder start finding the start point.
+ */
+void
+SnapBuildSetFindStartPoint(SnapBuild *builder, bool find_start_point)
+{
+	builder->finding_start_point = find_start_point;
+}
+
 /*
  * Increase refcount of a snapshot.
  *
@@ -1327,10 +1339,13 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
 	 *	  state while waiting on c)'s sub-states.
 	 *
 	 * b) This (in a previous run) or another decoding slot serialized a
-	 *	  snapshot to disk that we can use.  Can't use this method for the
-	 *	  initial snapshot when slot is being created and needs full snapshot
-	 *	  for export or direct use, as that snapshot will only contain catalog
-	 *	  modifying transactions.
+	 *	  snapshot to disk that we can use. Can't use this method while
+	 *	  finding the start point for decoding changes. Since in this case
+	 *	  the restart LSN could be in the middle of transactions we need to
+	 *	  find the start point where we won't see the data for partial
+	 *	  transactions. Also, we cannot use this method when a slot needs a
+	 *	  full snapshot for export or direct use, as that snapshot will only
+	 *	  contain catalog modifying transactions.
 	 *
 	 * c) First incrementally build a snapshot for catalog tuples
 	 *	  (BUILDING_SNAPSHOT), that requires all, already in-progress,
@@ -1395,8 +1410,13 @@ SnapBuildFindSnapshot(SnapBuild *builder, XLogRecPtr lsn, xl_running_xacts *runn
 
 		return false;
 	}
-	/* b) valid on disk state and not building full snapshot */
+
+	/*
+	 * b) valid on disk state and while neither building full snapshot nor
+	 * finding the start point.
+	 */
 	else if (!builder->building_full_snapshot &&
+			 !builder->finding_start_point &&
 			 SnapBuildRestore(builder, lsn))
 	{
 		/* there won't be any state to cleanup */
@@ -1580,7 +1600,7 @@ typedef struct SnapBuildOnDisk
 	offsetof(SnapBuildOnDisk, version)
 
 #define SNAPBUILD_MAGIC 0x51A1E001
-#define SNAPBUILD_VERSION 5
+#define SNAPBUILD_VERSION 6
 
 /*
  * Store/Load a snapshot from disk, depending on the snapshot builder's state.
diff --git a/src/include/replication/snapbuild.h b/src/include/replication/snapbuild.h
index a3360a1c5e..d44af34ba0 100644
--- a/src/include/replication/snapbuild.h
+++ b/src/include/replication/snapbuild.h
@@ -76,6 +76,7 @@ extern SnapBuildState SnapBuildCurrentState(SnapBuild *builder);
 extern Snapshot SnapBuildGetOrBuildSnapshot(SnapBuild *builder);
 
 extern bool SnapBuildXactNeedsSkip(SnapBuild *builder, XLogRecPtr ptr);
+extern void SnapBuildSetFindStartPoint(SnapBuild *builder, bool find_start_point);
 extern XLogRecPtr SnapBuildGetTwoPhaseAt(SnapBuild *builder);
 extern void SnapBuildSetTwoPhaseAt(SnapBuild *builder, XLogRecPtr ptr);
 
-- 
2.39.3