# Copyright (c) 2026, PostgreSQL Global Development Group

# Reproducer: WAIT FOR LSN on a standby can deadlock with WAL replay when
# max_standby_streaming_delay = -1.
#
# Snapshot conflict resolution captures the VXIDs of transactions whose
# snapshots conflict with a cleanup record, and then waits for those
# transactions to end.  A captured transaction can release its snapshot but
# stay open, and then run WAIT FOR LSN for a position beyond the cleanup
# record.  The startup process waits for the transaction, and the
# transaction waits for replay.  Neither side has a timeout.

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 => 1);
$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);

# Hold a snapshot through a table-free cursor.  No relation lock is taken,
# so WAIT FOR is allowed once the cursor is closed.
my $session = $standby->background_psql('postgres', on_error_stop => 0);
my $session_pid = $session->query_safe('SELECT pg_backend_pid()');
chomp $session_pid;
$session->query_safe(
	q[BEGIN;
	  DECLARE c CURSOR FOR SELECT 1;
	  FETCH c;]);

# Remove the old row version on the primary.  Replaying the prune record
# conflicts with the cursor's snapshot, so the startup process captures the
# session's VXID and waits for its transaction to end.
my $log_offset = -s $standby->logfile;
$primary->safe_psql('postgres', 'UPDATE tab SET a = a + 1');
$primary->safe_psql('postgres', 'VACUUM tab');
$standby->wait_for_log(qr/Conflicting process: $session_pid\b/, $log_offset);
ok(1, 'startup process waits for the session');

# A committed transaction flushes the WAL written so far, so the standby
# receives the target position below.
$primary->safe_psql('postgres', 'CREATE TABLE flush_marker ()');
my $target_lsn = $primary->lsn('flush');

# Release the snapshot but keep the transaction open, then wait for a
# position beyond the blocked prune record.
$session->query_until(
	qr/waiting/, qq[
CLOSE c;
\\echo waiting
WAIT FOR LSN '$target_lsn';
]);
ok( $standby->poll_query_until(
		'postgres',
		"SELECT wait_event = 'WaitForWalReplay' FROM pg_stat_activity WHERE pid = $session_pid"
	),
	'session is waiting in WAIT FOR');
$standby->poll_query_until('postgres',
	"SELECT pg_last_wal_receive_lsn() >= '$target_lsn'")
  or die "standby did not receive the target position";

# The session no longer holds a snapshot, but the startup process still
# waits for its transaction, and the session waits for replay.
is( $standby->safe_psql(
		'postgres',
		"SELECT backend_xmin IS NULL FROM pg_stat_activity WHERE pid = $session_pid"
	),
	't',
	'session no longer holds a snapshot');
is( $standby->safe_psql(
		'postgres', "SELECT pg_last_wal_replay_lsn() < '$target_lsn'"),
	't',
	'replay has not reached the received target position');
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');

# Break the cycle by canceling the WAIT.  The error aborts the transaction,
# which ends its VXID, so replay continues while the session is still in
# the aborted transaction block.
$standby->safe_psql('postgres', "SELECT pg_cancel_backend($session_pid)");
$primary->wait_for_replay_catchup($standby);
ok(1, 'replay resumes once the WAIT is canceled');

$session->query('ROLLBACK');
$session->quit;

done_testing();