Fix SUB_DISABLE reconnect retransmission

Author: rokoshy

src/spock_apply.c

@@ -264,6 +264,7 @@ static void spock_apply_worker_attach(void);
 static void spock_apply_worker_detach(void);
 
 static bool should_log_exception(bool failed);
+static void clear_transient_exception_state_on_connection_loss(void);
 
 static ApplyReplayEntry * apply_replay_entry_create(int r, char *buf);
 static void apply_replay_entry_free(ApplyReplayEntry * entry);
@@ -361,6 +362,47 @@ should_log_exception(bool failed)
 	return false;
 }
 
+/*
+ * Forget the in-flight transaction marker after a transport failure on the
+ * normal apply path.
+ *
+ * handle_begin() records commit_lsn in shared memory before any changes are
+ * applied.  That marker normally lets a restarted worker recognize a
+ * transaction which failed during apply and replay it under the configured
+ * exception policy.  A provider disconnect is different: PostgreSQL aborts
+ * the local transaction and the replication origin is deliberately left at
+ * the last durable commit so the provider can send the transaction again.
+ *
+ * Leaving commit_lsn behind makes the replacement worker misclassify that
+ * retransmitted transaction as an apply failure.  In SUB_DISABLE mode it then
+ * performs a read-only exception replay and disables the subscription instead
+ * of applying the transaction.  Clear the marker only for a first-pass
+ * transport failure.  Genuine apply failures and failures during exception
+ * replay retain their state.
+ */
+static void
+clear_transient_exception_state_on_connection_loss(void)
+{
+	SpockExceptionLog *exception_log;
+
+	if (!in_remote_transaction ||
+		MyApplyWorker == NULL ||
+		MyApplyWorker->use_try_block ||
+		xact_had_exception ||
+		exception_log_ptr == NULL ||
+		my_exception_log_index < 0)
+		return;
+
+	exception_log = &exception_log_ptr[my_exception_log_index];
+	exception_log->commit_lsn = InvalidXLogRecPtr;
+	exception_log->local_tuple = NULL;
+	exception_log->initial_error_message[0] = '\0';
+	exception_log->failed_action = 0;
+
+	elog(DEBUG1, "SPOCK %s: cleared transient exception state after provider connection loss",
+		 MySubscription->name);
+}
+
 /*
  * Check if given relation is in process of being synchronized.
  *
@@ -3692,6 +3734,7 @@ apply_work(PGconn *streamConn)
 			edata->sqlerrcode == ERRCODE_ADMIN_SHUTDOWN ||
 			(applyconn != NULL && PQstatus(applyconn) == CONNECTION_BAD))
 		{
+			clear_transient_exception_state_on_connection_loss();
 			elog(LOG, "SPOCK %s: connection error during apply, exiting via rethrow: %s",
 				 MySubscription->name, edata->message);
 			PG_RE_THROW();

tests/tap/schedule

@@ -57,3 +57,4 @@ test: 044_apply_change_logging
 
 # Regression tests
 test: 103_manager_worker_dboid_race
+test: 105_sub_disable_retransmit_after_disconnect

tests/tap/t/105_sub_disable_retransmit_after_disconnect.pl

@@ -0,0 +1,92 @@
+use strict;
+use warnings;
+use Test::More;
+use lib '.';
+use SpockTest qw(
+    create_cluster destroy_cluster
+    get_test_config scalar_query psql_or_bail
+    wait_for_sub_status
+);
+
+# A connection-class error after BEGIN aborts the local transaction and makes
+# the provider retransmit it.  The exception marker recorded by handle_begin()
+# must not make the replacement worker treat that valid retransmission as
+# exception replay under SUB_DISABLE.
+
+create_cluster(2, 'Create 2-node reconnect retransmission test cluster');
+
+my $config = get_test_config();
+my $p1 = $config->{node_ports}->[0];
+my $p2 = $config->{node_ports}->[1];
+my $conn = "host=$config->{host} dbname=$config->{db_name} port=$p1 " .
+           "user=$config->{db_user} password=$config->{db_password}";
+my $subscriber_log = "$config->{log_dir}/00${p2}.log";
+
+psql_or_bail(2, "ALTER SYSTEM SET spock.exception_behaviour = sub_disable");
+psql_or_bail(2, "SELECT pg_reload_conf()");
+
+psql_or_bail(1, "CREATE TABLE reconnect_retransmit (id int PRIMARY KEY, val text)");
+psql_or_bail(2, "CREATE TABLE reconnect_retransmit (id int PRIMARY KEY, val text)");
+
+# Sequence increments are non-transactional, so SQLSTATE 08006 is raised only
+# on the first apply attempt.  ENABLE REPLICA restricts the trigger to apply.
+psql_or_bail(2, q{
+    CREATE SEQUENCE reconnect_fail_once;
+    CREATE FUNCTION reconnect_fail_once() RETURNS trigger LANGUAGE plpgsql AS $$
+    BEGIN
+        IF nextval('reconnect_fail_once') = 1 THEN
+            RAISE EXCEPTION 'injected provider connection loss'
+                USING ERRCODE = '08006';
+        END IF;
+        RETURN NEW;
+    END $$;
+    CREATE TRIGGER reconnect_fail_once_trg
+        BEFORE INSERT ON reconnect_retransmit
+        FOR EACH ROW EXECUTE FUNCTION reconnect_fail_once();
+    ALTER TABLE reconnect_retransmit
+        ENABLE REPLICA TRIGGER reconnect_fail_once_trg;
+});
+
+psql_or_bail(2,
+    "SELECT spock.sub_create('sub_n1_n2', '$conn', " .
+    "ARRAY['default', 'default_insert_only', 'ddl_sql'], false, false)");
+ok(wait_for_sub_status(2, 'sub_n1_n2', 'replicating', 30),
+    'subscription starts in replicating state');
+
+my $log_offset = -s $subscriber_log // 0;
+psql_or_bail(1, "INSERT INTO reconnect_retransmit VALUES (1, 'must survive reconnect')");
+
+my $applied = 0;
+for (1 .. 60) {
+    $applied = scalar_query(2,
+        "SELECT count(*) FROM reconnect_retransmit WHERE id = 1");
+    last if defined $applied && $applied eq '1';
+    sleep(1);
+}
+is($applied, '1', 'retransmitted transaction is applied after worker restart');
+
+is(scalar_query(2,
+       "SELECT sub_enabled FROM spock.subscription WHERE sub_name = 'sub_n1_n2'"),
+   't', 'SUB_DISABLE subscription remains enabled');
+ok(wait_for_sub_status(2, 'sub_n1_n2', 'replicating', 30),
+    'subscription returns to replicating state');
+is(scalar_query(2, "SELECT last_value FROM reconnect_fail_once"),
+   '2', 'replica trigger proves the transaction was attempted twice');
+
+my $new_log = '';
+if (open(my $lf, '<', $subscriber_log)) {
+    seek($lf, $log_offset, 0);
+    local $/;
+    $new_log = <$lf> // '';
+    close($lf);
+}
+
+like($new_log,
+     qr/cleared transient exception state after provider connection loss/,
+     'connection-loss path clears transient exception state');
+unlike($new_log,
+       qr/Transaction failed, subscription will be disabled/,
+       'retransmission does not enter SUB_DISABLE exception replay');
+
+destroy_cluster('Destroy reconnect retransmission test cluster');
+done_testing();