From 1deb59afeca92ef11778c0c83b9a9c3e5e2aba39 Mon Sep 17 00:00:00 2001 From: Asif Rehman Date: Thu, 3 Sep 2026 20:34:20 +0500 Subject: [PATCH 1/4] Add spock-output-txn-stall injection point SPOCK_WORKER_DELAY() only covers the subscriber side (apply-worker start/finish). Add SPOCK_OUTPUT_TXN_STALL() for the provider side, called at the end of pg_decode_begin_txn() right after BEGIN is flushed to the subscriber, so a test can make the walsender go quiet mid-transaction. Same SPOCK_RANDOM_DELAYS / USE_INJECTION_POINTS / neither pattern as the existing macro, including firing spock_random_delay() under SPOCK_RANDOM_DELAYS for parity. No-op in production; no behavioral change. --- include/spock_injection.h | 35 +++++++++++++++++++++++++---------- src/spock_injection.c | 8 ++++---- src/spock_output_plugin.c | 9 +++++++++ 3 files changed, 38 insertions(+), 14 deletions(-) diff --git a/include/spock_injection.h b/include/spock_injection.h index 498536b8b..59443a2f4 100644 --- a/include/spock_injection.h +++ b/include/spock_injection.h @@ -3,15 +3,26 @@ * spock_injection.h * Injection point support for the Spock extension. * - * Defines SPOCK_WORKER_DELAY(), placed at worker start/finish sites: + * Two named injection points are defined, one per side of the wire: * - * SPOCK_RANDOM_DELAYS defined – calls spock_random_delay() directly; + * SPOCK_WORKER_DELAY() – subscriber side, at apply-worker + * start/finish sites ('spock-worker-delay'). + * SPOCK_OUTPUT_TXN_STALL() – provider side, right after a transaction's + * BEGIN has been sent to the subscriber + * ('spock-output-txn-stall'). Lets a test + * simulate the walsender going quiet + * mid-transaction (slow decode, network + * stall) without an unconditional sleep or + * an ad-hoc getenv()/marker-file hook wired + * into production output-plugin code. + * + * SPOCK_RANDOM_DELAYS defined – both call spock_random_delay() directly; * fires unconditionally, no runtime setup. - * USE_INJECTION_POINTS defined – expands to INJECTION_POINT(); the core - * injection_points module can attach to - * 'spock-worker-delay' when needed. + * USE_INJECTION_POINTS defined – both expand to INJECTION_POINT(); the + * core injection_points module can + * attach to either name when needed. * Requires --enable-injection-points. - * neither defined – compiles to nothing. + * neither defined – both compile to nothing. * * Copyright (c) 2022-2026, pgEdge, Inc. * @@ -23,21 +34,25 @@ #ifdef SPOCK_RANDOM_DELAYS extern void spock_random_delay(void); -#define SPOCK_WORKER_DELAY() spock_random_delay() +#define SPOCK_WORKER_DELAY() spock_random_delay() +#define SPOCK_OUTPUT_TXN_STALL() spock_random_delay() #elif defined(USE_INJECTION_POINTS) #include "utils/injection_point.h" #if PG_VERSION_NUM >= 180000 -#define SPOCK_WORKER_DELAY() INJECTION_POINT("spock-worker-delay", NULL) +#define SPOCK_WORKER_DELAY() INJECTION_POINT("spock-worker-delay", NULL) +#define SPOCK_OUTPUT_TXN_STALL() INJECTION_POINT("spock-output-txn-stall", NULL) #else -#define SPOCK_WORKER_DELAY() INJECTION_POINT("spock-worker-delay") +#define SPOCK_WORKER_DELAY() INJECTION_POINT("spock-worker-delay") +#define SPOCK_OUTPUT_TXN_STALL() INJECTION_POINT("spock-output-txn-stall") #endif #else -#define SPOCK_WORKER_DELAY() ((void) 0) +#define SPOCK_WORKER_DELAY() ((void) 0) +#define SPOCK_OUTPUT_TXN_STALL() ((void) 0) #endif /* SPOCK_RANDOM_DELAYS / USE_INJECTION_POINTS */ diff --git a/src/spock_injection.c b/src/spock_injection.c index 01de21f42..16882b5f3 100644 --- a/src/spock_injection.c +++ b/src/spock_injection.c @@ -1,15 +1,15 @@ /*------------------------------------------------------------------------- * * spock_injection.c - * Unconditional random delay for Spock worker start/finish sites. + * Unconditional random delay for Spock injection point sites. * * spock_random_delay() is compiled in only when SPOCK_RANDOM_DELAYS is set * in the environment at build time (make SPOCK_RANDOM_DELAYS=1). It sleeps * for a random duration in [1, SPOCK_INJ_MAX_DELAY_MS] ms. * - * On PG17+ without SPOCK_RANDOM_DELAYS, SPOCK_WORKER_DELAY() expands to - * INJECTION_POINT("spock-worker-delay") instead -- attach a callback via - * the core injection_points module when needed. + * On PG17+ without SPOCK_RANDOM_DELAYS, the injection point macros in + * spock_injection.h expand to INJECTION_POINT() instead -- attach a + * callback via the core injection_points module when needed. * * Copyright (c) 2022-2026, pgEdge, Inc. * diff --git a/src/spock_output_plugin.c b/src/spock_output_plugin.c index d54007b7e..a874ca075 100644 --- a/src/spock_output_plugin.c +++ b/src/spock_output_plugin.c @@ -37,6 +37,7 @@ #include "spock_output_plugin.h" #include "spock.h" +#include "spock_injection.h" #include "spock_output_config.h" #include "spock_executor.h" #include "spock_node.h" @@ -651,6 +652,14 @@ pg_decode_begin_txn(LogicalDecodingContext *ctx, ReorderBufferTXN *txn) Assert(CurrentMemoryContext == data->context); MemoryContextSwitchTo(old_ctx); + + /* + * Test hook: let a test stall the walsender here, once BEGIN has already + * reached the subscriber (whose handle_begin() has recorded this + * transaction's commit_lsn), to reproduce a provider that goes quiet + * mid-transaction. No-op unless a test attaches to this injection point. + */ + SPOCK_OUTPUT_TXN_STALL(); } /* From 79f0de120b40be29695ad500826808e7860e92ef Mon Sep 17 00:00:00 2001 From: Asif Rehman Date: Thu, 3 Sep 2026 20:34:43 +0500 Subject: [PATCH 2/4] Add test reproducing apply-worker exception misclassification Two related bugs in the apply worker's restart/exit machinery: a non-error event is misread as evidence of a prior apply failure, entering exception replay and, under spock.exception_behaviour = sub_disable, disabling a healthy subscription. Scenario 1: a mid-transaction worker restart (e.g. sub_alter_options()) can match a stale exception_log commit_lsn on reconnect. Scenario 2: spock.apply_idle_timeout's error isn't tagged connection-class, so a stalled provider is misclassified as a data conflict; reproduced via the spock-output-txn-stall injection point, which requires --enable-injection-points. It also asserts the apply worker's PID actually changes across the stall (recorded via application_name = "spock apply :"), proving the idle-timeout reconnect path really fired rather than just observing that nothing broke -- a reconnect attempt right after the timeout typically fails immediately on "replication slot is active" (the old, still-stalled walsender hasn't released it) and the worker dies again in milliseconds, so an empty PID counts as evidence of restart too, not only a differing non-empty one. Both scenarios share one cluster/subscription. Only scenario 2 (and the provider restart it needs) is skipped when injection_points is unavailable -- scenario 1 has no such dependency and always runs. Predates both fixes, so scenario 1's 4 subtests and scenario 2's 7 (when not skipped) fail as expected; the next two commits fix each in turn. --- tests/tap/schedule | 1 + ...pply_worker_exception_misclassification.pl | 302 ++++++++++++++++++ 2 files changed, 303 insertions(+) create mode 100644 tests/tap/t/046_apply_worker_exception_misclassification.pl diff --git a/tests/tap/schedule b/tests/tap/schedule index 7e8c457ea..bbf162112 100644 --- a/tests/tap/schedule +++ b/tests/tap/schedule @@ -58,6 +58,7 @@ test: 037_wire_format_datestyle test: 038_reserved_schema_ddl_guard test: 044_apply_change_logging test: 045_lsn_from_commit_ts +test: 046_apply_worker_exception_misclassification # Upgrade schema match test (builds from source, slow): #test: 018_upgrade_schema_match # diff --git a/tests/tap/t/046_apply_worker_exception_misclassification.pl b/tests/tap/t/046_apply_worker_exception_misclassification.pl new file mode 100644 index 000000000..fd5eded3f --- /dev/null +++ b/tests/tap/t/046_apply_worker_exception_misclassification.pl @@ -0,0 +1,302 @@ +use strict; +use warnings; +use Test::More; +use lib '.'; +use SpockTest qw( + create_cluster destroy_cluster + get_test_config scalar_query psql_or_bail system_or_bail system_maybe + wait_for_sub_status wait_for_pg_ready +); + +# ============================================================================= +# Test: 046_apply_worker_exception_misclassification.pl +# ============================================================================= +# Two related bugs in the apply worker's restart/exit machinery, both found +# by auditing that code after the first was found: a non-error event gets +# misread as evidence of a prior apply failure, entering exception replay +# and, under spock.exception_behaviour = sub_disable, disabling a healthy +# subscription for a transaction that never actually errored. +# +# Scenario 1 -- mid-transaction worker restart: +# handle_begin() stamps exception_log->commit_lsn on every BEGIN, not just +# ones that fail. A worker restart (e.g. sub_alter_options() reconnecting +# to pick up a new setting) can land mid-transaction, and the provider's +# ordinary retransmission of that same transaction to the replacement +# worker matches the stale commit_lsn on its first BEGIN. +# +# Scenario 2 -- apply-idle-timeout misclassification: +# spock.apply_idle_timeout's "no data received... reconnecting" error was +# a bare elog(ERROR), defaulting to a sqlerrcode the PG_CATCH +# discriminator in apply_work() doesn't recognize as connection-class, so +# a merely-slow or stalled provider mid-transaction fell into the same +# branch as a genuine data conflict. +# +# Both scenarios run against one shared 2-node cluster and subscription. +# Scenario 2 needs PostgreSQL's core injection_points test module, which +# only exists when the server was configured with --enable-injection-points +# -- it is not part of an ordinary `make install`/`ninja install`. Scenario 1 +# has no such dependency and always runs; only scenario 2 is skipped when +# the module is unavailable. +# ============================================================================= + +my $config = get_test_config(); +my $pg_bin = $config->{pg_bin}; + +my $pkglibdir = `"$pg_bin/pg_config" --pkglibdir`; +chomp $pkglibdir; +my $has_injection_points = (-e "$pkglibdir/injection_points.so") + || (-e "$pkglibdir/injection_points.dylib"); +my $skip_reason = + "server not built with --enable-injection-points " . + "(no injection_points test module in $pkglibdir)"; + +create_cluster(2, 'Create 2-node apply-worker exception-misclassification cluster'); + +$config = get_test_config(); +my $host = $config->{host}; +my $dbname = $config->{db_name}; +my $db_user = $config->{db_user}; +my $db_password = $config->{db_password}; +my $p1 = $config->{node_ports}->[0]; +my $p2 = $config->{node_ports}->[1]; +my $datadir1 = $config->{node_datadirs}->[0]; +my $log_file = $config->{log_file}; +my $conn = "host=$host dbname=$dbname port=$p1 user=$db_user password=$db_password"; + +# n1 (the provider) needs injection_points in shared_preload_libraries for +# scenario 2 to attach test injection points -- that's a postmaster-start +# GUC, so add it and restart just this node, before any subscription +# exists. A later duplicate setting in postgresql.conf overrides the +# earlier one, so appending is enough. Skipped entirely when the module +# isn't available; scenario 1 doesn't need it. +SKIP: { + skip $skip_reason, 1 unless $has_injection_points; + + open(my $conf, '>>', "$datadir1/postgresql.conf") + or die "Cannot open $datadir1/postgresql.conf: $!"; + print $conf "shared_preload_libraries='spock,injection_points'\n"; + close($conf); + + system_or_bail("$pg_bin/pg_ctl", '-D', $datadir1, '-w', '-m', 'fast', 'stop'); + system("$pg_bin/postgres -D $datadir1 >> '$log_file' 2>&1 &"); + ok(wait_for_pg_ready($host, $p1, $pg_bin, 30), 'n1 restarted with injection_points preloaded'); + + psql_or_bail(1, "CREATE EXTENSION injection_points"); +} + +psql_or_bail(2, "ALTER SYSTEM SET spock.exception_behaviour = sub_disable"); +psql_or_bail(2, "ALTER SYSTEM SET spock.apply_idle_timeout = 3"); +psql_or_bail(2, "SELECT pg_reload_conf()"); +sleep(1); + +psql_or_bail(2, + "SELECT spock.sub_create('sub_n1_n2', '$conn', " . + "ARRAY['default', 'default_insert_only', 'ddl_sql'], true, false)"); +ok(wait_for_sub_status(2, 'sub_n1_n2', 'replicating', 30), + 'subscription starts in replicating state'); + +# ============================================================================= +# Scenario 1: mid-transaction worker restart +# ============================================================================= + +psql_or_bail(1, "CREATE TABLE midtxn_restart (id bigint PRIMARY KEY, val text)"); +psql_or_bail(1, "CREATE SEQUENCE midtxn_restart_id_seq"); + +# A PROCEDURE (not a DO block) can issue its own internal COMMITs, so a +# single long-lived connection produces a continuous stream of separately +# committed, multi-row transactions -- wide enough for a worker restart to +# land mid-transaction -- without the overhead of spawning one psql process +# per transaction. +psql_or_bail(1, q{ + CREATE PROCEDURE midtxn_restart_load(n_batches int, batch_rows int) + LANGUAGE plpgsql AS $$ + DECLARE i int; + BEGIN + FOR i IN 1..n_batches LOOP + INSERT INTO midtxn_restart + SELECT nextval('midtxn_restart_id_seq'), 'x' || g + FROM generate_series(1, batch_rows) g; + COMMIT; + -- Bulk INSERT...SELECT is fast enough to produce tens of + -- millions of rows in the toggle loop's ~12s window, far more + -- than row-by-row logical apply can drain afterwards. Throttle + -- so total volume stays boundable while each transaction is + -- still wide enough for a worker restart to land mid-flight. + PERFORM pg_sleep(0.02); + END LOOP; + END $$; +}); + +# Background writer, left running for the duration of the restart-toggle +# loop below. Killing the *client* process on its own would not reliably +# stop it -- Postgres does not proactively notice a dropped client +# connection while a backend is busy inside a single long-running CALL +# (no client_connection_check_interval by default) -- so it is stopped +# server-side afterwards via pg_terminate_backend(). +my $writer_pid = fork(); +die "fork() failed: $!" unless defined $writer_pid; +if ($writer_pid == 0) { + exec("$pg_bin/psql", '-X', '-p', $p1, '-d', $dbname, + '-c', "CALL midtxn_restart_load(100000, 500)") + or exit(127); +} + +# Restart the apply worker repeatedly while the writer is running, via +# sub_alter_options() toggling apply_delay -- the same mechanism that +# uncovered this bug in 049_bidir_join_under_load.pl's lag-injection +# scenario. +for my $i (1 .. 40) { + system_or_bail("$pg_bin/psql", '-X', '-p', $p2, '-d', $dbname, '-c', + "SELECT spock.sub_alter_options('sub_n1_n2', '{\"apply_delay\": \"1 millisecond\"}'::jsonb)"); + select(undef, undef, undef, 0.15); + system_or_bail("$pg_bin/psql", '-X', '-p', $p2, '-d', $dbname, '-c', + "SELECT spock.sub_alter_options('sub_n1_n2', '{\"apply_delay\": \"0\"}'::jsonb)"); + select(undef, undef, undef, 0.15); +} + +# Stop the writer at the server: terminate the backend actually running the +# CALL, then confirm it is gone before treating n1's row count as final. +system_or_bail("$pg_bin/psql", '-X', '-p', $p1, '-d', $dbname, '-c', + "SELECT pg_terminate_backend(pid) FROM pg_stat_activity " . + "WHERE query LIKE 'CALL midtxn_restart_load%' AND pid <> pg_backend_pid()"); +for (1 .. 30) { + my $still_running = scalar_query(1, + "SELECT count(*) FROM pg_stat_activity WHERE query LIKE 'CALL midtxn_restart_load%'"); + last if defined $still_running && $still_running eq '0'; + sleep(1); +} +kill('TERM', $writer_pid); +waitpid($writer_pid, 0); + +ok(wait_for_sub_status(2, 'sub_n1_n2', 'replicating', 60), + 'scenario 1: subscription is still replicating after repeated mid-load worker restarts'); + +is(scalar_query(2, + "SELECT sub_enabled FROM spock.subscription WHERE sub_name = 'sub_n1_n2'"), + 't', 'scenario 1: SUB_DISABLE subscription remains enabled after repeated restarts'); + +my $s1_count1 = scalar_query(1, "SELECT count(*) FROM midtxn_restart"); + +my $s1_count2; +for (1 .. 120) { + $s1_count2 = scalar_query(2, "SELECT count(*) FROM midtxn_restart"); + last if defined $s1_count2 && $s1_count2 eq $s1_count1; + sleep(1); +} +is($s1_count2, $s1_count1, 'scenario 1: no rows lost/discarded across the restarts'); + +is(scalar_query(2, "SELECT count(*) FROM spock.exception_log"), + '0', 'scenario 1: no exceptions were ever logged'); + +# ============================================================================= +# Scenario 2: apply-idle-timeout misclassification +# ============================================================================= + +SKIP: { + skip $skip_reason, 7 unless $has_injection_points; + + psql_or_bail(1, "CREATE TABLE idle_timeout_midtxn (id int PRIMARY KEY, val text)"); + + # Apply workers set application_name to "spock apply :" + # (see BackgroundWorkerInitializeConnection() in spock_worker.c); with + # one subscription on n2 this identifies it unambiguously. Recorded + # before the stall so a later PID change proves the idle-timeout error + # actually fired and drove a real worker restart through the + # ERRCODE_CONNECTION_FAILURE path -- not just that nothing else broke. + my $worker_pid_before = scalar_query(2, + "SELECT pid FROM pg_stat_activity WHERE application_name LIKE 'spock apply %'"); + + psql_or_bail(1, "SELECT injection_points_attach('spock-output-txn-stall', 'wait')"); + + psql_or_bail(1, q{ + BEGIN; + INSERT INTO idle_timeout_midtxn VALUES (1, 'before'); + INSERT INTO idle_timeout_midtxn VALUES (2, 'stalled-here'); + INSERT INTO idle_timeout_midtxn VALUES (3, 'after'); + COMMIT; + }); + + my $stalled = 0; + for (1 .. 30) { + my $n = scalar_query(1, + "SELECT count(*) FROM pg_stat_activity WHERE backend_type = 'walsender' " . + "AND wait_event = 'spock-output-txn-stall'"); + if (defined $n && $n >= 1) { + $stalled = 1; + last; + } + sleep(1); + } + ok($stalled, 'scenario 2: walsender hit the mid-transaction stall injection point'); + + # apply_idle_timeout is 3s, and the worker paces its respawn by + # spock.restart_delay_default (5s default) after the idle-timeout + # error, so the new PID isn't expected to appear for up to ~8s. Watch + # well past that and confirm both that the apply worker actually + # restarts through the idle-timeout reconnect path (proving the fix's + # code path executed) and that the subscription is never disabled + # while doing so. + my $worker_restarted = 0; + my $disabled_during_stall = 0; + for (1 .. 20) { + my $enabled = scalar_query(2, + "SELECT sub_enabled FROM spock.subscription WHERE sub_name = 'sub_n1_n2'"); + if (defined $enabled && $enabled eq 'f') { + $disabled_during_stall = 1; + last; + } + my $worker_pid_now = scalar_query(2, + "SELECT pid FROM pg_stat_activity WHERE application_name LIKE 'spock apply %'"); + + # A reconnect attempt right after the idle-timeout exit typically + # fails immediately with "replication slot is active" (the old, + # still-stalled walsender hasn't released it yet) and the worker + # dies again in milliseconds -- far faster than this 1s polling + # interval can catch it alive with a new PID. So treat the PID + # going empty as proof of restart too, not just a differing + # non-empty PID: the original PID is guaranteed non-empty (the + # subscription was already confirmed replicating), so either + # outcome means the original worker is gone. + if (defined $worker_pid_now && $worker_pid_now ne $worker_pid_before) { + $worker_restarted = 1; + } + last if $worker_restarted; + sleep(1); + } + ok($worker_restarted, + 'scenario 2: apply worker restarted through the idle-timeout reconnect path'); + ok(!$disabled_during_stall, + 'scenario 2: subscription stays enabled while the provider is stalled past apply_idle_timeout'); + + # If scenario 1's bug already disabled the subscription, nothing ever + # decoded this transaction and the injection point was never reached -- + # waking a point nobody is waiting on is then an error in its own right, + # not a symptom worth crashing this test over. Best-effort only. + system_maybe("$pg_bin/psql", '-X', '-p', $p1, '-d', $dbname, '-c', + "SELECT injection_points_wakeup('spock-output-txn-stall')"); + system_maybe("$pg_bin/psql", '-X', '-p', $p1, '-d', $dbname, '-c', + "SELECT injection_points_detach('spock-output-txn-stall')"); + + my $s2_count1 = scalar_query(1, "SELECT count(*) FROM idle_timeout_midtxn"); + my $s2_count2; + for (1 .. 60) { + $s2_count2 = scalar_query(2, "SELECT count(*) FROM idle_timeout_midtxn"); + last if defined $s2_count2 && $s2_count2 eq $s2_count1; + sleep(1); + } + is($s2_count2, $s2_count1, + 'scenario 2: the stalled transaction applies in full once released, no rows lost'); + + is(scalar_query(2, + "SELECT sub_enabled FROM spock.subscription WHERE sub_name = 'sub_n1_n2'"), + 't', 'scenario 2: subscription is not disabled by the idle-timeout liveness reconnect'); + + ok(wait_for_sub_status(2, 'sub_n1_n2', 'replicating', 30), + 'scenario 2: subscription returns to replicating after the reconnect'); + + is(scalar_query(2, "SELECT count(*) FROM spock.exception_log"), + '0', 'scenario 2: no spurious exception_log entry for the idle-timeout liveness reconnect'); +} + +destroy_cluster('Destroy apply-worker exception-misclassification cluster'); +done_testing(); From 98aae5b7489dd0a9db5cb3a80237e14ba3576ef8 Mon Sep 17 00:00:00 2001 From: Asif Rehman Date: Thu, 3 Sep 2026 20:35:22 +0500 Subject: [PATCH 3/4] Fix spurious SUB_DISABLE on mid-transaction worker restart handle_begin() stamps exception_log->commit_lsn on every BEGIN, not just failures, and nothing cleared it on a clean SIGTERM exit. A restart landing mid-transaction let the provider's retransmission of that same transaction match the stale commit_lsn and get misread as a prior failure, disabling the subscription for no real error. Clear the marker via the existing clear_transient_exception_state() helper when apply_work() exits on got_SIGTERM. Fixes scenario 1 of 046_apply_worker_exception_misclassification.pl. --- src/spock_apply.c | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/src/spock_apply.c b/src/spock_apply.c index b1f34b436..9dc55e588 100644 --- a/src/spock_apply.c +++ b/src/spock_apply.c @@ -4123,6 +4123,21 @@ apply_work(PGconn *streamConn) goto stream_replay; } + /* + * A SIGTERM (e.g. sub_alter_options() restarting the worker to pick up a + * new setting) can land while a transaction is mid-flight: handle_begin() + * has already recorded its commit_lsn, but neither xact_had_exception nor + * an ERROR has occurred. The provider retransmits that same transaction + * to the replacement worker, which would otherwise misread the matching + * commit_lsn as a prior apply failure and enter exception replay under + * the configured policy. Clear the marker so the retransmission is + * applied normally; see clear_transient_exception_state()'s own comment + * for why this is safe (it no-ops when a genuine exception is already + * recorded). + */ + if (got_SIGTERM) + clear_transient_exception_state("subscription worker restart"); + elog(LOG, "SPOCK %s: falling out of apply_work() sigterm=%s", MySubscription->name, (got_SIGTERM) ? "true" : "false"); } From d7c74d2b8df79f294ddb61e68db38c5dc1ea02a5 Mon Sep 17 00:00:00 2001 From: Asif Rehman Date: Thu, 3 Sep 2026 20:35:49 +0500 Subject: [PATCH 4/4] Fix spurious apply-idle-timeout misclassification as data exception spock.apply_idle_timeout's "no data received... reconnecting" error was a bare elog(ERROR), defaulting to ERRCODE_INTERNAL_ERROR. The PG_CATCH discriminator in apply_work() only recognizes connection-class sqlerrcodes, so a merely-slow or stalled provider mid-transaction fell into the same branch as a genuine data conflict, spuriously disabling the subscription (or silently discarding a good transaction). Tag the error ERRCODE_CONNECTION_FAILURE, consistent with the other stream-level errors nearby, routing it through the already-correct connection-class branch. Fixes scenario 2 of 046_apply_worker_exception_misclassification.pl; all subtests now pass (or skip cleanly without --enable-injection-points). --- src/spock_apply.c | 22 +++++++++++++++++++--- 1 file changed, 19 insertions(+), 3 deletions(-) diff --git a/src/spock_apply.c b/src/spock_apply.c index 9dc55e588..e5e84ae72 100644 --- a/src/spock_apply.c +++ b/src/spock_apply.c @@ -3530,6 +3530,20 @@ apply_work(PGconn *streamConn) * safety net for the case where the walsender process is alive * but hung -- TCP probes succeed because the kernel ACKs them, * but no data is being sent. + * + * This can fire mid-transaction: handle_begin() may already have + * recorded the in-flight transaction's commit_lsn when the + * provider goes quiet for longer than apply_idle_timeout (a slow + * decode of a large transaction, a network stall). It is a + * liveness condition, not a data fault, so it must be tagged + * ERRCODE_CONNECTION_FAILURE like the other stream-level errors + * above -- otherwise the PG_CATCH discriminator below cannot + * distinguish it from a genuine apply exception, and would + * misclassify a merely-slow provider as a permanent data + * conflict: entering the same-process exception replay path for a + * transaction that never actually failed, and disabling the + * subscription (SUB_DISABLE) or discarding real rows + * (DISCARD/TRANSDISCARD) with nothing having gone wrong. */ if (rc & WL_TIMEOUT && spock_apply_idle_timeout > 0) { @@ -3540,9 +3554,11 @@ apply_work(PGconn *streamConn) if (GetCurrentTimestamp() > timeout) { MySpockWorker->worker_status = SPOCK_WORKER_STATUS_STOPPED; - elog(ERROR, "SPOCK %s: no data received for %d seconds, " - "reconnecting (spock.apply_idle_timeout)", - MySubscription->name, spock_apply_idle_timeout); + ereport(ERROR, + (errcode(ERRCODE_CONNECTION_FAILURE), + errmsg("SPOCK %s: no data received for %d seconds, " + "reconnecting (spock.apply_idle_timeout)", + MySubscription->name, spock_apply_idle_timeout))); } }