# Copyright (c) 2026, PostgreSQL Global Development Group

# A standby's synced slot can hold back catalog VACUUM on the primary in three
# ways. In each, the synced slot's catalog_xmin reaches the primary's physical
# slot sb_phys via hot_standby_feedback and pins the catalog horizon, so dead
# pg_class tuples cannot be removed.
#
# 1. A sync worker that keeps erroring on a failover slot in a database the
#    standby has not yet replayed, so the synced slot stays frozen.
# 2. Slot sync off with no manual sync, so the synced slot never advances.
# 3. A long feedback interval, so the synced slot advances but its new value
#    is never reported and the physical slot keeps the old one.

use strict;
use warnings FATAL => 'all';
use PostgreSQL::Test::Cluster;
use PostgreSQL::Test::Utils;
use Test::More;

# Consume XIDs so a slot's catalog_xmin can be pushed forward.
my $consume_xid = q{
	CREATE PROCEDURE consume_xid(cnt int) AS $$
	DECLARE i int;
	BEGIN FOR i IN 1..cnt LOOP PERFORM pg_current_xact_id(); COMMIT; END LOOP; END;
	$$ LANGUAGE plpgsql;
};

# Advance a logical slot's catalog_xmin past $beyond, returning the new value.
sub advance_slot
{
	my ($node, $db, $slot, $beyond) = @_;
	my $cur = -1;
	foreach (1 .. 80)
	{
		$node->safe_psql('postgres', "CALL consume_xid(20)");
		$node->safe_psql('postgres', "SELECT pg_log_standby_snapshot()");
		$node->safe_psql($db,
			"SELECT pg_replication_slot_advance('$slot', pg_current_wal_lsn())");
		$cur = $node->safe_psql('postgres',
			"SELECT catalog_xmin FROM pg_replication_slots WHERE slot_name='$slot'");
		last if $cur ne '' and $cur > $beyond;
	}
	return $cur;
}

# Leave dead pg_class tuples on the primary; return an XID past all of them.
sub make_dead_catalog
{
	my ($node) = @_;
	$node->safe_psql('postgres', q{
		DO $$ BEGIN FOR i IN 1..200 LOOP
			EXECUTE format('CREATE TABLE dead_%s (a int)', i);
			EXECUTE format('DROP TABLE dead_%s', i);
		END LOOP; END $$;
	});
	return $node->safe_psql('postgres', "SELECT pg_current_xact_id()::text::int8");
}

# VACUUM a relation and return (removable cutoff, dead-but-not-removable count).
sub table_vacuum
{
	my ($node, $rel) = @_;
	my $out;
	$node->psql('postgres', "VACUUM (VERBOSE) $rel", stderr => \$out);
	my ($cutoff) = $out =~ /removable cutoff: (\d+)/;
	my ($dead) = $out =~ /(\d+) are dead but not yet removable/;
	return ($cutoff, $dead);
}

# A primary set up for logical decoding, with the physical slot the standby
# connects through.
sub make_primary
{
	my ($name) = @_;
	my $p = PostgreSQL::Test::Cluster->new($name);
	$p->init(allows_streaming => 'logical');
	$p->append_conf('postgresql.conf', 'autovacuum = off');
	$p->start;
	$p->safe_psql('postgres', $consume_xid);
	$p->safe_psql('postgres',
		"SELECT pg_create_physical_replication_slot('sb_phys')");
	return $p;
}

# Attach a standby from the primary's backup with the given extra conf and
# catch it up.
sub attach_standby
{
	my ($p, $name, $conf) = @_;
	my $s = PostgreSQL::Test::Cluster->new($name);
	$s->init_from_backup($p, 'backup', has_streaming => 1);
	my $connstr = $p->connstr;
	$s->append_conf(
		'postgresql.conf', qq(
primary_slot_name = 'sb_phys'
primary_conninfo = '$connstr dbname=postgres'
hot_standby_feedback = on
$conf
));
	$s->start;
	$p->wait_for_replay_catchup($s);
	$p->safe_psql('postgres', "SELECT pg_log_standby_snapshot()");
	$p->wait_for_replay_catchup($s);
	return $s;
}

# Read a slot's catalog_xmin.
sub cat_xmin
{
	my ($node, $slot) = @_;
	return $node->safe_psql('postgres',
		"SELECT catalog_xmin FROM pg_replication_slots WHERE slot_name='$slot'");
}

# Case 1: sync worker erroring out. A failover slot in a database the standby
# has not yet replayed stalls the whole sync cycle, so failover_slot stays
# frozen while the worker keeps retrying.
{
	my $p = make_primary('primary1');
	$p->backup('backup');

	# Drop a placeholder slot first so newdb_slot below reuses its low array
	# index and is walked before failover_slot, which would otherwise sync
	# first.
	$p->safe_psql('postgres',
		"SELECT pg_create_logical_replication_slot('placeholder', 'pgoutput', false, false, true)");
	$p->safe_psql('postgres',
		"SELECT pg_create_logical_replication_slot('failover_slot', 'pgoutput', false, false, true)");
	$p->safe_psql('postgres', "SELECT pg_drop_replication_slot('placeholder')");

	my $s = attach_standby($p, 'standby1',
		"wal_receiver_status_interval = 1\nsync_replication_slots = on");
	$s->poll_query_until('postgres',
		"SELECT count(*) = 1 FROM pg_replication_slots WHERE slot_name = 'failover_slot' AND synced AND NOT temporary"
	) or die "failover_slot not synced";

	# Nudge the catalog_xmin off its initial value so the case is realistic,
	# then let the worker carry the new value to the synced copy.
	my $start = $p->safe_psql('postgres', "SELECT pg_current_xact_id()::text::int8");
	advance_slot($p, 'postgres', 'failover_slot', $start);
	$s->poll_query_until('postgres',
		"SELECT catalog_xmin::text::int8 > '$start' FROM pg_replication_slots WHERE slot_name = 'failover_slot'"
	) or die "synced failover_slot did not advance";

	my $c0 = cat_xmin($s, 'failover_slot');
	$p->poll_query_until('postgres',
		"SELECT catalog_xmin = '$c0' FROM pg_replication_slots WHERE slot_name = 'sb_phys'"
	) or die "sb_phys did not pick up failover_slot catalog_xmin";

	my $dead_top = make_dead_catalog($p);

	# Delay apply so the standby streams but does not apply the CREATE
	# DATABASE below. newdb stays out of the standby catalog and the sync
	# worker keeps retrying, while replay stays live not hard-paused.
	$s->safe_psql('postgres', "ALTER SYSTEM SET recovery_min_apply_delay = '5min'");
	$s->reload;
	$s->poll_query_until('postgres',
		"SELECT current_setting('recovery_min_apply_delay') = '5min'")
	  or die "recovery_min_apply_delay not in effect";
	$p->safe_psql('postgres', "CREATE DATABASE newdb");
	$p->safe_psql('newdb',
		"SELECT pg_create_logical_replication_slot('newdb_slot', 'pgoutput', false, false, true)");
	$p->safe_psql('postgres', "SELECT pg_log_standby_snapshot()");
	$p->wait_for_catchup($s, 'flush');
	$s->wait_for_log(qr/database "newdb" does not exist/, -s $s->logfile);

	# Advance both primary slots past the dead tuples, leaving the standby's
	# frozen failover_slot (seen via feedback) as the only catalog horizon.
	advance_slot($p, 'postgres', 'failover_slot', $dead_top);
	advance_slot($p, 'newdb', 'newdb_slot', $dead_top);

	is(cat_xmin($s, 'failover_slot'), $c0,
		'case 1: synced slot frozen while the worker retries');
	is(cat_xmin($p, 'sb_phys'), $c0,
		'case 1: physical slot pinned at the frozen value');

	my ($ccut, $cdead) = table_vacuum($p, 'pg_class');
	is($ccut, $c0, 'case 1: catalog vacuum cutoff pinned at the frozen value');
	cmp_ok($cdead, '>', 0, 'case 1: dead catalog tuples cannot be removed');

	$s->stop;
	$p->stop;
}

# Case 2: sync off, no manual sync. The slot is synced once, then never again
# while the primary's slot advances, so the synced copy stays frozen.
{
	my $p = make_primary('primary2');
	$p->backup('backup');
	$p->safe_psql('postgres',
		"SELECT pg_create_logical_replication_slot('failover_slot', 'pgoutput', false, false, true)");

	my $s = attach_standby($p, 'standby2',
		"wal_receiver_status_interval = 1\nsync_replication_slots = off");
	$s->safe_psql('postgres', "SELECT pg_sync_replication_slots()");

	# Nudge the catalog_xmin off its initial value, then re-sync once.
	my $start = $p->safe_psql('postgres', "SELECT pg_current_xact_id()::text::int8");
	advance_slot($p, 'postgres', 'failover_slot', $start);
	$s->safe_psql('postgres', "SELECT pg_sync_replication_slots()");

	my $c0 = cat_xmin($s, 'failover_slot');
	$p->poll_query_until('postgres',
		"SELECT catalog_xmin = '$c0' FROM pg_replication_slots WHERE slot_name = 'sb_phys'"
	) or die "sb_phys did not pick up failover_slot catalog_xmin";

	my $dead_top = make_dead_catalog($p);

	# Advance the primary's slot past the dead tuples, but do not re-sync.
	advance_slot($p, 'postgres', 'failover_slot', $dead_top);

	is(cat_xmin($s, 'failover_slot'), $c0,
		'case 2: synced slot frozen without a re-sync');
	is(cat_xmin($p, 'sb_phys'), $c0,
		'case 2: physical slot pinned at the frozen value');

	my ($ccut, $cdead) = table_vacuum($p, 'pg_class');
	is($ccut, $c0, 'case 2: catalog vacuum cutoff pinned at the frozen value');
	cmp_ok($cdead, '>', 0, 'case 2: dead catalog tuples cannot be removed');

	$s->stop;
	$p->stop;
}

# Case 3: long feedback interval. The synced slot advances normally, but the
# standby stops reporting its catalog_xmin, so the primary's physical slot keeps
# the old value.
{
	my $p = make_primary('primary3');
	$p->backup('backup');
	$p->safe_psql('postgres',
		"SELECT pg_create_logical_replication_slot('failover_slot', 'pgoutput', false, false, true)");

	my $s = attach_standby($p, 'standby3',
		"wal_receiver_status_interval = 1\nsync_replication_slots = on");
	$s->poll_query_until('postgres',
		"SELECT count(*) = 1 FROM pg_replication_slots WHERE slot_name = 'failover_slot' AND synced AND NOT temporary"
	) or die "failover_slot not synced";

	# Nudge the catalog_xmin off its initial value, and let the worker carry
	# it to the synced copy.
	my $start = $p->safe_psql('postgres', "SELECT pg_current_xact_id()::text::int8");
	advance_slot($p, 'postgres', 'failover_slot', $start);
	$s->poll_query_until('postgres',
		"SELECT catalog_xmin::text::int8 > '$start' FROM pg_replication_slots WHERE slot_name = 'failover_slot'"
	) or die "synced failover_slot did not advance";

	my $c0 = cat_xmin($s, 'failover_slot');
	$p->poll_query_until('postgres',
		"SELECT catalog_xmin = '$c0' FROM pg_replication_slots WHERE slot_name = 'sb_phys'"
	) or die "sb_phys did not pick up failover_slot catalog_xmin";

	# Stop reporting the standby's catalog_xmin to the primary.
	$s->safe_psql('postgres', "ALTER SYSTEM SET wal_receiver_status_interval = 3600");
	$s->reload;

	my $dead_top = make_dead_catalog($p);

	# The worker keeps the synced slot current, so it moves past the dead
	# tuples.
	advance_slot($p, 'postgres', 'failover_slot', $dead_top);
	$s->poll_query_until('postgres',
		"SELECT catalog_xmin::text::int8 > '$dead_top' FROM pg_replication_slots WHERE slot_name = 'failover_slot'"
	) or die "synced failover_slot did not advance";

	cmp_ok(cat_xmin($s, 'failover_slot'), '>', $dead_top,
		'case 3: synced slot is current');
	is(cat_xmin($p, 'sb_phys'), $c0,
		'case 3: physical slot stale, the new value was never reported');

	my ($ccut, $cdead) = table_vacuum($p, 'pg_class');
	is($ccut, $c0, 'case 3: catalog vacuum cutoff pinned at the stale value');
	cmp_ok($cdead, '>', 0, 'case 3: dead catalog tuples cannot be removed');

	$s->stop;
	$p->stop;
}

done_testing();