Add max_retention_duration option to subscriptions.

This commit introduces a new subscription parameter,
max_retention_duration, aimed at mitigating excessive accumulation of dead
tuples when retain_dead_tuples is enabled and the apply worker lags behind
the publisher.

When the time spent advancing a non-removable transaction ID exceeds the
max_retention_duration threshold, the apply worker will stop retaining
conflict detection information. In such cases, the conflict slot's xmin
will be set to InvalidTransactionId, provided that all apply workers
associated with the subscription (with retain_dead_tuples enabled) confirm
the retention duration has been exceeded.

To ensure retention status persists across server restarts, a new column
subretentionactive has been added to the pg_subscription catalog. This
prevents unnecessary reactivation of retention logic after a restart.

The conflict detection slot will not be automatically re-initialized
unless a new subscription is created with retain_dead_tuples = true, or
the user manually re-enables retain_dead_tuples.

A future patch will introduce support for automatic slot re-initialization
once at least one apply worker confirms that the retention duration is
within the configured max_retention_duration.

Author: Zhijie Hou <houzj.fnst@fujitsu.com>
Reviewed-by: shveta malik <shveta.malik@gmail.com>
Reviewed-by: Nisha Moond <nisha.moond412@gmail.com>
Reviewed-by: Masahiko Sawada <sawada.mshk@gmail.com>
Reviewed-by: Dilip Kumar <dilipbalaut@gmail.com>
Reviewed-by: Amit Kapila <amit.kapila16@gmail.com>
Discussion: https://postgr.es/m/OS0PR01MB5716BE80DAEB0EE2A6A5D1F5949D2@OS0PR01MB5716.jpnprd01.prod.outlook.com

1 files changed, 53 insertions, 0 deletions

diff --git a/src/test/subscription/t/035_conflicts.pl b/src/test/subscription/t/035_conflicts.pl
index 6b4a9fb8815..51b23a39fa9 100644
--- a/src/test/subscription/t/035_conflicts.pl
+++ b/src/test/subscription/t/035_conflicts.pl
@@ -387,6 +387,59 @@ ok( $logfile =~
 	'update target row was deleted in tab');
 
 ###############################################################################
+# Check that dead tuple retention stops due to the wait time surpassing
+# max_retention_duration.
+###############################################################################
+
+# Create a physical slot
+$node_B->safe_psql('postgres',
+	"SELECT * FROM pg_create_physical_replication_slot('blocker');");
+
+# Add the inactive physical slot to synchronized_standby_slots
+$node_B->append_conf('postgresql.conf',
+	"synchronized_standby_slots = 'blocker'");
+$node_B->reload;
+
+# Enable failover to activate the synchronized_standby_slots setting
+$node_A->safe_psql('postgres', "ALTER SUBSCRIPTION $subname_AB DISABLE;");
+$node_A->safe_psql('postgres', "ALTER SUBSCRIPTION $subname_AB SET (failover = true);");
+$node_A->safe_psql('postgres', "ALTER SUBSCRIPTION $subname_AB ENABLE;");
+
+# Insert a record
+$node_B->safe_psql('postgres', "INSERT INTO tab VALUES (5, 5);");
+
+# Advance the xid on Node A to trigger the next cycle of oldest_nonremovable_xid
+# advancement.
+$node_A->safe_psql('postgres', "SELECT txid_current() + 1;");
+
+$log_offset = -s $node_A->logfile;
+
+# Set max_retention_duration to a minimal value to initiate retention stop.
+$node_A->safe_psql('postgres',
+	"ALTER SUBSCRIPTION $subname_AB SET (max_retention_duration = 1);");
+
+# Confirm that the retention is stopped
+$node_A->wait_for_log(
+	qr/logical replication worker for subscription "tap_sub_a_b" has stopped retaining the information for detecting conflicts/,
+	$log_offset);
+
+ok( $node_A->poll_query_until(
+		'postgres',
+		"SELECT xmin IS NULL from pg_replication_slots WHERE slot_name = 'pg_conflict_detection'"
+	),
+	"the xmin value of slot 'pg_conflict_detection' is invalid on Node A");
+
+$result = $node_A->safe_psql('postgres',
+	"SELECT subretentionactive FROM pg_subscription WHERE subname='$subname_AB';");
+is($result, qq(f), 'retention is inactive');
+
+# Drop the physical slot and reset the synchronized_standby_slots setting
+$node_B->safe_psql('postgres',
+	"SELECT * FROM pg_drop_replication_slot('blocker');");
+$node_B->adjust_conf('postgresql.conf', 'synchronized_standby_slots', "''");
+$node_B->reload;
+
+###############################################################################
 # Check that the replication slot pg_conflict_detection is dropped after
 # removing all the subscriptions.
 ###############################################################################