# Copyright (c) 2026, PostgreSQL Global Development Group

# Reproducer: creating a logical replication slot on a standby can deadlock
# with WAL replay through a *heavyweight lock*, not only through a snapshot.
#
# pg_create_logical_replication_slot() keeps waiting for replay to supply an
# xl_running_xacts record to decode from.  If the creating transaction also
# holds a heavyweight lock on a relation, and replay next reaches a record
# that needs a conflicting lock on that relation (e.g. an AccessExclusiveLock
# taken on the primary), then the startup process waits for the slot creator
# to release the lock, while the slot creator waits for replay.  Neither side
# has a timeout when max_standby_streaming_delay = -1.
#
# This is the same cycle WAIT FOR refuses to enter: WAIT errors out with
# "cannot wait for a standby LSN while holding locks".  Slot creation has no
# such guard.  The assertions below 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 $primary = PostgreSQL::Test::Cluster->new('primary');
$primary->init(allows_streaming => 'logical');
$primary->start;
$primary->safe_psql('postgres',
	'CREATE TABLE foo (a int); CREATE TABLE bar (a int);');

$primary->backup('backup');
my $standby = PostgreSQL::Test::Cluster->new('standby');
$standby->init_from_backup($primary, 'backup', has_streaming => 1);
$standby->append_conf(
	'postgresql.conf', qq[
max_standby_streaming_delay = -1
log_recovery_conflict_waits = on
]);
$standby->start;
$primary->wait_for_replay_catchup($standby);

# Transaction Z on the primary stays open so that any xl_running_xacts record
# logged meanwhile lists it as running; the slot creator therefore cannot find
# a start point and keeps waiting for replay.
my $xact_z = $primary->background_psql('postgres');
$xact_z->query_safe('BEGIN; SELECT pg_current_xact_id();');

# The slot creator's session first takes an AccessShareLock on foo, then starts
# creating a logical slot in the same transaction.  It now holds the lock and
# waits for replay.
my $creator = $standby->background_psql('postgres', on_error_stop => 0);
my $creator_pid = $creator->query_safe('SELECT pg_backend_pid()');
chomp $creator_pid;
$creator->query_safe('BEGIN; LOCK TABLE foo IN ACCESS SHARE MODE;');
$creator->query_until(
	qr/started/, q[
\echo started
SELECT pg_create_logical_replication_slot('standby_slot', 'test_decoding');
]);
$standby->poll_query_until('postgres',
	"SELECT restart_lsn IS NOT NULL FROM pg_replication_slots WHERE slot_name = 'standby_slot'"
) or die "slot creation did not reserve WAL";
is( $standby->safe_psql(
		'postgres',
		"SELECT count(*) FROM pg_locks WHERE pid = $creator_pid AND relation = 'foo'::regclass AND granted"
	),
	'1',
	'slot creator holds a lock on foo while waiting for replay');

# On the primary, take an AccessExclusiveLock on foo.  This logs a standby lock
# record.  Replaying it requires an AccessExclusiveLock on foo on the standby,
# which conflicts with the creator's AccessShareLock.
my $log_offset = -s $standby->logfile;
my $lock_holder = $primary->background_psql('postgres');
$lock_holder->query_safe('BEGIN; LOCK TABLE foo IN ACCESS EXCLUSIVE MODE;');

# Log a marker record after the lock record so we have a concrete LSN that sits
# behind the blocked lock record.
my $marker_lsn =
  $primary->safe_psql('postgres', 'SELECT pg_log_standby_snapshot()');
$standby->poll_query_until('postgres',
	"SELECT pg_last_wal_receive_lsn() >= '$marker_lsn'")
  or die "standby did not receive the lock record";

# The startup process is now waiting for the creator to release its lock on foo.
$standby->wait_for_log(qr/recovery conflict on lock|Conflicting process: $creator_pid/,
	$log_offset);
ok(1, 'startup process is waiting on the lock conflict');

# The standby has received the lock record but cannot replay it: the startup
# process waits for the slot creator's lock, and the slot creator waits for
# replay.
is( $standby->safe_psql(
		'postgres', "SELECT pg_last_wal_replay_lsn() < '$marker_lsn'"),
	't',
	'replay has not reached the received marker record');
is( $standby->safe_psql(
		'postgres',
		"SELECT wait_event_type = 'Lock' FROM pg_stat_activity WHERE backend_type = 'startup'"
	),
	't',
	'startup process is waiting on a lock');
is( $standby->safe_psql(
		'postgres',
		"SELECT confirmed_flush_lsn IS NULL FROM pg_replication_slots WHERE slot_name = 'standby_slot'"
	),
	't',
	'slot creation has not found a start point');

# For contrast: WAIT FOR refuses to enter this cycle at all.  Holding a lock
# and running WAIT FOR for an unreached target errors out immediately instead
# of hanging.  This is the guard slot creation lacks.
my $target = $primary->lsn('flush');
my ($ret, $stdout, $stderr) = $standby->psql('postgres',
	"BEGIN; LOCK TABLE bar IN ACCESS SHARE MODE; WAIT FOR LSN '$target';");
like(
	$stderr,
	qr/cannot wait for standby replay while holding|cannot wait for a standby LSN while holding locks/,
	'WAIT FOR refuses to wait while holding a lock (the guard slot creation lacks)'
);

# Break the cycle by canceling the slot creator.  The error aborts its
# transaction, releasing the lock and ending its VXID, so replay proceeds.
$standby->safe_psql('postgres', "SELECT pg_cancel_backend($creator_pid)");
$primary->wait_for_replay_catchup($standby);
ok(1, 'replay resumes once the slot creator is canceled');
ok( $standby->poll_query_until(
		'postgres', 'SELECT count(*) = 0 FROM pg_replication_slots'),
	'canceled slot creation left no slot');

$lock_holder->quit;
$xact_z->quit;
eval { $creator->quit };

done_testing();