# Copyright (c) 2026, PostgreSQL Global Development Group

# Probe: the slot sync worker keeps a synced slot temporary, and active for
# its own PID, when the remote slot is behind what the standby can reserve
# ("could lead to data loss").  Replaying an XLOG_DBASE_DROP for that slot's
# database then runs ReplicationSlotsDropDBSlots(), which errors on an active
# slot; in the startup process that is FATAL and shuts the standby down.
#
# Default max_standby_streaming_delay, no cancel, no replay pause.
#
#   mv: ALTER DATABASE d SET TABLESPACE on the primary (the primary keeps
#       the slot)
#   dr: DROP DATABASE d on the primary (the primary drops the slot)

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

for my $case (qw(mv dr))
{
	my $primary = PostgreSQL::Test::Cluster->new("${case}_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('sb_slot')");
	$primary->backup('backup');

	my $standby = PostgreSQL::Test::Cluster->new("${case}_standby");
	$standby->init_from_backup($primary, 'backup', has_streaming => 1);
	my $connstr = $primary->connstr;
	$standby->append_conf(
		'postgresql.conf', qq[
hot_standby_feedback = on
primary_slot_name = 'sb_slot'
primary_conninfo = '$connstr dbname=postgres'
log_recovery_conflict_waits = on
]);
	$standby->start;

	# An inactive failover slot in d, then a transaction that moves the
	# standby's safe catalog xmin past the slot's.
	$primary->safe_psql('d',
		"SELECT pg_create_logical_replication_slot('fs', 'test_decoding', false, false, true)"
	);
	$primary->safe_psql('postgres', 'CREATE TABLE t (a int)');
	$primary->wait_for_replay_catchup($standby);

	my $offset = -s $standby->logfile;
	$standby->append_conf('postgresql.conf', 'sync_replication_slots = on');
	$standby->reload;
	$standby->wait_for_log(qr/could not synchronize replication slot "fs"/,
		$offset);
	my $worker = $standby->safe_psql('postgres',
		"SELECT pid FROM pg_stat_activity WHERE backend_type = 'slotsync worker'"
	);
	is( $standby->safe_psql(
			'postgres',
			"SELECT temporary, active_pid FROM pg_replication_slots WHERE slot_name = 'fs'"
		),
		"t|$worker",
		"$case: synced slot stays temporary and active for the worker between cycles"
	);

	$offset = -s $standby->logfile;
	if ($case eq 'mv')
	{
		$primary->safe_psql('postgres', 'ALTER DATABASE d SET TABLESPACE ts2');
	}
	else
	{
		$primary->safe_psql('postgres', 'DROP DATABASE d');
	}

	# Either the standby shuts down, or it replays past the drop.
	my $lsn = $primary->lsn('insert');
	my $result;
	for (1 .. 100)
	{
		my $log = slurp_file($standby->logfile, $offset);
		if ($log =~ /(startup\[\d+\] FATAL:  [^\n]*)\n[^\n]*(CONTEXT:  [^\n]*)/)
		{
			$result = "$1 / $2";
			last;
		}
		my ($ret, $out) = $standby->psql('postgres',
			"SELECT pg_last_wal_replay_lsn() >= '$lsn'");
		if ($ret == 0 && $out eq 't')
		{
			$result = 'replayed past the drop';
			last;
		}
		select(undef, undef, undef, 0.1);
	}
	note "$case: $result";
	like($result, qr/FATAL:  replication slot "fs" is active for PID $worker/,
		"$case: replaying the database drop is FATAL in the startup process");
	like(
		slurp_file($standby->logfile, $offset),
		qr/shutting down due to startup process failure/,
		"$case: the standby shut down");

	$primary->stop;
}

done_testing();