Fix/spurious subdisable mid transaction restarts and apply-idle-timeout misclassification

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 */
 

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)));
 				}
 			}
 
@@ -4123,6 +4139,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");
 }

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.
  *

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();
 }
 
 /*

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
 #