diff --git a/src/spock_relcache.c b/src/spock_relcache.c index b30bb848..b5f82551 100644 --- a/src/spock_relcache.c +++ b/src/spock_relcache.c @@ -29,6 +29,7 @@ #include "spock.h" #include "spock_common.h" #include "spock_relcache.h" +#include "spock_worker.h" #define SPOCKRELATIONHASH_INITIAL_SIZE 128 static HTAB *SpockRelationHash = NULL; @@ -102,6 +103,8 @@ spock_relation_open(uint32 remoteid, LOCKMODE lockmode) int i; TupleDesc desc; ResultRelInfo *relinfo; + StringInfoData missing; + int num_missing = 0; rv->schemaname = (char *) entry->nspname; rv->relname = (char *) entry->relname; @@ -141,6 +144,26 @@ spock_relation_open(uint32 remoteid, LOCKMODE lockmode) entry->attmap[i] = tupdesc_get_att_by_name(desc, entry->attnames[i]); + /* + * A missing local column is handled like a missing local + * relation. Collect every missing column rather than stopping at + * the first, so an operator repairing a drifted schema gets the + * whole list at once instead of one column per apply attempt. + * Skip the rest of the loop body: it indexes the local tupdesc by + * attmap[i]. + */ + if (unlikely(entry->attmap[i] < 0)) + { + if (num_missing == 0) + initStringInfo(&missing); + else + appendStringInfoString(&missing, ", "); + + appendStringInfo(&missing, "\"%s\"", entry->attnames[i]); + num_missing++; + continue; + } + /* * If we find attribute options for this column and the * delta_apply_function is set, lookup the oid for it. @@ -172,6 +195,30 @@ spock_relation_open(uint32 remoteid, LOCKMODE lockmode) } } + /* + * Report the mismatch on the first pass so apply_work() records the + * cause, then return NULL during exception replay so the caller can + * apply the configured exception behaviour instead of erroring again. + * Close with NoLock like every other apply-path close: the lock stays + * until the replication transaction ends rather than being dropped on + * a table we are part way through. + */ + if (unlikely(num_missing > 0)) + { + if (MyApplyWorker == NULL || !MyApplyWorker->use_try_block) + ereport(ERROR, + (errcode(ERRCODE_UNDEFINED_COLUMN), + errmsg_plural("unknown column name %s in relation \"%s\".\"%s\"", + "unknown column names %s in relation \"%s\".\"%s\"", + num_missing, + missing.data, entry->nspname, + entry->relname))); + + pfree(missing.data); + spock_relation_close(entry, NoLock); + return NULL; + } + relinfo = makeNode(ResultRelInfo); InitResultRelInfo(relinfo, entry->rel, 1, NULL, 0); entry->reloid = RelationGetRelid(entry->rel); @@ -249,7 +296,10 @@ spock_relation_cache_update(uint32 remoteid, char *schemaname, entry->delta_apply_functions = palloc0(natts * sizeof(Oid)); MemoryContextSwitchTo(oldcontext); - /* XXX Should we validate the relation against local schema here? */ + /* + * Local-schema validation requires the lock taken by + * spock_relation_open(). + */ entry->reloid = InvalidOid; } @@ -287,7 +337,10 @@ spock_relation_cache_updater(SpockRemoteRel *remoterel) entry->delta_apply_functions = palloc0(remoterel->natts * sizeof(Oid)); MemoryContextSwitchTo(oldcontext); - /* XXX Should we validate the relation against local schema here? */ + /* + * Local-schema validation requires the lock taken by + * spock_relation_open(). + */ entry->reloid = InvalidOid; } @@ -381,6 +434,10 @@ spock_relcache_init(void) /* * Find attribute index in TupleDesc struct by attribute name. + * + * Returns -1 if no such column exists locally; the caller decides how to + * react (this can legitimately happen on a schema mismatch between nodes, + * which is not this function's business to escalate). */ static int tupdesc_get_att_by_name(TupleDesc desc, const char *attname) @@ -395,7 +452,7 @@ tupdesc_get_att_by_name(TupleDesc desc, const char *attname) return i; } - elog(ERROR, "unknown column name %s", attname); + return -1; } diff --git a/tests/tap/schedule b/tests/tap/schedule index 51c49a02..6b4012c9 100644 --- a/tests/tap/schedule +++ b/tests/tap/schedule @@ -67,3 +67,4 @@ test: 046_apply_worker_exception_misclassification # Regression tests test: 103_manager_worker_dboid_race test: 105_sub_disable_retransmit_after_disconnect +test: 108_apply_unknown_column_exception diff --git a/tests/tap/t/108_apply_unknown_column_exception.pl b/tests/tap/t/108_apply_unknown_column_exception.pl new file mode 100644 index 00000000..fd986fa2 --- /dev/null +++ b/tests/tap/t/108_apply_unknown_column_exception.pl @@ -0,0 +1,226 @@ +use strict; +use warnings; +use Test::More; +use lib '.'; +use SpockTest qw( + create_cluster destroy_cluster + system_or_bail system_maybe command_ok + get_test_config scalar_query psql_or_bail + wait_for_sub_status wait_for_exception_log wait_for_pg_ready +); + +# ============================================================================= +# Test 108: apply-time schema mismatch ("unknown column name") does not +# crash-loop the apply worker +# ============================================================================= +# Reproduce a provider-only column with DDL replication disabled. The +# mismatch is detected before row-level exception handling begins. Under +# transdiscard, each bad transaction must be logged and discarded while the +# apply worker remains healthy and unrelated tables continue to replicate. +# ============================================================================= + +create_cluster(2, 'Create 2-node cluster for unknown-column exception test'); + +my $config = get_test_config(); +my $node_ports = $config->{node_ports}; +my $node_datadirs = $config->{node_datadirs}; +my $host = $config->{host}; +my $dbname = $config->{db_name}; +my $db_user = $config->{db_user}; +my $db_password = $config->{db_password}; +my $pg_bin = $config->{pg_bin}; + +my $p1 = $node_ports->[0]; # n1 - provider +my $p2 = $node_ports->[1]; # n2 - subscriber +my $n2_datadir = $node_datadirs->[1]; + +my $conn_n1 = "host=$host dbname=$dbname port=$p1 user=$db_user password=$db_password"; + +# PG log file for n2, to look for the crash-loop log lines directly. +my $pg_log_n2 = "$config->{log_dir}/00${p2}.log"; + +# Force spock.exception_behaviour = transdiscard on n2 explicitly (it is the +# default, but pin it so the test does not depend on that default). +open(my $fh, '>>', "$n2_datadir/postgresql.conf") + or die "Cannot append to postgresql.conf: $!"; +print $fh "spock.exception_behaviour=transdiscard\n"; +close($fh); +psql_or_bail(2, "SELECT pg_reload_conf()"); +sleep(2); + +# --------------------------------------------------------------------------- +# Create the schema mismatch directly: n1 has an extra column n2 lacks. +# DDL replication is off so the CREATE TABLE itself never has to replicate -- +# this isolates the DML-apply bug from anything DDL-replication related. +# +# Once t1 diverges like this, its logical row images always carry all 3 +# columns (Postgres decodes the whole stored row, not just the columns an +# INSERT statement happened to name), so every future change to t1 is +# expected to keep failing -- t1_control, with identical schema on both +# nodes, is the control used to prove general replication health. +# --------------------------------------------------------------------------- + +psql_or_bail(1, + "SET spock.enable_ddl_replication = off; " . + "CREATE TABLE t1 (a INT PRIMARY KEY, b TEXT, c TEXT); " . + "SELECT spock.repset_add_table('default', 't1')"); + +psql_or_bail(2, "CREATE TABLE t1 (a INT PRIMARY KEY, b TEXT)"); + +# A control table with identical schema on both nodes, used to prove the +# apply worker is alive and replicating both before and after the bad +# transaction. +psql_or_bail(1, + "SET spock.enable_ddl_replication = off; " . + "CREATE TABLE t1_control (a INT PRIMARY KEY, b TEXT); " . + "SELECT spock.repset_add_table('default', 't1_control')"); +psql_or_bail(2, "CREATE TABLE t1_control (a INT PRIMARY KEY, b TEXT)"); + +psql_or_bail(2, + "SELECT spock.sub_create('sub_n1_n2', '$conn_n1', " . + "ARRAY['default', 'default_insert_only'], false, false)"); + +ok(wait_for_sub_status(2, 'sub_n1_n2', 'replicating', 30), + 'sub_n1_n2 reaches replicating state'); + +# Baseline: a schema-consistent table replicates fine before we do anything +# to t1. +psql_or_bail(1, "INSERT INTO t1_control (a, b) VALUES (1, 'baseline')"); + +my $baseline_ok = 0; +for (1..30) { + sleep(1); + my $v = scalar_query(2, "SELECT count(*) FROM t1_control WHERE a = 1"); + if (defined $v && $v eq '1') { $baseline_ok = 1; last; } +} +ok($baseline_ok, 'baseline row on the control table replicates from n1 to n2'); + +psql_or_bail(2, "TRUNCATE spock.exception_log"); +my $exc_before = scalar_query(2, "SELECT count(*) FROM spock.exception_log"); + +# Record the n2 log offset so later checks only look at what this test adds. +my $log_offset = -s $pg_log_n2 // 0; + +# Trigger the bug: insert into the table n2 has a stale/narrower schema for. +psql_or_bail(1, "INSERT INTO t1 VALUES (4, 'data4', 'data4')"); + +# --------------------------------------------------------------------------- +# Core regression checks +# --------------------------------------------------------------------------- + +# Without the fix this never happens -- the "unknown column name" error +# bypasses exception_behaviour/exception_log entirely and the apply worker +# crash-loops instead, so this count never grows. +my $got_exception_row = 0; +for (1..30) { + sleep(1); + my $cnt = scalar_query(2, "SELECT count(*) FROM spock.exception_log"); + if (defined $cnt && $cnt > $exc_before) { $got_exception_row = 1; last; } +} +ok($got_exception_row, + 'exception_log gains an entry for the unknown-column transaction ' + . '(without the fix, this never happens and the worker crash-loops)'); + +my $err_msg = scalar_query(2, + "SELECT error_message FROM spock.exception_log " . + "ORDER BY retry_errored_at DESC LIMIT 1"); +isnt($err_msg, '', 'exception_log entry has a non-empty error_message'); + +# The exception_log row itself uses the same generic "discarded" wording as +# the pre-existing missing-relation case (log_insert_exception's literal is +# not specific to this failure); the real, specific cause is what actually +# matters for diagnosis and is what an operator would grep for, so check it +# in the server log instead. +my $log_at_error = ''; +if (open(my $lf, '<', $pg_log_n2)) { + seek($lf, $log_offset, 0); + local $/; + $log_at_error = <$lf> // ''; + close($lf); +} +like($log_at_error, qr/unknown column name "c" in relation "public"\."t1"/, + 'n2 server log names the real cause: unknown column "c" on relation t1'); + +# TRANSDISCARD: the whole offending transaction is rolled back, not applied. +my $row4 = scalar_query(2, "SELECT count(*) FROM t1 WHERE a = 4"); +is($row4, '0', 'row referencing the missing column is not applied on n2 (TRANSDISCARD)'); + +# The subscription must still be up -- not disabled, not stuck restarting. +ok(wait_for_sub_status(2, 'sub_n1_n2', 'replicating', 30), + 'sub_n1_n2 stays in replicating state (no crash loop, no SUB_DISABLE)'); + +# Replication of other transactions must continue. Insert into the control +# table after the bad transaction and confirm it still replicates. +psql_or_bail(1, "INSERT INTO t1_control (a, b) VALUES (100, 'after_bad_txn')"); + +my $post_replicated = 0; +for (1..30) { + sleep(1); + my $v = scalar_query(2, "SELECT count(*) FROM t1_control WHERE a = 100"); + if (defined $v && $v eq '1') { $post_replicated = 1; last; } +} +ok($post_replicated, + 'a later, unrelated transaction still replicates after the unknown-column ' + . 'transaction (replication did not stop)'); + +# A second, independent row on the still-mismatched t1 must also be handled +# gracefully -- proving this is not a one-shot fluke and there is no +# creeping crash-loop building up under repeated failures. +my $exc_before_2 = scalar_query(2, "SELECT count(*) FROM spock.exception_log"); +psql_or_bail(1, "INSERT INTO t1 VALUES (5, 'data5', 'data5')"); + +my $got_second_exception_row = 0; +for (1..30) { + sleep(1); + my $cnt = scalar_query(2, "SELECT count(*) FROM spock.exception_log"); + if (defined $cnt && $cnt > $exc_before_2) { $got_second_exception_row = 1; last; } +} +ok($got_second_exception_row, + 'a second, independent unknown-column transaction is also discarded and logged ' + . '(no degradation after the first occurrence)'); + +ok(wait_for_sub_status(2, 'sub_n1_n2', 'replicating', 30), + 'sub_n1_n2 still replicating after a second unknown-column transaction'); + +# And the control table must still be unaffected. +psql_or_bail(1, "INSERT INTO t1_control (a, b) VALUES (101, 'after_second_bad_txn')"); + +my $post_replicated_2 = 0; +for (1..30) { + sleep(1); + my $v = scalar_query(2, "SELECT count(*) FROM t1_control WHERE a = 101"); + if (defined $v && $v eq '1') { $post_replicated_2 = 1; last; } +} +ok($post_replicated_2, + 'the control table keeps replicating after two unknown-column transactions'); + +# Confirm there is no infinite-restart signature in the n2 log for this +# window (repeated "error during exception handling" is the crash-loop's +# fingerprint -- distinct from "caught initial exception", which +# legitimately fires once per new transaction against the still-mismatched +# table). +my $new_log = ''; +if (open(my $lf, '<', $pg_log_n2)) { + seek($lf, $log_offset, 0); + local $/; + $new_log = <$lf> // ''; + close($lf); +} +my $retry_storm = () = ($new_log =~ /error during exception handling/g); +is($retry_storm, 0, + "no crash-loop signature in n2 log for the unknown-column transactions " + . "(found $retry_storm occurrences)"); + +# --------------------------------------------------------------------------- +# Cleanup +# --------------------------------------------------------------------------- + +system_maybe("$pg_bin/psql", '-h', $host, '-p', $p2, '-U', $db_user, '-d', $dbname, + '-c', "SELECT spock.sub_disable('sub_n1_n2')"); +sleep(2); +system_maybe("$pg_bin/psql", '-h', $host, '-p', $p2, '-U', $db_user, '-d', $dbname, + '-c', "SELECT spock.sub_drop('sub_n1_n2')"); + +destroy_cluster('Destroy cluster after unknown-column exception test'); + +done_testing();