# Copyright (c) 2026, PostgreSQL Global Development Group

# Reproducer: creating a logical replication slot on a standby can deadlock
# with WAL replay when max_standby_streaming_delay = -1.
#
# pg_create_logical_replication_slot() keeps its statement snapshot while it
# waits for replay to supply an xl_running_xacts record to start decoding
# from.  If replay first reaches a cleanup record that conflicts with that
# snapshot, the startup process waits for the slot creator, and the slot
# creator waits for replay.  Neither side has a timeout.
#
# The assertions below describe the deadlock, so they pass on affected
# servers.  hot_standby_feedback is off (the default); with it on, the
# primary would normally keep the old row version and no conflict would
# arise.

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 tab (a int) WITH (autovacuum_enabled = off);
	 INSERT INTO tab VALUES (1);');

$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 X updates the row; VACUUM will later remove the old version.
# X stays open until the slot creator has taken its snapshot, so that the
# snapshot still sees the old version.
my $xact_x = $primary->background_psql('postgres');
$xact_x->query_safe('BEGIN; UPDATE tab SET a = a + 1;');

# Transaction Z is newer than X and stays open until the conflict is in
# place.  Any xl_running_xacts record logged meanwhile lists Z as running,
# so slot creation cannot find a start point before the conflict arises.
my $xact_z = $primary->background_psql('postgres');
$xact_z->query_safe('BEGIN; SELECT pg_current_xact_id();');

# Start creating a logical slot on the standby, and wait until it has
# reserved WAL.  It now waits for replay while holding the SELECT's snapshot.
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_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";

# Commit X and vacuum.  Replaying the prune record conflicts with the slot
# creator's snapshot, so the startup process waits for the slot creator.
my $log_offset = -s $standby->logfile;
$xact_x->query_safe('COMMIT');
$primary->safe_psql('postgres', 'VACUUM tab');
$standby->wait_for_log(qr/Conflicting process: $creator_pid\b/, $log_offset);
ok(1, 'startup process waits for the slot creator');

# Commit Z and log an xl_running_xacts record.  It lists no running
# transactions, which is all the slot creator needs, but it follows the
# blocked prune record in the WAL.
$xact_z->query_safe('COMMIT');
my $running_xacts_lsn =
  $primary->safe_psql('postgres', 'SELECT pg_log_standby_snapshot()');
$standby->poll_query_until('postgres',
	"SELECT pg_last_wal_receive_lsn() >= '$running_xacts_lsn'")
  or die "standby did not receive the xl_running_xacts record";

# The standby has received that record but cannot replay it: the startup
# process waits for the slot creator, which waits for replay.
is( $standby->safe_psql(
		'postgres', "SELECT pg_last_wal_replay_lsn() < '$running_xacts_lsn'"),
	't',
	'replay has not reached the received xl_running_xacts record');
is( $standby->safe_psql(
		'postgres',
		"SELECT wait_event FROM pg_stat_activity WHERE backend_type = 'startup'"
	),
	'RecoveryConflictSnapshot',
	'startup process is still waiting on the snapshot conflict');
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');

# Break the cycle by canceling the slot creator.
$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');

$creator->quit;
$xact_x->quit;
$xact_z->quit;

done_testing();