Subject: Re: Slow catchup of 2PC (twophase) transactions on replica in LR

From: Amit Kapila <[email-masked]>
Date: Tue, 9 Jul 2024 10:35:37 +0530

On Mon, Jul 8, 2024 at 5:25 PM Amit Kapila <[email-masked]> wrote:
>
>
> I see that in 0003/0004, the patch first aborts pending prepared
> transactions, update's catalog, and then change slot's property via
> walrcv_alter_slot. What if there is any ERROR (say the remote node is
> not reachable or there is an error while updating the catalog) after
> we abort the pending prepared transaction? Won't we end up with lost
> prepared transactions in such a case?
>

Considering the above is a problem the other possibility I thought of
is to change the order like abort prepared xacts after slot update.
That is also dangerous because any failure while aborting could make a
slot change permanent whereas the subscription option will still be
old value. Now, because the slot's two_phase property is off, at
commit, it can resend the entire transaction which can create a
problem because the corresponding prepared transaction will already be
present.

One more thing to think about in this regard is what if we fail after
aborting a few prepared transactions and not all?

At this stage, I am not able to think of a good solution for these
problems. So, if we don't get a solution for these, we can document
that users can first manually abort prepared transactions and then
switch off the two_phase option using Alter Subscription command.

-- 
With Regards,
Amit Kapila.

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Tue, 9 Jul 2024 11:49:38 +0000

Dear Amit,

> > prepared transactions in such a case?

Yes, v16 could happen the case, becasue FinishPreparedTransaction() itself is not
the transactional operation. In below example, the subscription was altered after
stopping the publisher. You could see that prepared transaction were rollbacked.

```
subscriber=# SELECT gid FROM pg_prepared_xacts ;
       gid        
------------------
 pg_gid_16390_741
 pg_gid_16390_742
(2 rows)
subscriber=# ALTER SUBSCRIPTION sub SET (TWO_PHASE = off, FORCE_ALTER = on);
NOTICE:  requested altering to two_phase = false but there are prepared transactions done by the subscription
DETAIL:  Such transactions are being rollbacked.
ERROR:  could not connect to the publisher: connection to server on socket "/tmp/.s.PGSQL.5431" failed: No such file or directory
        Is the server running locally and accepting connections on that socket?
subscriber=# SELECT gid FROM pg_prepared_xacts ;
 gid 
-----
(0 rows)
```

> problem because the corresponding prepared transaction will already be

I feel it is rare case but still possible. E.g., race condition by TwoPhaseStateLock
locking, oom, disk failures and so on.
And since prepared transactions hold locks, duplicated arrival of transactions
may cause table-lock failures. 

> aborting a few prepared transactions and not all?

It's bit hard to emulate, but I imagine part of prepared transactions remains.

> switch off the two_phase option using Alter Subscription command.

I'm also not sure what should we do. Ideally, it may be happy to make
FinishPreparedTransaction() transactional, but not sure it is realistic. So
changes for aborting prepared txns are removed, documentation patch was added
instead.

Here is a summary of updates for patches. Dropping-prepared-transaction patch
was removed for now.

0001 - Codes for SUBOPT_TWOPHASE_COMMIT are moved per requirement [1].
       Also, checks for failover and two_phase are unified into one function.
0002 - updated accordingly. An argument for the check function is added.
0003 - this contains documentation changes required in [2].

[1]: https://www.postgresql.org/message-id/CAA4eK1%2BFRrL_fLWLsWQGHZRESg39ixzDX_S9hU8D7aFtU%2Ba8uQ%40mail.gmail.com
[2]: https://www.postgresql.org/message-id/CAA4eK1Khy_YWFoQ1HOF_tGtiixD8YoTg86coX1-ckxt8vK3U%3DQ%40mail.gmail.com

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Tue, 9 Jul 2024 12:52:55 +0000

> 0003 - this contains documentation changes required in [2].

Previous patch set could not be accepted due to the initialization miss.
PSA new version. 

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/

---

From: "Zhijie Hou (Fujitsu)" <[email-masked]>
Date: Sat, 13 Jul 2024 10:48:55 +0000

> Previous patch set could not be accepted due to the initialization miss.

Thanks for the patches ! I initially reviewed the 0001 and found that
the implementation of ALTER_REPLICATION_SLOT has a issue, e.g.
it doesn't handle the case when there is only one specified option
in the replication command:

ALTER_REPLICATION_SLOT slot (two_phase)

In this case, it always overwrites the un-specified option(failover) to false even
when the failover was set to true. I tried to make a small fix which is on
top of 0001 (please see the attachment).

I also added the doc of the new two_phase option of the replication command
and a missing period of errhint in the topup patch.

Best Regards,
Hou zj
From 3bbaaba53a0cb3db43cc893acbd3ffbedd61bff1 Mon Sep 17 00:00:00 2001
From: Hou Zhijie <[email-masked]>
Date: Sat, 13 Jul 2024 18:31:28 +0800
Subject: [PATCH] fix alter replication slot

---
 doc/src/sgml/protocol.sgml              | 16 ++++++++++++++
 src/backend/commands/subscriptioncmds.c |  2 +-
 src/backend/replication/slot.c          | 16 ++++++++------
 src/backend/replication/walsender.c     | 29 +++++++++++++++----------
 src/include/replication/slot.h          |  4 ++--
 5 files changed, 46 insertions(+), 21 deletions(-)

diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 1b27d0a547..3ac4a4be28 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2206,6 +2206,22 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
        </varlistentry>
       </variablelist>
 
+      <variablelist>
+       <varlistentry>
+        <term><literal>TWO_PHASE [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
+        <listitem>
+         <para>
+          If true, this logical replication slot supports decoding of two-phase
+          commit. With this option, commands related to two-phase commit such as
+          <literal>PREPARE TRANSACTION</literal>, <literal>COMMIT PREPARED</literal>
+          and <literal>ROLLBACK PREPARED</literal> are decoded and transmitted.
+          The transaction will be decoded and transmitted at
+          <literal>PREPARE TRANSACTION</literal> time.
+         </para>
+        </listitem>
+       </varlistentry>
+      </variablelist>
+
      </listitem>
     </varlistentry>
 
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 7604e228e8..c48b6d0549 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1308,7 +1308,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 						ereport(ERROR,
 								(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
 								 errmsg("cannot disable two_phase when uncommitted prepared transactions present"),
-								 errhint("Resolve these transactions and try again")));
+								 errhint("Resolve these transactions and try again.")));
 
 					/* Change system catalog acoordingly */
 					values[Anum_pg_subscription_subtwophasestate - 1] =
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 2ad6dca993..2f167a2adc 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -804,11 +804,12 @@ ReplicationSlotDrop(const char *name, bool nowait)
  * Change the definition of the slot identified by the specified name.
  */
 void
-ReplicationSlotAlter(const char *name, bool failover, bool two_phase)
+ReplicationSlotAlter(const char *name, bool *failover, bool *two_phase)
 {
 	bool		update_slot = false;
 
 	Assert(MyReplicationSlot == NULL);
+	Assert(failover || two_phase);
 
 	ReplicationSlotAcquire(name, false);
 
@@ -834,7 +835,7 @@ ReplicationSlotAlter(const char *name, bool failover, bool two_phase)
 		 * Do not allow users to enable failover on the standby as we do not
 		 * support sync to the cascading standby.
 		 */
-		if (failover)
+		if (failover && *failover)
 			ereport(ERROR,
 					errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 					errmsg("cannot enable failover for a replication slot"
@@ -845,24 +846,25 @@ ReplicationSlotAlter(const char *name, bool failover, bool two_phase)
 	 * Do not allow users to enable failover for temporary slots as we do not
 	 * support syncing temporary slots to the standby.
 	 */
-	if (failover && MyReplicationSlot->data.persistency == RS_TEMPORARY)
+	if (failover && *failover &&
+		MyReplicationSlot->data.persistency == RS_TEMPORARY)
 		ereport(ERROR,
 				errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				errmsg("cannot enable failover for a temporary replication slot"));
 
-	if (MyReplicationSlot->data.failover != failover)
+	if (failover && MyReplicationSlot->data.failover != *failover)
 	{
 		SpinLockAcquire(&MyReplicationSlot->mutex);
-		MyReplicationSlot->data.failover = failover;
+		MyReplicationSlot->data.failover = *failover;
 		SpinLockRelease(&MyReplicationSlot->mutex);
 
 		update_slot = true;
 	}
 
-	if (MyReplicationSlot->data.two_phase != two_phase)
+	if (two_phase && MyReplicationSlot->data.two_phase != *two_phase)
 	{
 		SpinLockAcquire(&MyReplicationSlot->mutex);
-		MyReplicationSlot->data.two_phase = two_phase;
+		MyReplicationSlot->data.two_phase = *two_phase;
 		SpinLockRelease(&MyReplicationSlot->mutex);
 
 		update_slot = true;
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 5224ea6c2c..f3b5068d95 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1411,30 +1411,31 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd)
  */
 static void
 ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd,
-						  bool *failover, bool *two_phase)
+						  bool *failover_given, bool *failover,
+						  bool *two_phase_given, bool *two_phase)
 {
-	bool		failover_given = false;
-	bool		two_phase_given = false;
+	*failover_given = false;
+	*two_phase_given = false;
 
 	/* Parse options */
 	foreach_ptr(DefElem, defel, cmd->options)
 	{
 		if (strcmp(defel->defname, "failover") == 0)
 		{
-			if (failover_given)
+			if (*failover_given)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
-			failover_given = true;
+			*failover_given = true;
 			*failover = defGetBoolean(defel);
 		}
 		else if (strcmp(defel->defname, "two_phase") == 0)
 		{
-			if (two_phase_given)
+			if (*two_phase_given)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
-			two_phase_given = true;
+			*two_phase_given = true;
 			*two_phase = defGetBoolean(defel);
 		}
 		else
@@ -1448,11 +1449,17 @@ ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd,
 static void
 AlterReplicationSlot(AlterReplicationSlotCmd *cmd)
 {
-	bool		failover = false;
-	bool		two_phase = false;
+	bool		failover_given;
+	bool		two_phase_given;
+	bool		failover;
+	bool		two_phase;
+
+	ParseAlterReplSlotOptions(cmd, &failover_given, &failover,
+							  &two_phase_given, &two_phase);
 
-	ParseAlterReplSlotOptions(cmd, &failover, &two_phase);
-	ReplicationSlotAlter(cmd->slotname, failover, two_phase);
+	ReplicationSlotAlter(cmd->slotname,
+						 failover_given ? &failover : NULL,
+						 two_phase_given ? &two_phase : NULL);
 }
 
 /*
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index 163a4a911a..cde164472a 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -243,8 +243,8 @@ extern void ReplicationSlotCreate(const char *name, bool db_specific,
 extern void ReplicationSlotPersist(void);
 extern void ReplicationSlotDrop(const char *name, bool nowait);
 extern void ReplicationSlotDropAcquired(void);
-extern void ReplicationSlotAlter(const char *name, bool failover,
-								 bool two_phase);
+extern void ReplicationSlotAlter(const char *name, bool *failover,
+								 bool *two_phase);
 
 extern void ReplicationSlotAcquire(const char *name, bool nowait);
 extern void ReplicationSlotRelease(void);
-- 
2.30.0.windows.2

---

From: Amit Kapila <[email-masked]>
Date: Mon, 15 Jul 2024 17:09:10 +0530

> Previous patch set could not be accepted due to the initialization miss.

Few minor comments:
=================
0001-patch
1.
.git/rebase-apply/patch:253: space before tab in indent.

errmsg("slot_name and two_phase cannot be altered at the same
time")));
warning: 1 line adds whitespace errors.

White space issue as shown by git am command.

2.
+/*
+ * Common checks for altering failover and two_phase option
+ */
+static void
+CommonChecksForFailoverAndTwophase(Subscription *sub, const char *option,
+    bool isTopLevel)

The function name looks odd to me. How about something along the lines
of CheckAlterSubOption()?

3.
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot disable two_phase when uncommitted prepared
transactions present"),

We can slightly change the above error message to: "cannot disable
two_phase when prepared transactions are present".

0003-patch
Alter the altering from
+      <literal>true</literal> to <literal>false</literal>, the publisher will
+      replicate transactions again when they are committed.

The beginning of the sentence sounds awkward.

-- 
With Regards,
Amit Kapila.

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Tue, 16 Jul 2024 05:17:06 +0000

Dear Amit, Hou,

Thanks for giving comments! PSA new versions.
What's new:

0001: included Hou's patch [1] not to overwrite slot options.
      Some other comments were also addressed.
0002: not so changed, just rebased.
0003: Typo was fixed, s/Alter/After/.

[1]: https://www.postgresql.org/message-id/OS3PR01MB57184E0995521300AC06CB4B94A72%40OS3PR01MB5718.jpnprd01.prod.outlook.com 

Best regards,
Hayato Kuroda
FUJITSU LIMITED

---

From: Peter Smith <[email-masked]>
Date: Tue, 16 Jul 2024 20:02:21 +1000

Hi, here are some review comments for patch v18-0001.

======
doc/src/sgml/protocol.sgml

nitpick - Although it is no fault of your patch, IMO it would be nicer for
the TWO_PHASE description (of CREATE REPLICATION SLOT) to also be in the
same consistent order as what you have (e.g. below FAILOVER). So I moved it.

======
src/backend/access/transam/twophase.c

LookupGXactBySubid:
nitpick - add a blank line before return

======
src/backend/commands/subscriptioncmds.c

CommonChecksForFailoverAndTwophase:
nitpick - added Assert for the generic-looking "option" parameter name
nitpick - modified comment about transaction block

~~~

1. AlterSubscription
+ * Workers may still survive even if the subscription has
+ * been disabled. They may read the pg_subscription
+ * catalog and detect that the twophase parameter is
+ * updated, which causes the assertion failure. Ensure
+ * workers have already been exited to avoid it.

"which causes the assertion failure" -- what assertion failure is that? The
comment is not very clear.

~

nitpick - in comment /twophase/two_phase/
nitpick - typo /acoordingly/accordingly/

======
src/backend/replication/logical/launcher.c

logicalrep_workers_find:
nitpick - /require_lock/acquire_lock/
nitpick - take the Assert out of the else.

======
src/backend/replication/slot.c

nitpick - refactor the code to check (failover) only one time. See the
nitpicks attachment.

~

2. ParseAlterReplSlotOptions

nitpick -- IMO the ParseAlterReplSlotOptions(). function does more harm
than good here by adding the unnecessary complexity of messing around with
multiple parameters that are passed-by-reference. All this would be simpler
if it was just coded inline in the AlterReplicationSlot() function, which
is the only caller. I've refactored all this to demonstrate (see nitpicks
attachment)

======
src/include/replication/worker_internal.h

nitpick - /require_lock/acquire_lock/

======
src/test/regress/sql/subscription.sql

nitpick - tweak comments

======
src/test/subscription/t/021_twophase.pl

nitpick - change comment style to indicate each test part better.

======
99.
Please also see the attached diffs patch which implements any nitpicks
mentioned above.

======
Kind Regards,
Peter Smith.
Fujitsu Australia
diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index 3ac4a4b..cba6661 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2050,21 +2050,6 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
 
       <variablelist>
        <varlistentry>
-        <term><literal>TWO_PHASE [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
-        <listitem>
-         <para>
-          If true, this logical replication slot supports decoding of two-phase
-          commit. With this option, commands related to two-phase commit such as
-          <literal>PREPARE TRANSACTION</literal>, <literal>COMMIT PREPARED</literal>
-          and <literal>ROLLBACK PREPARED</literal> are decoded and transmitted.
-          The transaction will be decoded and transmitted at
-          <literal>PREPARE TRANSACTION</literal> time.
-          The default is false.
-         </para>
-        </listitem>
-       </varlistentry>
-
-       <varlistentry>
         <term><literal>RESERVE_WAL [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
         <listitem>
          <para>
@@ -2104,6 +2089,21 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
          </para>
         </listitem>
        </varlistentry>
+
+       <varlistentry>
+        <term><literal>TWO_PHASE [ <replaceable class="parameter">boolean</replaceable> ]</literal></term>
+        <listitem>
+         <para>
+          If true, this logical replication slot supports decoding of two-phase
+          commit. With this option, commands related to two-phase commit such as
+          <literal>PREPARE TRANSACTION</literal>, <literal>COMMIT PREPARED</literal>
+          and <literal>ROLLBACK PREPARED</literal> are decoded and transmitted.
+          The transaction will be decoded and transmitted at
+          <literal>PREPARE TRANSACTION</literal> time.
+          The default is false.
+         </para>
+        </listitem>
+       </varlistentry>
       </variablelist>
 
       <para>
diff --git a/src/backend/access/transam/twophase.c b/src/backend/access/transam/twophase.c
index 35bce68..f3c6e1f 100644
--- a/src/backend/access/transam/twophase.c
+++ b/src/backend/access/transam/twophase.c
@@ -2741,5 +2741,6 @@ LookupGXactBySubid(Oid subid)
 		}
 	}
 	LWLockRelease(TwoPhaseStateLock);
+
 	return found;
 }
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 6995a62..3703cf6 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1076,6 +1076,8 @@ CheckAlterSubOption(Subscription *sub, const char *option, bool isTopLevel)
 {
 	StringInfoData cmd;
 
+	Assert(strstr("two_phase,failover", option));
+
 	if (!sub->slotname)
 		ereport(ERROR,
 				(errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
@@ -1098,8 +1100,8 @@ CheckAlterSubOption(Subscription *sub, const char *option, bool isTopLevel)
 	appendStringInfo(&cmd, "ALTER SUBSCRIPTION ... SET (%s)", option);
 
 	/*
-	 * The changed option of the slot can't be rolled back: prevent we are in
-	 * the transaction state.
+	 * The changed option of the slot can't be rolled back, so disallow if we
+	 * are in a transaction block.
 	 */
 	PreventInTransactionBlock(isTopLevel, cmd.data);
 
@@ -1282,7 +1284,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 					/*
 					 * Workers may still survive even if the subscription has
 					 * been disabled. They may read the pg_subscription
-					 * catalog and detect that the twophase parameter is
+					 * catalog and detect that the two_phase parameter is
 					 * updated, which causes the assertion failure. Ensure
 					 * workers have already been exited to avoid it.
 					 */
@@ -1304,7 +1306,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 								 errmsg("cannot disable two_phase when prepared transactions are present"),
 								 errhint("Resolve these transactions and try again.")));
 
-					/* Change system catalog acoordingly */
+					/* Change system catalog accordingly */
 					values[Anum_pg_subscription_subtwophasestate - 1] =
 						CharGetDatum(opts.twophase ?
 									 LOGICALREP_TWOPHASE_STATE_PENDING :
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 45744b7..c566d50 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -272,15 +272,15 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
  * the subscription, instead of just one.
  */
 List *
-logicalrep_workers_find(Oid subid, bool only_running, bool require_lock)
+logicalrep_workers_find(Oid subid, bool only_running, bool acquire_lock)
 {
 	int			i;
 	List	   *res = NIL;
 
-	if (require_lock)
+	if (acquire_lock)
 		LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
-	else
-		Assert(LWLockHeldByMe(LogicalRepWorkerLock));
+
+	Assert(LWLockHeldByMe(LogicalRepWorkerLock));
 
 	/* Search for attached worker for a given subscription id. */
 	for (i = 0; i < max_logical_replication_workers; i++)
@@ -291,7 +291,7 @@ logicalrep_workers_find(Oid subid, bool only_running, bool require_lock)
 			res = lappend(res, w);
 	}
 
-	if (require_lock)
+	if (acquire_lock)
 		LWLockRelease(LogicalRepWorkerLock);
 
 	return res;
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 2f167a2..e75f24b 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -842,23 +842,25 @@ ReplicationSlotAlter(const char *name, bool *failover, bool *two_phase)
 						   " on the standby"));
 	}
 
-	/*
-	 * Do not allow users to enable failover for temporary slots as we do not
-	 * support syncing temporary slots to the standby.
-	 */
-	if (failover && *failover &&
-		MyReplicationSlot->data.persistency == RS_TEMPORARY)
+	if (failover)
+	{
+		/*
+		 * Do not allow users to enable failover for temporary slots as we do not
+		 * support syncing temporary slots to the standby.
+		 */
+		if (*failover && MyReplicationSlot->data.persistency == RS_TEMPORARY)
 		ereport(ERROR,
 				errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
 				errmsg("cannot enable failover for a temporary replication slot"));
 
-	if (failover && MyReplicationSlot->data.failover != *failover)
-	{
-		SpinLockAcquire(&MyReplicationSlot->mutex);
-		MyReplicationSlot->data.failover = *failover;
-		SpinLockRelease(&MyReplicationSlot->mutex);
+		if (MyReplicationSlot->data.failover != *failover)
+		{
+			SpinLockAcquire(&MyReplicationSlot->mutex);
+			MyReplicationSlot->data.failover = *failover;
+			SpinLockRelease(&MyReplicationSlot->mutex);
 
-		update_slot = true;
+			update_slot = true;
+		}
 	}
 
 	if (two_phase && MyReplicationSlot->data.two_phase != *two_phase)
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 04f65e0..af8e958 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1405,56 +1405,42 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd)
 	ReplicationSlotDrop(cmd->slotname, !cmd->wait);
 }
 
+
 /*
- * Process extra options given to ALTER_REPLICATION_SLOT.
+ * Change the definition of a replication slot.
  */
 static void
-ParseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd,
-						  bool *failover_given, bool *failover,
-						  bool *two_phase_given, bool *two_phase)
+AlterReplicationSlot(AlterReplicationSlotCmd *cmd)
 {
-	*failover_given = false;
-	*two_phase_given = false;
+	bool		failover_given = false;
+	bool		two_phase_given = false;
+	bool		failover;
+	bool		two_phase;
 
 	/* Parse options */
 	foreach_ptr(DefElem, defel, cmd->options)
 	{
 		if (strcmp(defel->defname, "failover") == 0)
 		{
-			if (*failover_given)
+			if (failover_given)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
-			*failover_given = true;
-			*failover = defGetBoolean(defel);
+			failover_given = true;
+			failover = defGetBoolean(defel);
 		}
 		else if (strcmp(defel->defname, "two_phase") == 0)
 		{
-			if (*two_phase_given)
+			if (two_phase_given)
 				ereport(ERROR,
 						(errcode(ERRCODE_SYNTAX_ERROR),
 						 errmsg("conflicting or redundant options")));
-			*two_phase_given = true;
-			*two_phase = defGetBoolean(defel);
+			two_phase_given = true;
+			two_phase = defGetBoolean(defel);
 		}
 		else
 			elog(ERROR, "unrecognized option: %s", defel->defname);
 	}
-}
-
-/*
- * Change the definition of a replication slot.
- */
-static void
-AlterReplicationSlot(AlterReplicationSlotCmd *cmd)
-{
-	bool		failover_given;
-	bool		two_phase_given;
-	bool		failover;
-	bool		two_phase;
-
-	ParseAlterReplSlotOptions(cmd, &failover_given, &failover,
-							  &two_phase_given, &two_phase);
 
 	ReplicationSlotAlter(cmd->slotname,
 						 failover_given ? &failover : NULL,
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 990f524..9646261 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -241,7 +241,7 @@ extern void logicalrep_worker_attach(int slot);
 extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
 												bool only_running);
 extern List *logicalrep_workers_find(Oid subid, bool only_running,
-									 bool require_lock);
+									 bool acquire_lock);
 extern bool logicalrep_worker_launch(LogicalRepWorkerType wtype,
 									 Oid dbid, Oid subid, const char *subname,
 									 Oid userid, Oid relid,
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index 51fa4b9..40e1a07 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -377,7 +377,7 @@ HINT:  To initiate replication, you must manually create the replication slot, e
  regress_testsub | regress_subscription_user | f       | {testpub}   | f      | off       | p                | f                | any    | t                 | f             | f        | off                | dbname=regress_doesnotexist | 0/0
 (1 row)
 
--- We can alter streaming when two_phase enabled
+-- we can alter streaming when two_phase is enabled
 ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
 \dRs+
                                                                                                                 List of subscriptions
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index a3886d7..b64f419 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -256,7 +256,7 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
 CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, two_phase = true);
 
 \dRs+
--- We can alter streaming when two_phase enabled
+-- we can alter streaming when two_phase is enabled
 ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
 
 \dRs+
diff --git a/src/test/subscription/t/021_twophase.pl b/src/test/subscription/t/021_twophase.pl
index 4e8f627..66265c7 100644
--- a/src/test/subscription/t/021_twophase.pl
+++ b/src/test/subscription/t/021_twophase.pl
@@ -371,8 +371,8 @@ is($result, qq(2), 'replicated data in subscriber table');
 $node_subscriber->safe_psql('postgres', "DROP SUBSCRIPTION tap_sub");
 
 ###############################
-# Disable the subscription and alter it to two_phase = false,
-# then verify that the altered subscription reflects the two_phase option.
+# Alter the subscription to two_phase = false.
+# Verify that the altered subscription reflects the two_phase option.
 ###############################
 
 # Alter subscription two_phase to false
@@ -395,7 +395,10 @@ $result = $node_subscriber->safe_psql('postgres',
 );
 is($result, qq(d), 'two-phase should be disabled');
 
-# Now do a prepare on the publisher and make sure that it is not replicated.
+###############################
+# Now do a prepare on the publisher.
+# Verify that it is not replicated.
+###############################
 $node_publisher->safe_psql(
 	'postgres', qq{
     BEGIN;
@@ -411,7 +414,10 @@ $result = $node_subscriber->safe_psql('postgres',
 	"SELECT count(*) FROM pg_prepared_xacts;");
 is($result, qq(0), 'should be no prepared transactions on subscriber');
 
-# Now commit the insert and verify that it is replicated
+###############################
+# Now commit the insert.
+# Verify that it is replicated.
+###############################
 $node_publisher->safe_psql('postgres', "COMMIT PREPARED 'newgid';");
 
 # Wait for the subscriber to catchup
@@ -422,7 +428,10 @@ $result =
   $node_subscriber->safe_psql('postgres', "SELECT count(*) FROM tab_copy;");
 is($result, qq(3), 'replicated data in subscriber table');
 
-# Alter subscription two_phase to true
+###############################
+# Alter the subscription to two_phase = true.
+# Verify that the altered subscription reflects the two_phase option.
+###############################
 $node_subscriber->safe_psql('postgres',
     "ALTER SUBSCRIPTION tap_sub_copy DISABLE;");
 $node_subscriber->poll_query_until('postgres',

---

From: Peter Smith <[email-masked]>
Date: Wed, 17 Jul 2024 13:31:14 +1000

Here are some review comments for patch v18-0002.

======

1. CheckAlterSubOption

1a.
It's not obvious why we are only checking the 'slot name' when
needs_update==true, but OTOH is always checking the  'enabled' state.

~

1b.
Param 'needs_update' is a vague name. It needs more explanatory comments or
a better name. e.g. First impression was "Why are we calling 'Alter'
function if needs_update is false?". I know it encapsulates some common
code, but if special cases cause the logic to be more confusing then that
cost may outweigh the benefit of this function.

~

1c.
If the error checks can be moved to be done up-front, then all the
'needs_update' can be combined. Avoiding multiple checks to 'needs_update'
will make this function simpler.

~~~

AlterSubscription:
nitpick - typo /needs/need/
nitpick - typo /wo_phase/two_phase/
nitpick - The comment wording "the later part...", was confusing. I've
reworded the whole comment. But this belongs in patch 0001.

======

nitpick - Use the same "###############################" comment style as
in patch 0001 to indicate each main TEST scenario.

======
99.
Please refer to the diffs attachment patch, which implements any nitpicks
mentioned above.

======
Kind Regards,
Peter Smith.
Fujitsu Australia
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index d80d60c..5c6f0a5 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1262,7 +1262,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 				if (IsSet(opts.specified_opts, SUBOPT_FAILOVER))
 				{
 					/*
-					 * First mark the needs to alter the replication slot.
+					 * First, mark the need to alter the replication slot.
 					 * Failover option is controlled by both the publisher (as
 					 * a slot option) and the subscriber (as a subscription
 					 * option).
@@ -1280,7 +1280,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 				if (IsSet(opts.specified_opts, SUBOPT_TWOPHASE_COMMIT))
 				{
 					/*
-					 * First check the need to alter the replication slot.
+					 * First, check the need to alter the replication slot.
 					 * Two_phase option is controlled by both the publisher
 					 * (as a slot option) and the subscriber (as a
 					 * subscription option). The slot option must be altered
@@ -1302,11 +1302,9 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 										isTopLevel);
 
 					/*
-					 * If the wo_phase slot option must be altered, this
-					 * cannot be altered with slot_name simultaneously. The
-					 * latter part refers to the pre-set slot name and tries
-					 * to modify the slot option, so changing both does not
-					 * make sense.
+					 * Modifying the two_phase slot option requires a slot
+					 * lookup by slot name, so changing the slot name
+					 * at the same time is not allowed.
 					 */
 					if (update_two_phase &&
 						IsSet(opts.specified_opts, SUBOPT_SLOT_NAME))
diff --git a/src/test/subscription/t/021_twophase.pl b/src/test/subscription/t/021_twophase.pl
index f56dff4..26b9e2c 100644
--- a/src/test/subscription/t/021_twophase.pl
+++ b/src/test/subscription/t/021_twophase.pl
@@ -423,9 +423,12 @@ $result = $node_subscriber->safe_psql('postgres',
 	"SELECT count(*) FROM pg_prepared_xacts;");
 is($result, qq(0), 'should be no prepared transactions on subscriber');
 
-# Toggle the two_phase to "true" *before* the COMMIT PREPARED. Since we are the
-# special path for the case where both two_phase and failover are altered, it
-# is also set to "true".
+###############################
+# Toggle the two_phase to "true" before the COMMIT PREPARED.
+#
+# Since we are the special path for the case where both two_phase
+# and failover are altered, it is also set to "true".
+###############################
 $node_subscriber->safe_psql('postgres',
     "ALTER SUBSCRIPTION tap_sub_copy DISABLE;");
 $node_subscriber->poll_query_until('postgres',