# Copyright (c) 2026, PostgreSQL Global Development Group

# Probe: with a finite max_standby_streaming_delay the lock cycle of 061
# (syncer holds AccessShareLock on database d and waits for replay; replay of
# ALTER DATABASE d SET TABLESPACE needs AccessExclusiveLock on d) is broken by
# the startup process canceling the syncer.  What happens next?
#
# The syncer releases its heavyweight locks when it aborts, before it
# releases its temporary synced slot.  If the startup process gets the
# database lock in that window, ReplicationSlotsDropDBSlots() finds the slot
# still active and errors, which is FATAL in the startup process.
#
#   wk_fin:  slot sync worker
#   fn_finN: pg_sync_replication_slots()

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

my @names = qw(wk_fin fn_fin1 fn_fin2 fn_fin3);

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 DATABASE d');
$primary->safe_psql('postgres', "CREATE TABLESPACE ts2 LOCATION ''");
$primary->safe_psql('postgres',
	"SELECT pg_create_physical_replication_slot('${_}_slot')")
  for @names;
$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 = 1s
log_recovery_conflict_waits = on
]);
	$node->start;
	$primary->wait_for_replay_catchup($node);
	$node->safe_psql('postgres', 'SELECT pg_wal_replay_pause()');
	$node{$name} = $node;
}

$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'"
);
$node{$_}->poll_query_until('postgres',
	"SELECT pg_last_wal_receive_lsn() >= '$slot_lsn'")
  for @names;

my (%sync_pid, %psql);
for my $name (@names)
{
	my $node = $node{$name};
	if ($name =~ /^wk_/)
	{
		$node->append_conf('postgresql.conf', 'sync_replication_slots = on');
		$node->reload;
	}
	else
	{
		my $p = $node->background_psql('postgres', on_error_stop => 0);
		$sync_pid{$name} = $p->query_safe('SELECT pg_backend_pid()');
		chomp $sync_pid{$name};
		$p->query_until(qr/started/,
			"\\echo started\nSELECT pg_sync_replication_slots();\n");
		$psql{$name} = $p;
	}
	$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";
	$sync_pid{$name} //= $node->safe_psql('postgres',
		"SELECT pid FROM pg_stat_activity WHERE backend_type = 'slotsync worker'"
	);
}

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

# The startup process cancels each syncer after about 1s.
for my $name (@names)
{
	my $pid = $sync_pid{$name};
	$node{$name}->wait_for_log(
		qr/\[$pid\] [^\n]*ERROR:  [^\n]*conflict with recovery/,
		$log_offset{$name});
	ok(1, "$name: startup canceled the syncer ($pid)");
}
sleep(2);

for my $name (@names)
{
	my $log = slurp_file($node{$name}->logfile, $log_offset{$name});
	my ($cancel) = $log =~ /(ERROR:  [^\n]*conflict with recovery\n[^\n]*DETAIL:  [^\n]*)/;
	note "$name: syncer got: " . ($cancel // '?');
	if ($log =~ /(startup\[\d+\] FATAL:  [^\n]*)\n[^\n]*(CONTEXT:  [^\n]*)/)
	{
		note "$name: $1 / $2";
		like($log, qr/shutting down due to startup process failure/,
			"$name: standby shut down after the startup process failed");
	}
	else
	{
		$primary->wait_for_replay_catchup($node{$name});
		pass("$name: standby survived and caught up");
	}
}

for my $p (values %psql)
{
	eval { $p->quit };
}

done_testing();