# Experimento propio (#5151): que se PIERDE al cambiar SnapshotDirty por un
# snapshot MVCC fresco en RelationFindReplTupleByIndex.
#
# SnapshotDirty ve las filas de transacciones EN CURSO, y por eso el codigo de
# master toma su xmin/xmax y hace XactLockTableWait: espera al que esta
# insertando y recien despues decide.  Un scan MVCC no ve esa fila, asi que no
# hay a quien esperar.
#
# Escenario determinista, sin injection points: el INSERT ya esta en vuelo en
# el subscriber ANTES de que llegue el UPDATE del publisher.
#
#   sub:  BEGIN; INSERT (1,'fromsub');        <- abierta, sin commit
#   pub:  UPDATE t SET data='frompubnew' WHERE a=1;
#   sub:  COMMIT;
#
# master  -> el apply worker espera al insertador y aplica el UPDATE encima.
# parche  -> el apply worker no ve nada, reporta update_missing y sigue.
#
# No se juzga cual es "correcto": se mide que son DISTINTOS, porque el hilo
# discute exactamente si SnapshotDirty daba o no una garantia extra.
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->start;

$node_publisher->safe_psql('postgres',
	"CREATE TABLE t (a int PRIMARY KEY, data text)");
$node_subscriber->safe_psql('postgres',
	"CREATE TABLE t (a int PRIMARY KEY, data text)");

# La fila existe SOLO en el publisher: con copy_data=false el subscriber
# arranca vacio, asi que la unica version de la fila en el subscriber sera la
# que inserte la sesion local.
$node_publisher->safe_psql('postgres',
	"INSERT INTO t VALUES (1, 'frompub')");
$node_publisher->safe_psql('postgres', "CREATE PUBLICATION pub FOR TABLE t");

my $connstr = $node_publisher->connstr . ' dbname=postgres';
my $appname = 'sub_inflight';
$node_subscriber->safe_psql('postgres',
	"CREATE SUBSCRIPTION sub CONNECTION '$connstr application_name=$appname'
	 PUBLICATION pub WITH (copy_data = false)");
$node_subscriber->wait_for_subscription_sync($node_publisher, $appname);

is($node_subscriber->safe_psql('postgres', 'SELECT count(*) FROM t'),
	'0', 'el subscriber arranca sin la fila');

# 1) INSERT en vuelo en el subscriber (transaccion abierta).
my $s = $node_subscriber->background_psql('postgres');
$s->query_until(
	qr/inflight/, qq[
	\\echo inflight
	BEGIN;
	INSERT INTO t VALUES (1, 'fromsub');
]);

my $log_offset = -s $node_subscriber->logfile;

# 2) El publisher actualiza la fila.
$node_publisher->safe_psql('postgres',
	"UPDATE t SET data = 'frompubnew' WHERE a = 1");

# 3) Se le da tiempo al apply worker para que llegue al lookup y decida.
#    En master se queda esperando al insertador; con el parche resuelve al tiro.
sleep 3;
my $quedo_esperando = $node_subscriber->safe_psql('postgres',
	"SELECT count(*) FROM pg_stat_activity
	  WHERE backend_type = 'logical replication apply worker'
	    AND wait_event_type = 'Lock'");
my $reporto_missing = $node_subscriber->log_contains(
	qr/conflict=update_missing/, $log_offset) ? 1 : 0;

# 4) Recien ahora termina la sesion local.  Con ACCION=ROLLBACK se comprueba
#    el otro desenlace: master espera y, si el insertador aborta, reporta
#    update_missing igual que el parche (o sea, master no "siempre aplica":
#    espera y decide segun lo que pase de verdad).
my $accion = $ENV{ACCION} // 'COMMIT';
$s->query_safe($accion);
$s->quit;

$node_publisher->wait_for_catchup($appname);

my $final = $node_subscriber->safe_psql('postgres',
	"SELECT coalesce(max(data), '(sin fila)') FROM t WHERE a = 1");

note("apply worker esperando un lock: $quedo_esperando");
note("reporto update_missing antes del commit: $reporto_missing");
note("dato final en el subscriber: $final");

# Lo que queda registrado para el correo (no se juzga, se mide):
is("espera=$quedo_esperando missing=$reporto_missing final=$final",
	$ENV{ESPERADO} // 'sin ESPERADO',
	'resultado medido');

done_testing();