# Copyright (c) 2026, PostgreSQL Global Development Group

# Apply on a subscriber whose default_transaction_isolation is serializable.
#
# The apply worker inherits default_transaction_isolation, so its transaction
# is serializable here.  Apply only reads the row it looks up in order to
# overwrite it, and it locks that row under the latest snapshot anyway, so
# the search must not take part in SSI: no SIREAD locks, no conflict-out
# checks.  Otherwise a local serializable transaction touching the same rows
# makes apply fail with a serialization failure and the worker restart (or
# the subscription get disabled with disable_on_error).
#
# Both cases make apply wait on a row lock held by a local transaction, so
# that the local transaction's commit is ordered after apply's search.

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

my $node_publisher = PostgreSQL::Test::Cluster->new('publisher');
$node_publisher->init(allows_streaming => 'logical');
$node_publisher->start;

my $node_subscriber = PostgreSQL::Test::Cluster->new('subscriber');
$node_subscriber->init;
$node_subscriber->append_conf('postgresql.conf',
	'default_transaction_isolation = serializable');
$node_subscriber->start;

# tab is searched through its primary key; i is a subscriber-only column,
# indexed so that updating it is never HOT.  tab_full has no index and is
# searched sequentially.
my $ddl = q[
	CREATE TABLE tab (a int PRIMARY KEY, data text);
	CREATE TABLE tab_full (id int, v int);
	ALTER TABLE tab_full REPLICA IDENTITY FULL;
];
$node_publisher->safe_psql('postgres', $ddl);
$node_subscriber->safe_psql('postgres', $ddl);
$node_subscriber->safe_psql(
	'postgres', q[
	ALTER TABLE tab ADD COLUMN i int DEFAULT 0;
	CREATE INDEX tab_i ON tab (i);
]);

$node_publisher->safe_psql(
	'postgres', q[
	INSERT INTO tab SELECT g, 'init' FROM generate_series(1, 10) g;
	INSERT INTO tab_full SELECT g, 0 FROM generate_series(1, 200) g;
	CREATE PUBLICATION tap_pub FOR TABLE tab, tab_full;
]);

my $appname = 'tap_sub';
my $publisher_connstr = $node_publisher->connstr . ' dbname=postgres';
$node_subscriber->safe_psql('postgres',
	"CREATE SUBSCRIPTION tap_sub CONNECTION '$publisher_connstr application_name=$appname' PUBLICATION tap_pub"
);
$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);

is( $node_subscriber->safe_psql(
		'postgres',
		q[SELECT current_setting('default_transaction_isolation')]),
	'serializable',
	'subscriber defaults to serializable');

sub wait_apply_blocked
{
	$node_subscriber->poll_query_until('postgres',
		q[SELECT count(*) = 1 FROM pg_stat_activity
		  WHERE backend_type = 'logical replication apply worker'
		    AND wait_event_type = 'Lock']
	) or die 'timed out waiting for apply to block on a row lock';
}

# Wait for catchup, then report whether apply failed since $log_offset.
sub apply_errors_since
{
	my ($log_offset) = @_;
	$node_publisher->wait_for_catchup($appname);
	my $log = slurp_file($node_subscriber->logfile, $log_offset);
	my @errors = $log =~
	  /logical replication apply worker.*ERROR:\s+(.*)/g;
	return join("\n", @errors);
}

# Index search.  A local serializable transaction reads row 2 and updates
# row 1 (non-HOT).  Apply then writes row 2 (local -> apply) and, searching
# for row 1, finds the version the local transaction deleted and waits for
# it.  If apply's search is an SSI read, that adds apply -> local and apply
# is a pivot when the local transaction commits.
my $log_offset = -s $node_subscriber->logfile;
my $local = $node_subscriber->background_psql('postgres');
$local->query_safe(
	q[
	BEGIN ISOLATION LEVEL SERIALIZABLE;
	SELECT * FROM tab WHERE a = 2;
	UPDATE tab SET i = 1 WHERE a = 1;
]);
$node_publisher->safe_psql(
	'postgres', q[
	BEGIN;
	UPDATE tab SET data = 'frompub' WHERE a = 2;
	UPDATE tab SET data = 'frompub' WHERE a = 1;
	COMMIT;
]);
wait_apply_blocked();
$local->query_safe('COMMIT');
$local->quit;

is(apply_errors_since($log_offset), '',
	'index search: apply did not fail');
is( $node_subscriber->safe_psql(
		'postgres', 'SELECT a, data, i FROM tab WHERE a IN (1, 2) ORDER BY a'),
	"1|frompub|1\n2|frompub|0",
	'index search: both updates applied');

# Sequential search with REPLICA IDENTITY FULL.  Apply searches and updates
# tab_full, then waits for a row of tab locked by a local read committed
# transaction.  Meanwhile a local serializable transaction updates another
# row of tab_full.  If apply's sequential search is an SSI read, it holds a
# relation-level SIREAD lock on tab_full, and that write makes apply a pivot.
$log_offset = -s $node_subscriber->logfile;
my $locker = $node_subscriber->background_psql('postgres');
$locker->query_safe(
	q[
	BEGIN ISOLATION LEVEL READ COMMITTED;
	SELECT a FROM tab WHERE a = 3 FOR UPDATE;
]);
$node_publisher->safe_psql(
	'postgres', q[
	BEGIN;
	UPDATE tab_full SET v = 1 WHERE id = 1;
	UPDATE tab SET data = 'frompub' WHERE a = 3;
	COMMIT;
]);
wait_apply_blocked();
$node_subscriber->safe_psql('postgres',
	'UPDATE tab_full SET v = v + 1 WHERE id = 99');
$locker->query_safe('COMMIT');
$locker->quit;

is(apply_errors_since($log_offset), '',
	'sequential search: apply did not fail');
is( $node_subscriber->safe_psql(
		'postgres',
		q[SELECT string_agg(id || '=' || v, ' ' ORDER BY id)
		  FROM tab_full WHERE id IN (1, 99)]),
	'1=1 99=1',
	'sequential search: both updates applied');

is( $node_subscriber->safe_psql(
		'postgres',
		q[SELECT apply_error_count FROM pg_stat_subscription_stats
		  WHERE subname = 'tap_sub']),
	'0',
	'no apply errors counted');

done_testing();