From 00c26def9a00da7c9f866206373cebfe53867dba Mon Sep 17 00:00:00 2001 From: Ibrar Ahmed Date: Wed, 2 Sep 2026 13:59:24 +0500 Subject: [PATCH] Do not wedge table sync when the target already holds rows. A COPY into a populated table aborts on the first duplicate key and the table is left at SYNC_STATUS_FAILED. After that the apply worker drops every change for it, so the table stops replicating and nothing says so. Easy to hit: adding a populated table to a replication set with synchronize_data := true makes every peer copy rows it already has. Copy into a temp staging table and merge with ON CONFLICT DO NOTHING, keeping the local rows. The staging table needs INCLUDING DEFAULTS, GENERATED and IDENTITY; a bare LIKE takes the NOT NULL but not what fills it. Recheck the empty case under an EXCLUSIVE lock, since a row can land between the check and the COPY. All tables share one transaction, so the lock is bounded and taken in a savepoint, and we fall back to staging if it is not granted. New GUC spock.sync_stage_and_merge, default off. With it off the COPY goes straight into the table as before. Also warn when a table first enters SYNC_STATUS_FAILED, and give the sub_resync_table() call that clears it. --- docs/configuring.md | 30 ++ include/spock.h | 1 + src/spock.c | 16 + src/spock_apply.c | 26 ++ src/spock_sync.c | 321 ++++++++++++++++++++- tests/tap/t/014_pgdump_restore_conflict.pl | 4 +- 6 files changed, 390 insertions(+), 8 deletions(-) diff --git a/docs/configuring.md b/docs/configuring.md index 883d36e4a..2bba1fef0 100644 --- a/docs/configuring.md +++ b/docs/configuring.md @@ -252,6 +252,36 @@ liveness detection. Default: `300` (5 minutes). spock.apply_idle_timeout = 300 ``` +### `spock.sync_stage_and_merge` + +Controls how the initial table copy behaves when the table on the subscriber +already holds rows. Default: `off`. + +With this `off`, the copy is a direct `COPY` into the table, which is what +Spock has always done. If the table is not empty, that `COPY` aborts on the +first duplicate key, the table's entry in `spock.local_sync_status` is left at +`failed`, and from that point the apply worker discards every change for the +table - so it stops replicating and silently diverges. + +With this `on`, a table that already holds rows is copied into a temporary +staging table instead and then merged with `ON CONFLICT DO NOTHING`. Rows +already present locally are kept rather than overwritten, so a sync of data the +node already has converges instead of wedging. + +This matters most when adding an already-populated table to a replication set +with `synchronize_data := true`, which asks every peer to copy rows it may +already hold. + +The copy runs in Spock's sync worker rather than in your session, so a +session-level `SET` has no effect on it. Set it in `postgresql.conf`, or with +`ALTER SYSTEM SET spock.sync_stage_and_merge = on` followed by +`SELECT pg_reload_conf()`, and set it on the subscriber - the node receiving +the copy. + +``` +spock.sync_stage_and_merge = off +``` + ### `spock.sync_timeout` Overrides the time (in seconds) budgeted for a single synchronisation wait in diff --git a/include/spock.h b/include/spock.h index 61e150556..c16acc2d0 100644 --- a/include/spock.h +++ b/include/spock.h @@ -51,6 +51,7 @@ extern int restart_delay_on_exception; extern int spock_replay_queue_size; extern int spock_pause_timeout; extern int spock_sync_timeout; +extern bool spock_sync_stage_and_merge; extern int spock_read_retry_count; extern bool check_all_uc_indexes; extern bool spock_enable_quiet_mode; diff --git a/src/spock.c b/src/spock.c index e64357eab..6c8465815 100644 --- a/src/spock.c +++ b/src/spock.c @@ -178,6 +178,9 @@ int spock_pause_timeout = 10; /* seconds to wait for apply workers * to pause */ int spock_sync_timeout = 0; /* seconds per sync wait; 0 = routine's * own default */ +bool spock_sync_stage_and_merge = false; /* stage the initial COPY + * when the target is not + * empty */ int spock_read_retry_count = 5; /* heap update/delete: retries when * local tuple is missing */ bool check_all_uc_indexes = false; @@ -1324,6 +1327,19 @@ _PG_init(void) 0, NULL, NULL, NULL); + DefineCustomBoolVariable("spock.sync_stage_and_merge", + "Load the initial table copy through a staging table when the target already holds rows.", + "Off by default, which keeps the direct COPY: a target that is not " + "empty aborts on the first duplicate key and the table's sync status " + "stays failed. On, such a table is copied into a temporary staging " + "table and merged with ON CONFLICT DO NOTHING, so rows already " + "present are kept and the sync converges instead.", + &spock_sync_stage_and_merge, + false, + PGC_USERSET, + 0, + NULL, NULL, NULL); + DefineCustomIntVariable("spock.restart_delay_default", "Default apply-worker restart delay in ms", NULL, diff --git a/src/spock_apply.c b/src/spock_apply.c index b1f34b436..f13bdb22c 100644 --- a/src/spock_apply.c +++ b/src/spock_apply.c @@ -4471,7 +4471,33 @@ process_syncing_tables(XLogRecPtr end_lsn) /* * Failed SYNC operation should be ignored until someone * processes the error and changes the status. + * + * Say so once, on the transition. From here on every change + * for this table is dropped by should_apply_changes_for_rel(), + * so the table stops replicating and diverges; without this + * the only trace is a status column in + * spock.local_sync_status that nobody thinks to read. */ + if (sync->status != SYNC_STATUS_FAILED) + ereport(WARNING, + (errmsg("SPOCK %s: synchronization of table %s.%s failed, changes for it are no longer applied", + MySubscription->name, + NameStr(sync->nspname), + NameStr(sync->relname)), + /* + * The hint is meant to be pasted into psql, so both + * arguments have to survive names that need quoting: the + * relation is a regclass, which for a mixed-case or + * dotted name resolves to the wrong table (or nothing) + * unqualified, and an apostrophe in either name would + * truncate the literal. + */ + errhint("Re-synchronize with spock.sub_resync_table(%s, %s) once the cause is fixed.", + quote_literal_cstr(MySubscription->name), + quote_literal_cstr( + quote_qualified_identifier(NameStr(sync->nspname), + NameStr(sync->relname)))))); + sync->status = SYNC_STATUS_FAILED; sync->statuslsn = InvalidXLogRecPtr; } diff --git a/src/spock_sync.c b/src/spock_sync.c index a982c6f77..23fae4dc4 100644 --- a/src/spock_sync.c +++ b/src/spock_sync.c @@ -72,6 +72,16 @@ #define PGDUMP_BINARY "pg_dump" #define PGRESTORE_BINARY "pg_restore" +/* + * Staging table used by copy_table_data() when the target already holds rows. + * Lives in pg_temp on the target connection for the duration of the COPY. + */ +#define SPOCK_SYNC_STAGE_RELNAME "spock_sync_stage" + +/* Savepoint and bound for the empty-table lock in copy_table_data(). */ +#define SPOCK_SYNC_LOCK_SAVEPOINT "spock_sync_lock" +#define SPOCK_SYNC_LOCK_TIMEOUT_MS 5000 + #define Natts_local_sync_state 6 #define Anum_sync_kind 1 #define Anum_sync_subid 2 @@ -998,6 +1008,69 @@ make_copy_attnamelist(SpockRelation *rel) return attnamelist; } +/* + * Does the target table already hold rows? + * + * Chooses between the direct COPY and the staging path in copy_table_data(). + * Errors out rather than guessing, because getting this wrong in the "empty" + * direction is what wedges the table's sync status. + */ +static bool +target_table_has_rows(PGconn *target_conn, SpockRemoteRel *remoterel, + const char *relident) +{ + PGresult *res; + bool has_rows; + StringInfoData query; + + initStringInfo(&query); + appendStringInfo(&query, "SELECT 1 FROM %s LIMIT 1", relident); + res = PQexec(target_conn, query.data); + pfree(query.data); + + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + char *msg = pstrdup(PQerrorMessage(target_conn)); + + PQclear(res); + ereport(ERROR, + (errmsg("could not check whether target table %s.%s is empty", + remoterel->nspname, remoterel->relname), + errdetail("destination connection reported: %s", msg))); + } + has_rows = PQntuples(res) > 0; + PQclear(res); + + return has_rows; +} + +/* + * Run one command on the target connection during a table copy. + * + * A failed command is reported with ereport(ERROR), which does not return + * to the caller, so every caller can treat the returned PGresult as valid + * and is responsible for PQclear()ing it. + */ +static PGresult * +sync_target_cmd(PGconn *target_conn, const char *sql, + SpockRemoteRel *remoterel, const char *what) +{ + PGresult *res = PQexec(target_conn, sql); + + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + char *msg = pstrdup(PQerrorMessage(target_conn)); + + PQclear(res); + ereport(ERROR, + (errmsg("could not %s for %s.%s", what, + remoterel->nspname, remoterel->relname), + errdetail("destination connection reported: %s", msg))); + } + + return res; +} + /* * COPY single table over wire. */ @@ -1012,8 +1085,12 @@ copy_table_data(PGconn *origin_conn, PGconn *target_conn, List *attnamelist; ListCell *lc; bool first; + bool stage_load; + bool override_identity = false; + char *merged = NULL; StringInfoData query; StringInfoData attlist; + StringInfoData relident; MemoryContext curctx = CurrentMemoryContext, oldctx; @@ -1030,6 +1107,25 @@ copy_table_data(PGconn *origin_conn, PGconn *target_conn, attnamelist = make_copy_attnamelist(rel); + /* + * COPY may write GENERATED ALWAYS AS IDENTITY columns, an INSERT may not + * without OVERRIDING SYSTEM VALUE. Remember whether we need it for the + * staged-load path below. + */ + { + TupleDesc desc = RelationGetDescr(rel->rel); + int attnum; + + for (attnum = 0; attnum < desc->natts; attnum++) + { + if (TupleDescAttr(desc, attnum)->attidentity == ATTRIBUTE_IDENTITY_ALWAYS) + { + override_identity = true; + break; + } + } + } + initStringInfo(&attlist); first = true; foreach(lc, attnamelist) @@ -1129,13 +1225,166 @@ copy_table_data(PGconn *origin_conn, PGconn *target_conn, PQerrorMessage(origin_conn)))); } - /* Build COPY FROM query. */ - resetStringInfo(&query); - appendStringInfo(&query, "COPY %s.%s ", - PQescapeIdentifier(origin_conn, remoterel->nspname, + /* + * Decide whether to load straight into the table or through a staging + * table. + * + * A direct COPY into a table that already holds rows aborts on the first + * key collision, and that failure is not recoverable: the table's sync + * status ends up SYNC_STATUS_FAILED, and from then on the apply worker + * drops every change for it (see should_apply_changes_for_rel), silently + * and permanently. In a mesh this is the normal case rather than an edge + * case, because adding an already-populated table to a replication set + * with synchronize_data := true asks every peer to copy rows it already + * has. Load those tables into an unconstrained staging table and merge, + * so a sync of data we already hold converges instead of wedging. + */ + initStringInfo(&relident); + appendStringInfo(&relident, "%s.%s", + PQescapeIdentifier(target_conn, remoterel->nspname, strlen(remoterel->nspname)), - PQescapeIdentifier(origin_conn, remoterel->relname, + PQescapeIdentifier(target_conn, remoterel->relname, strlen(remoterel->relname))); + + /* + * Off by default: spock.sync_stage_and_merge has to be turned on before + * any of this happens. With it off the COPY goes straight into the + * table as it always has, the probe below is not even taken, and a + * populated target still fails the way it used to -- which is the + * behaviour an existing deployment is relying on. + */ + stage_load = spock_sync_stage_and_merge && + target_table_has_rows(target_conn, remoterel, relident.data); + + if (spock_sync_stage_and_merge && !stage_load) + { + /* + * That probe took no lock, so on its own it does not settle anything: + * another transaction can commit a row between it and the COPY, and + * the COPY then aborts on the duplicate key and wedges the table's + * sync status, which is the exact failure the staging path exists to + * avoid. Lock writers out and ask again; the second answer holds for + * the rest of the copy transaction. + * + * The lock is taken only on this path. Here the table is empty, so + * nothing should be contending for it, and blocking writes to a table + * that is mid initial load is what we want anyway. Locking before the + * first probe would instead hold EXCLUSIVE on a populated table for + * the whole sync, which on a live node is a real availability cost. + * + * The wait is bounded, because every table is copied in one + * transaction: this lock is held until the last table is done, and an + * apply worker holding ROW EXCLUSIVE on a table this sync has not + * reached yet would deadlock against it. A savepoint keeps the failure + * recoverable, since an error would otherwise abort the copy + * transaction. If the lock does not arrive, fall back to the staging + * path, which is correct whether or not the table is empty. + */ + PQclear(sync_target_cmd(target_conn, + "SAVEPOINT " SPOCK_SYNC_LOCK_SAVEPOINT, + remoterel, "open a savepoint")); + + resetStringInfo(&query); + appendStringInfo(&query, + "SET LOCAL lock_timeout = %d;" + "LOCK TABLE %s IN EXCLUSIVE MODE", + SPOCK_SYNC_LOCK_TIMEOUT_MS, relident.data); + res = PQexec(target_conn, query.data); + + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + char *sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE); + bool busy = sqlstate != NULL && + strcmp(sqlstate, "55P03" /*ERRCODE_LOCK_NOT_AVAILABLE*/) == 0; + char *msg = pstrdup(PQerrorMessage(target_conn)); + + PQclear(res); + + /* + * Rolling back leaves the savepoint live, so release it too. Every + * table in the sync shares this transaction, and one dangling + * subtransaction per table would push a large sync past the 64 the + * snapshot can track without overflowing. + */ + PQclear(sync_target_cmd(target_conn, + "ROLLBACK TO SAVEPOINT " SPOCK_SYNC_LOCK_SAVEPOINT, + remoterel, "roll back to the lock savepoint")); + PQclear(sync_target_cmd(target_conn, + "RELEASE SAVEPOINT " SPOCK_SYNC_LOCK_SAVEPOINT, + remoterel, "release the lock savepoint")); + + if (!busy) + ereport(ERROR, + (errmsg("could not lock target table %s.%s for synchronization", + remoterel->nspname, remoterel->relname), + errdetail("destination connection reported: %s", msg))); + + elog(LOG, "SPOCK: could not lock %s.%s within %dms, staging the copy instead", + remoterel->nspname, remoterel->relname, + SPOCK_SYNC_LOCK_TIMEOUT_MS); + stage_load = true; + } + else + { + PQclear(res); + + stage_load = target_table_has_rows(target_conn, remoterel, + relident.data); + + if (stage_load) + { + /* + * Rows landed between the first probe and the lock. Staging + * copes with that, and holding EXCLUSIVE on a populated table + * for the rest of the sync is the availability cost this path + * exists to avoid, so drop the lock again. Rolling back to the + * savepoint releases locks taken inside it, and undoes the + * lock_timeout with them. + */ + PQclear(sync_target_cmd(target_conn, + "ROLLBACK TO SAVEPOINT " SPOCK_SYNC_LOCK_SAVEPOINT, + remoterel, "roll back to the lock savepoint")); + } + else + { + /* Keep the lock, restore what start_copy_target_tx() set. */ + PQclear(sync_target_cmd(target_conn, "SET LOCAL lock_timeout = 0", + remoterel, "reset lock_timeout")); + } + + PQclear(sync_target_cmd(target_conn, + "RELEASE SAVEPOINT " SPOCK_SYNC_LOCK_SAVEPOINT, + remoterel, "release the lock savepoint")); + } + } + + if (stage_load) + { + /* + * A bare LIKE copies NOT NULL but not the defaults, generation + * expressions or identity behind it, and make_copy_attnamelist() + * leaves generated and provider-absent columns out of the COPY. The + * staging table would then take a NULL where the real table computes + * a value, and the COPY would abort on the not-null constraint. + */ + resetStringInfo(&query); + appendStringInfo(&query, + "DROP TABLE IF EXISTS pg_temp.%s;" + "CREATE TEMP TABLE %s (LIKE %s" + " INCLUDING DEFAULTS INCLUDING GENERATED" + " INCLUDING IDENTITY)", + SPOCK_SYNC_STAGE_RELNAME, SPOCK_SYNC_STAGE_RELNAME, + relident.data); + PQclear(sync_target_cmd(target_conn, query.data, remoterel, + "create the staging table")); + } + + /* Build COPY FROM query. */ + resetStringInfo(&query); + if (stage_load) + appendStringInfo(&query, "COPY pg_temp.%s ", SPOCK_SYNC_STAGE_RELNAME); + else + appendStringInfo(&query, "COPY %s ", relident.data); if (list_length(attnamelist)) appendStringInfo(&query, "(%s) ", attlist.data); appendStringInfoString(&query, "FROM stdin"); @@ -1201,8 +1450,66 @@ copy_table_data(PGconn *origin_conn, PGconn *target_conn, } PQclear(res); - elog(INFO, "finished synchronization of data for table %s.%s", - remoterel->nspname, remoterel->relname); + /* + * Merge the staged rows. Rows we already have are left alone rather than + * overwritten: the local copy is the one the rest of the cluster has + * already replicated from us, so keeping it is the conservative choice. + */ + if (stage_load) + { + const char *mergelist = attlist.data; + + /* + * The copy had no column list, so it moved every non-generated column. + * Name them for the merge rather than using SELECT *, which would hand + * the target a generated column and be rejected. + */ + if (!list_length(attnamelist)) + { + res = sync_target_cmd(target_conn, + "SELECT string_agg(quote_ident(attname), ', ' ORDER BY attnum)" + " FROM pg_attribute" + " WHERE attrelid = 'pg_temp." SPOCK_SYNC_STAGE_RELNAME "'::regclass" + " AND attnum > 0 AND NOT attisdropped AND attgenerated = ''", + remoterel, "list the staging table columns"); + + if (PQntuples(res) != 1 || PQgetisnull(res, 0, 0)) + { + PQclear(res); + ereport(ERROR, + (errmsg("staging table for %s.%s has no columns to merge", + remoterel->nspname, remoterel->relname))); + } + mergelist = pstrdup(PQgetvalue(res, 0, 0)); + PQclear(res); + } + + resetStringInfo(&query); + appendStringInfo(&query, + "INSERT INTO %s (%s) %sSELECT %s FROM pg_temp.%s " + "ON CONFLICT DO NOTHING", + relident.data, mergelist, + override_identity ? "OVERRIDING SYSTEM VALUE " : "", + mergelist, SPOCK_SYNC_STAGE_RELNAME); + + res = sync_target_cmd(target_conn, query.data, remoterel, + "merge the staged rows"); + merged = pstrdup(PQcmdTuples(res)); + PQclear(res); + + resetStringInfo(&query); + appendStringInfo(&query, "DROP TABLE pg_temp.%s", + SPOCK_SYNC_STAGE_RELNAME); + PQclear(sync_target_cmd(target_conn, query.data, remoterel, + "drop the staging table")); + } + + if (stage_load) + elog(INFO, "finished synchronization of data for table %s.%s, %s row(s) added to existing data", + remoterel->nspname, remoterel->relname, merged); + else + elog(INFO, "finished synchronization of data for table %s.%s", + remoterel->nspname, remoterel->relname); } /* diff --git a/tests/tap/t/014_pgdump_restore_conflict.pl b/tests/tap/t/014_pgdump_restore_conflict.pl index 7c11a6daf..1b6111ecb 100755 --- a/tests/tap/t/014_pgdump_restore_conflict.pl +++ b/tests/tap/t/014_pgdump_restore_conflict.pl @@ -1,6 +1,6 @@ use strict; use warnings; -use Test::More tests => 20; +use Test::More; use lib '.'; use SpockTest qw(create_cluster destroy_cluster system_or_bail command_ok get_test_config scalar_query psql_or_bail); @@ -341,3 +341,5 @@ sub wait_for_value { unlink $dump_file if -e $dump_file; destroy_cluster('Cleanup pg_dump/restore conflict test cluster'); + +done_testing();