# Copyright (c) 2026, PostgreSQL Global Development Group

# Reproducer: pg_sync_replication_slots() on a standby can deadlock with WAL
# replay when max_standby_streaming_delay = -1.
#
# To sync a failover slot, the standby decodes WAL from the slot's start and
# waits for replay to reach it.  Suppose replay first reaches a DROP
# TABLESPACE record whose directories cannot be removed yet, because a standby
# session has temporary files there.  Tablespace conflict resolution then
# waits for every transaction that was active, including the slot sync, which
# is waiting for replay.  Once the temporary files are gone, only the slot
# sync is left, and neither side has a timeout.
#
# Slot synchronization requires hot_standby_feedback, which prevents most
# snapshot conflicts but not tablespace conflicts.  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->append_conf('postgresql.conf', 'allow_in_place_tablespaces = on');
$primary->start;
$primary->safe_psql('postgres', "CREATE TABLESPACE ts LOCATION ''");
$primary->safe_psql('postgres',
	"SELECT pg_create_physical_replication_slot('sb1_slot')");

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

# A standby session with temporary files in the tablespace.  They prevent
# replay from removing the tablespace directories.
my $temp_user = $standby->background_psql('postgres');
$temp_user->query_safe(
	q[BEGIN;
	  SET temp_tablespaces = ts;
	  SET work_mem = '64kB';
	  DECLARE c CURSOR FOR SELECT count(*) FROM generate_series(1, 6000);
	  FETCH c;]);
is( $standby->safe_psql(
		'postgres',
		"SELECT count(*) > 0 FROM pg_ls_tmpdir((SELECT oid FROM pg_tablespace WHERE spcname = 'ts'))"
	),
	't',
	'standby session has temporary files in the tablespace');

# Pause replay so that the following steps happen in a known order.  This
# stands in for ordinary replay lag.
$standby->safe_psql('postgres', 'SELECT pg_wal_replay_pause()');

# Drop the tablespace, then create a failover slot.  The slot's WAL starts
# after the DROP TABLESPACE record, so syncing it requires replay to pass
# that record.
$primary->safe_psql('postgres', 'DROP TABLESPACE ts');
$primary->safe_psql('postgres',
	"SELECT pg_create_logical_replication_slot('failover_slot', 'test_decoding', false, false, true)"
);
my $slot_lsn = $primary->safe_psql('postgres',
	"SELECT confirmed_flush_lsn FROM pg_replication_slots WHERE slot_name = 'failover_slot'"
);

# Slot sync skips a slot whose position the standby has not received yet.
$standby->poll_query_until('postgres',
	"SELECT pg_last_wal_receive_lsn() >= '$slot_lsn'")
  or die "standby did not receive the slot's WAL";

# Start syncing.  The sync creates the local slot, then decodes WAL from the
# slot's start and waits for replay, which is paused.
my $syncer = $standby->background_psql('postgres', on_error_stop => 0);
my $syncer_pid = $syncer->query_safe('SELECT pg_backend_pid()');
chomp $syncer_pid;
$syncer->query_until(
	qr/started/, q[
\echo started
SELECT pg_sync_replication_slots();
]);
$standby->poll_query_until('postgres',
	"SELECT count(*) = 1 FROM pg_replication_slots WHERE slot_name = 'failover_slot' AND synced"
) or die "slot sync did not create the local slot";

# Resume replay.  Replaying DROP TABLESPACE cannot remove the directories, so
# the startup process waits for every active transaction, including the slot
# sync.
my $log_offset = -s $standby->logfile;
$standby->safe_psql('postgres', 'SELECT pg_wal_replay_resume()');
$standby->wait_for_log(qr/Conflicting process(?:es)?: [^\n]*\b$syncer_pid\b/,
	$log_offset);
ok(1, 'startup process waits for the slot sync');

# Release the temporary files and end that transaction.  The startup process
# rechecks the transaction it waits for at least once a second, so a few
# seconds later it can only be waiting for the slot sync.
$temp_user->query_safe('CLOSE c; COMMIT;');
sleep(3);

# Replay cannot pass the DROP TABLESPACE record: the startup process waits for
# the slot sync, which waits for replay.
is( $standby->safe_psql(
		'postgres', "SELECT pg_last_wal_replay_lsn() < '$slot_lsn'"),
	't',
	'replay has not reached the slot position');
is( $standby->safe_psql(
		'postgres',
		"SELECT wait_event FROM pg_stat_activity WHERE backend_type = 'startup'"
	),
	'RecoveryConflictTablespace',
	'startup process is still waiting on the tablespace conflict');
is( $standby->safe_psql(
		'postgres',
		"SELECT state FROM pg_stat_activity WHERE pid = $syncer_pid"),
	'active',
	'slot sync is still running');
is( $standby->safe_psql(
		'postgres',
		"SELECT temporary FROM pg_replication_slots WHERE slot_name = 'failover_slot'"
	),
	't',
	'synced slot has not become persistent');

# Break the cycle by canceling the slot sync.
$standby->safe_psql('postgres', "SELECT pg_cancel_backend($syncer_pid)");
$primary->wait_for_replay_catchup($standby);
ok(1, 'replay resumes once the slot sync is canceled');

$syncer->quit;
$temp_user->quit;

done_testing();