# Copyright (c) 2026, PostgreSQL Global Development Group

# Reproducer: slot synchronization on a standby can deadlock with WAL replay
# through a heavyweight lock when max_standby_streaming_delay = -1.
#
# Slot sync decodes WAL from the slot's start and waits for replay to reach
# the remote confirmed_lsn.  While it waits it holds heavyweight locks:
#
#   fn_rel: a relation lock the calling transaction took before calling
#           pg_sync_replication_slots().  Replaying a primary's
#           AccessExclusiveLock on that relation waits for it.
#   fn_db:  AccessShareLock on the slot's database, which synchronize_slots()
#           holds around synchronize_one_slot().  Replaying the XLOG_DBASE_DROP
#           record that ALTER DATABASE ... SET TABLESPACE writes takes
#           AccessExclusiveLock on the database first, and waits for it.
#   wk_db:  the same as fn_db, but the slot sync worker holds the lock.
#
# In all three, the startup process waits in ResolveRecoveryConflictWithLock(),
# which under -1 never cancels anyone.  Its deadlock_timeout request is
# ignored because the syncer is waiting for replay, not for a lock.
#
# "side" separately checks what replaying the moved database's
# XLOG_DBASE_DROP does to a logical slot created on the standby.
#
# The assertions describe the deadlock, so they pass on affected servers.

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

my @names = qw(fn_rel fn_db wk_db);

my $primary = PostgreSQL::Test::Cluster->new('primary');
$primary->init(allows_streaming => 'logical');
$primary->append_conf('postgresql.conf', 'allow_in_place_tablespaces = on');
$primary->start;
$primary->safe_psql('postgres', 'CREATE TABLE foo (a int)');
$primary->safe_psql('postgres', 'CREATE DATABASE d');
$primary->safe_psql('postgres', "CREATE TABLESPACE ts2 LOCATION ''");
$primary->safe_psql('postgres',
	"SELECT pg_create_physical_replication_slot('${_}_slot')")
  for @names;

# A logical slot on the primary in d, created before the move.
$primary->safe_psql('d',
	"SELECT pg_create_logical_replication_slot('primary_keep', 'test_decoding')"
);

$primary->backup('backup');
my $connstr = $primary->connstr;
my %node;
for my $name (@names)
{
	my $node = PostgreSQL::Test::Cluster->new($name);
	$node->init_from_backup($primary, 'backup', has_streaming => 1);
	$node->append_conf(
		'postgresql.conf', qq[
hot_standby_feedback = on
primary_slot_name = '${name}_slot'
primary_conninfo = '$connstr dbname=postgres'
max_standby_streaming_delay = -1
deadlock_timeout = 100ms
log_recovery_conflict_waits = on
]);
	$node->start;
	$primary->wait_for_replay_catchup($node);
	$node{$name} = $node;
}

my $side = PostgreSQL::Test::Cluster->new('side');
$side->init_from_backup($primary, 'backup', has_streaming => 1);
$side->start;
$primary->wait_for_replay_catchup($side);
$side->create_logical_slot_on_standby($primary, 'standby_own', 'd');

# fn_rel: the calling transaction holds a lock on foo.
my $fn_rel = $node{fn_rel}->background_psql('postgres', on_error_stop => 0);
my %sync_pid;
$sync_pid{fn_rel} = $fn_rel->query_safe('SELECT pg_backend_pid()');
chomp $sync_pid{fn_rel};
$fn_rel->query_safe('BEGIN; LOCK TABLE foo IN ACCESS SHARE MODE;');

# Pause replay so that the following steps happen in a known order.  This
# stands in for replay lag (e.g. replay still copying the moved database).
$node{$_}->safe_psql('postgres', 'SELECT pg_wal_replay_pause()') for @names;

# On the primary: an AccessExclusiveLock on foo, then move d, then create a
# failover slot in d whose WAL starts after both.
$primary->safe_psql('postgres',
	'BEGIN; LOCK TABLE foo IN ACCESS EXCLUSIVE MODE; COMMIT;');
$primary->safe_psql('postgres', 'ALTER DATABASE d SET TABLESPACE ts2');
$primary->safe_psql('d',
	"SELECT pg_create_logical_replication_slot('fs', 'test_decoding', false, false, true)"
);
my $slot_lsn = $primary->safe_psql('postgres',
	"SELECT confirmed_flush_lsn FROM pg_replication_slots WHERE slot_name = 'fs'"
);
for my $name (@names)
{
	$node{$name}->poll_query_until('postgres',
		"SELECT pg_last_wal_receive_lsn() >= '$slot_lsn'")
	  or die "$name did not receive the slot's WAL";
}

# side: replaying the move's XLOG_DBASE_DROP drops every logical slot of d on
# the standby, although the primary kept its own.
$primary->wait_for_replay_catchup($side);
is( $primary->safe_psql(
		'postgres',
		"SELECT count(*) FROM pg_replication_slots WHERE slot_name = 'primary_keep'"
	),
	'1',
	'side: the primary keeps its logical slot in d after the move');
is( $side->safe_psql(
		'postgres',
		"SELECT count(*) FROM pg_replication_slots WHERE slot_name = 'standby_own'"
	),
	'0',
	'side: the standby lost its own logical slot in d after replaying the move'
);

# Start syncing.
my $fn_db = $node{fn_db}->background_psql('postgres', on_error_stop => 0);
$sync_pid{fn_db} = $fn_db->query_safe('SELECT pg_backend_pid()');
chomp $sync_pid{fn_db};
$fn_rel->query_until(qr/started/,
	"\\echo started\nSELECT pg_sync_replication_slots();\n");
$fn_db->query_until(qr/started/,
	"\\echo started\nSELECT pg_sync_replication_slots();\n");
$node{wk_db}->append_conf('postgresql.conf', 'sync_replication_slots = on');
$node{wk_db}->reload;

for my $name (@names)
{
	my $node = $node{$name};
	$node->poll_query_until('postgres',
		"SELECT count(*) = 1 FROM pg_replication_slots WHERE slot_name = 'fs' AND synced"
	) or die "$name: slot sync did not create the local slot";
	if ($name eq 'wk_db')
	{
		$sync_pid{$name} = $node->safe_psql('postgres',
			"SELECT pid FROM pg_stat_activity WHERE backend_type = 'slotsync worker'"
		);
	}
	is( $node->safe_psql(
			'postgres',
			"SELECT count(*) FROM pg_locks
			 WHERE pid = $sync_pid{$name} AND locktype = 'object'
			   AND classid = 'pg_database'::regclass
			   AND objid = (SELECT oid FROM pg_database WHERE datname = 'd')
			   AND mode = 'AccessShareLock' AND granted"
		),
		'1',
		"$name: syncer holds AccessShareLock on database d while waiting for replay"
	);
}

my %log_offset;
for my $name (@names)
{
	$log_offset{$name} = -s $node{$name}->logfile;
	$node{$name}->safe_psql('postgres', 'SELECT pg_wal_replay_resume()');
}

my %expect = (
	fn_rel => [ 'relation', "locktype = 'relation' AND relation = 'foo'::regclass" ],
	fn_db => [ 'object', "locktype = 'object' AND classid = 'pg_database'::regclass" ],
	wk_db => [ 'object', "locktype = 'object' AND classid = 'pg_database'::regclass" ]);

for my $name (@names)
{
	my $node = $node{$name};
	my $pid = $sync_pid{$name};
	my ($event, $lockqual) = @{ $expect{$name} };

	ok( $node->poll_query_until(
			'postgres',
			"SELECT wait_event_type = 'Lock' AND wait_event = '$event' FROM pg_stat_activity WHERE backend_type = 'startup'"
		),
		"$name: startup waits for a heavyweight lock ($event)");
	$node->wait_for_log(
		qr/recovery conflict on lock[^\n]*\n[^\n]*Conflicting process: $pid\b/,
		$log_offset{$name});
	ok(1, "$name: startup reports the syncer as the conflicting process");
}

# Many deadlock_timeout periods pass; nothing breaks the cycle.
sleep(3);

for my $name (@names)
{
	my $node = $node{$name};
	my $pid = $sync_pid{$name};
	my ($event, $lockqual) = @{ $expect{$name} };

	is( $node->safe_psql(
			'postgres', "SELECT pg_last_wal_replay_lsn() < '$slot_lsn'"),
		't',
		"$name: replay has not reached the slot position");
	is( $node->safe_psql(
			'postgres',
			"SELECT wait_event_type || ':' || wait_event FROM pg_stat_activity WHERE backend_type = 'startup'"
		),
		"Lock:$event",
		"$name: startup still waits for the lock");
	is( $node->safe_psql(
			'postgres',
			"SELECT count(*) FROM pg_locks
			 WHERE pid = (SELECT pid FROM pg_stat_activity WHERE backend_type = 'startup')
			   AND $lockqual AND mode = 'AccessExclusiveLock' AND NOT granted"
		),
		'1',
		"$name: startup's AccessExclusiveLock request is not granted");
	is( $node->safe_psql(
			'postgres',
			"SELECT coalesce(wait_event_type, '-') <> 'Lock' FROM pg_stat_activity WHERE pid = $pid"
		),
		't',
		"$name: the syncer is not waiting for a lock");
	is( $node->safe_psql(
			'postgres',
			"SELECT temporary FROM pg_replication_slots WHERE slot_name = 'fs'"),
		't',
		"$name: synced slot has not become persistent");
	note "$name: syncer activity: "
	  . $node->safe_psql('postgres',
		"SELECT backend_type, state, coalesce(wait_event_type, '-'), coalesce(wait_event, '-') FROM pg_stat_activity WHERE pid = $pid"
	  );
}

# Break each cycle by canceling the syncer.  The syncer releases its
# heavyweight locks when it aborts, but releases its temporary synced slot
# only later.  In the fn_db and wk_db cases the lock release wakes the startup
# process at once, and if it reaches ReplicationSlotsDropDBSlots() before the
# slot is released, it fails with "replication slot is active", which is
# FATAL in the startup process and shuts the standby down.  Which side wins is
# timing; the worker's longer exit path makes the crash likely there.
for my $name (@names)
{
	my $node = $node{$name};
	my $offset = -s $node->logfile;
	$node->safe_psql('postgres', "SELECT pg_cancel_backend($sync_pid{$name})");

	my $lsn = $primary->lsn('insert');
	my $outcome;
	for (1 .. 300)
	{
		my $log = slurp_file($node->logfile, $offset);
		if ($log =~ /startup\[\d+\] (FATAL:  [^\n]*)/)
		{
			$outcome = "standby shut down: $1";
			last;
		}
		my ($ret, $out) = $node->psql('postgres',
			"SELECT pg_last_wal_replay_lsn() >= '$lsn'");
		if ($ret == 0 && $out eq 't')
		{
			$outcome = 'replay caught up';
			last;
		}
		select(undef, undef, undef, 0.1);
	}
	note "$name: after canceling the syncer: " . ($outcome // 'no progress');
	ok(defined $outcome, "$name: replay is no longer blocked once the syncer is canceled");
}

$fn_rel->quit;
$fn_db->quit;

done_testing();