/*------------------------------------------------------------------------- * * spock_sync.c * table synchronization functions * * Copyright (c) 2022-2026, pgEdge, Inc. * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group * Portions Copyright (c) 1994, The Regents of the University of California * *------------------------------------------------------------------------- */ #include "postgres.h" #include #include #include "libpq-fe.h" #include "miscadmin.h" #include "access/genam.h" #include "access/hash.h" #include "access/heapam.h" #include "access/skey.h" #include "access/stratnum.h" #include "access/xact.h" #include "catalog/indexing.h" #include "catalog/namespace.h" #include "commands/dbcommands.h" #include "commands/extension.h" #include "commands/tablecmds.h" #include "lib/stringinfo.h" #include "utils/memutils.h" #include "nodes/makefuncs.h" #include "nodes/parsenodes.h" #include "pgstat.h" #include "replication/origin.h" #include "storage/fd.h" #include "storage/ipc.h" #include "storage/proc.h" #include "tcop/utility.h" #include "utils/builtins.h" #include "utils/fmgroids.h" #include "utils/guc.h" #include "utils/pg_lsn.h" #include "utils/rel.h" #include "utils/resowner.h" #include "spock_common.h" #include "spock_group.h" #include "spock_exception_handler.h" #include "spock_relcache.h" #include "spock_repset.h" #include "spock_rpc.h" #include "spock_sync.h" #include "spock_monitor.h" #include "spock_worker.h" #include "spock.h" #define CATALOG_LOCAL_SYNC_STATUS "local_sync_status" #define PGDUMP_BINARY "pg_dump" #define PGRESTORE_BINARY "pg_restore" #define Natts_local_sync_state 6 #define Anum_sync_kind 1 #define Anum_sync_subid 2 #define Anum_sync_nspname 3 #define Anum_sync_relname 4 #define Anum_sync_status 5 #define Anum_sync_statuslsn 6 /* * Columns (PP_ = peer progress) of the queries that read a provider's * replication progress, in read_provider_progress() and * spock_create_slot_and_read_progress(). Every form of those queries, for * 6.0.0 and pre-6.0.0 providers alike, selects exactly these, in this order, * and peer_progress_from_row() turns one such row into a SpockApplyProgress. */ enum PeerProgressCol { PP_REMOTE_NODE_ID = 0, PP_REMOTE_COMMIT_TS, PP_REMOTE_COMMIT_LSN, PP_REMOTE_INSERT_LSN, PP_LAST_UPDATED_TS, PP_UPDATED_BY_DECODE }; #define PEER_PROGRESS_COLUMNS \ "remote_node_id, remote_commit_ts, %s AS remote_commit_lsn, " \ "remote_insert_lsn, last_updated_ts, updated_by_decode" PGDLLEXPORT void spock_sync_main(Datum main_arg); static SpockSyncWorker *MySyncWorker = NULL; /* Set when the initial copy of this worker's table failed. */ static bool sync_worker_failed = false; /* * Run a command and wait for it to exit, then return its exit code * in the same format as waitpid(). * * Does not elog(ERROR). * * 'cmd' must be a full relative or absolute path to the executable to * start. The PATH is not searched. * * Preserves each argument in cmdargv as a discrete argument to the child * process. The first entry in cmdargv is passed as the child process's * argv[0], so the first "real" argument begins at index 1. * * Uses the current environment and working directory. * * Note that if we elog(ERROR) or elog(FATAL) here we won't kill the * child proc. * * The return value is a waitpid() status, not an errno: a child that ran and * exited non-zero leaves errno untouched, so report failures with * wait_result_to_str() rather than %m. Only the -1 return, where waitpid() * itself failed, is an errno case, and wait_result_to_str() renders that one * as %m for us. */ static int exec_cmd(const char *cmd, char *cmdargv[]) { pid_t pid; int stat; /* Fire off execv in child */ fflush(stdout); fflush(stderr); if ((pid = fork()) == 0) { if (execv(cmd, cmdargv) < 0) { ereport(ERROR, (errmsg("could not execute \"%s\": %m", cmd))); /* We're already in the child process here, can't return */ exit(1); } } if (waitpid(pid, &stat, 0) != pid) stat = -1; return stat; } static void get_pg_executable(char *cmdname, char *cmdbuf) { uint32 version; if (find_other_exec_version(my_exec_path, cmdname, &version, cmdbuf)) elog(ERROR, "spock subscriber init failed to find %s relative to binary %s", cmdname, my_exec_path); if (version / 100 != PG_VERSION_NUM / 100) elog(ERROR, "spock subscriber init found %s with wrong major version %d.%d, expected %d.%d", cmdname, version / 100 / 100, version / 100 % 100, PG_VERSION_NUM / 100 / 100, PG_VERSION_NUM / 100 % 100); } /* * Couple of helper functions to centralise the extension and schema skipping * logic. Test existence of the object before adding it to the list to avoid * noisy WARNINGs. * The sub->skip_schema typically doesn't include our globally-skipped objects. * So, don't care about duplicates. */ #if PG_VERSION_NUM >= 170000 static List * build_exclude_extension_string(void) { List *lst = NIL; char *arg; List *reserved; ListCell *lc; reserved = spock_reserved_object_names(RESERVED_KIND_EXTENSION, RESERVED_PURPOSE_DUMP); foreach(lc, reserved) { const char *ext_name = (const char *) lfirst(lc); /* Unconditional: see build_exclude_schema_string(). */ arg = psprintf("--exclude-extension=%s", ext_name); lst = lappend(lst, arg); } list_free_deep(reserved); return lst; } #endif static List * build_exclude_schema_string(SpockSubscription *sub) { List *lst = NIL; char *arg; List *reserved; ListCell *lc; reserved = spock_reserved_object_names(RESERVED_KIND_SCHEMA, RESERVED_PURPOSE_DUMP); foreach(lc, reserved) { const char *schema_name = (const char *) lfirst(lc); /* * Exclude unconditionally: we run on the subscriber but pg_dump reads * the provider (e.g. pgedge_ace exists only there on a fresh join), * and pg_dump ignores an exclude pattern that matches nothing. */ arg = psprintf("--exclude-schema=%s", schema_name); lst = lappend(lst, arg); } list_free_deep(reserved); if (sub) { foreach(lc, sub->skip_schema) { const char *schema_name = (const char *) lfirst(lc); if (!OidIsValid(LookupExplicitNamespace(schema_name, true))) continue; arg = psprintf("--exclude-schema=%s", schema_name); lst = lappend(lst, arg); } } return lst; } static void dump_structure(SpockSubscription *sub, const char *destfile, const char *snapshot) { char *dsn; char *err_msg; char pg_dump[MAXPGPATH]; char **cmdargv; int cmdargc = 0; int stat; List *args = NIL; char *arg; ListCell *lc; dsn = spk_get_connstr((char *) sub->origin_if->dsn, NULL, NULL, &err_msg); if (dsn == NULL) elog(ERROR, "invalid connection string \"%s\": %s", sub->origin_if->dsn, err_msg); get_pg_executable(PGDUMP_BINARY, pg_dump); args = lappend(args, pg_dump); args = lappend(args, "-Fc"); /* custom format */ args = lappend(args, "-s"); /* schema only */ arg = psprintf("--snapshot=%s", snapshot); args = lappend(args, arg); /* Filter out schemas and extensions, skipped globally. */ args = list_concat(args, build_exclude_schema_string(sub)); #if PG_VERSION_NUM >= 170000 args = list_concat(args, build_exclude_extension_string()); #endif /* destination file */ arg = psprintf("--file=%s", destfile); args = lappend(args, arg); /* connection string */ arg = psprintf("--dbname=%s", dsn); args = lappend(args, arg); free(dsn); /* build args' array for the execv call */ cmdargv = palloc((list_length(args) + 1) * sizeof(char *)); foreach(lc, args) { arg = (char *) lfirst(lc); cmdargv[cmdargc++] = arg; } cmdargv[cmdargc++] = NULL; stat = exec_cmd(pg_dump, cmdargv); if (stat != 0) ereport(ERROR, (errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION), errmsg("could not execute pg_dump (\"%s\"): %s", pg_dump, wait_result_to_str(stat)))); /* * Allocations have been made in the transaction context. Hence, don't * bother freeing memory - it will be released soon. Also, some elements * in the args list are string literals, so freeing the list would be * unsafe anyway. */ } static void restore_structure(SpockSubscription *sub, const char *srcfile, const char *section) { char *dsn; char *err_msg; char pg_restore[MAXPGPATH]; char *cmdargv[20]; int cmdargc = 0; int stat; StringInfoData s; dsn = spk_get_connstr((char *) sub->target_if->dsn, NULL, "-cspock.subscription_schema_restore=true", &err_msg); if (dsn == NULL) elog(ERROR, "invalid connection string \"%s\": %s", sub->target_if->dsn, err_msg); get_pg_executable(PGRESTORE_BINARY, pg_restore); cmdargv[cmdargc++] = pg_restore; /* section */ if (section) { initStringInfo(&s); appendStringInfo(&s, "--section=%s", section); cmdargv[cmdargc++] = pstrdup(s.data); resetStringInfo(&s); } /* stop execution on any error */ cmdargv[cmdargc++] = "--exit-on-error"; /* apply everything in single tx */ cmdargv[cmdargc++] = "-1"; /* connection string */ initStringInfo(&s); appendStringInfo(&s, "--dbname=%s", dsn); cmdargv[cmdargc++] = pstrdup(s.data); free(dsn); /* source file */ cmdargv[cmdargc++] = pstrdup(srcfile); cmdargv[cmdargc++] = NULL; stat = exec_cmd(pg_restore, cmdargv); if (stat != 0) ereport(ERROR, (errcode(ERRCODE_EXTERNAL_ROUTINE_EXCEPTION), errmsg("could not execute pg_restore (\"%s\"): %s", pg_restore, wait_result_to_str(stat)), errhint("check the error log to determine the cause of the issue"))); } /* * upstream_version_supports_structure_sync * True if our pg_dump/pg_restore can read this upstream. * * The subscriber's pg_dump (its own, per get_pg_executable()) cannot read a * provider whose server version falls outside the inclusive * [minRemoteVersion, maxRemoteVersion] window that pg_dump itself enforces in * _check_database_version() (src/bin/pg_dump/pg_backup_db.c). The bounds * mirror those set in main() of src/bin/pg_dump/pg_dump.c. */ bool upstream_version_supports_structure_sync(PGconn *origin_conn) { int remoteversion; const int minRemoteVersion = 90200; const int maxRemoteVersion = (PG_VERSION_NUM / 100) * 100 + 99; Assert(origin_conn != NULL && PQstatus(origin_conn) == CONNECTION_OK); remoteversion = PQserverVersion(origin_conn); /* * Treat an unknown version (0 -- a dead connection or a non-PostgreSQL * endpoint) as unsupported, and reject upstreams below minRemoteVersion * or above maxRemoteVersion, matching pg_dump's own bounds check. */ return remoteversion != 0 && remoteversion >= minRemoteVersion && remoteversion <= maxRemoteVersion; } /* * Create slot and get the exported snapshot. * * This will try to recreate slot if already exists and not active. * * The reported LSN is the confirmed flush LSN at the point the slot reached * consistency and exported its snapshot. */ static char * ensure_replication_slot_snapshot(PGconn *sql_conn, PGconn *repl_conn, char *slot_name, bool use_failover_slot, XLogRecPtr *lsn) { PGresult *res; StringInfoData query; char *snapshot; retry: initStringInfo(&query); appendStringInfo(&query, "CREATE_REPLICATION_SLOT \"%s\" LOGICAL %s", slot_name, "spock_output"); /*---------- * Mark the slot with (FAILOVER) when the *remote* provider is PG17+. * PG17+ supports logical slot synchronization to physical standbys via * sync_replication_slots = on. PG17+ uses parenthesised option syntax: * CREATE_REPLICATION_SLOT "name" LOGICAL plugin (FAILOVER) * * Detect the version on the SQL connection (sql_conn), which we hold open * anyway for the slot-reclaim queries below. */ if (PQserverVersion(sql_conn) >= 170000) appendStringInfo(&query, " (FAILOVER)"); #if PG_VERSION_NUM < 170000 else if (use_failover_slot) appendStringInfo(&query, " (FAILOVER)"); #endif res = PQexec(repl_conn, query.data); if (PQresultStatus(res) != PGRES_TUPLES_OK) { const char *sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE); /* * If our slot already exists but is not used, it's leftover from * previous unsuccessful attempt to synchronize table, try dropping it * and recreating. */ if (sqlstate && strcmp(sqlstate, "42710" /* ERRCODE_DUPLICATE_OBJECT */ ) == 0 && !spock_remote_slot_active(sql_conn, slot_name)) { pfree(query.data); PQclear(res); spock_drop_remote_slot(sql_conn, slot_name); goto retry; } elog(ERROR, "could not create replication slot on provider: %s\n" "query: %s", PQresultErrorMessage(res), query.data); } *lsn = DatumGetLSN(DirectFunctionCall1Coll(pg_lsn_in, InvalidOid, CStringGetDatum(PQgetvalue(res, 0, 1)))); snapshot = pstrdup(PQgetvalue(res, 0, 2)); PQclear(res); return snapshot; } /* * Get or create replication origin for a given slot. */ static RepOriginId ensure_replication_origin(char *slot_name) { RepOriginId origin = replorigin_by_name(slot_name, true); if (origin == InvalidRepOriginId) origin = replorigin_create(slot_name); return origin; } /* * Does the provider's spock.progress carry the 6.0.0 column names? Before * 6.0.0, and on a 6.0 binary whose extension has not been updated yet, it is * a plain table whose commit-LSN column is called remote_lsn. Ask the * catalog, not the library version. */ static bool provider_progress_has_commit_lsn(PGconn *conn) { PGresult *res; bool found; res = PQexec(conn, "SELECT EXISTS (SELECT 1 FROM pg_attribute " "WHERE attrelid = 'spock.progress'::regclass " " AND attname = 'remote_commit_lsn' AND NOT attisdropped)"); if (PQresultStatus(res) != PGRES_TUPLES_OK) elog(ERROR, "could not inspect spock.progress on origin: %s", PQerrorMessage(conn)); found = (PQgetvalue(res, 0, 0)[0] == 't'); PQclear(res); return found; } /* * Build a SpockApplyProgress for this node from row rno of a result laid * out per enum PeerProgressCol. The caller owns the allocation, made in * CacheMemoryContext. */ static SpockApplyProgress * peer_progress_from_row(PGresult *res, int rno) { SpockApplyProgress *sap; char *remote_node_id = PQgetvalue(res, rno, PP_REMOTE_NODE_ID); char *remote_commit_ts = NULL; char *last_updated_ts = NULL; sap = MemoryContextAlloc(CacheMemoryContext, sizeof(SpockApplyProgress)); sap->key.dbid = MyDatabaseId; sap->key.node_id = MySubscription->target->id; sap->key.remote_node_id = atooid(remote_node_id); Assert(OidIsValid(sap->key.remote_node_id)); /* Alloc above doesn't zero; set every non-key field to "unset" */ spock_init_progress_fields(sap); if (!PQgetisnull(res, rno, PP_REMOTE_COMMIT_TS)) { remote_commit_ts = PQgetvalue(res, rno, PP_REMOTE_COMMIT_TS); sap->remote_commit_ts = str_to_timestamptz(remote_commit_ts); Assert(IS_VALID_TIMESTAMP(sap->remote_commit_ts)); } sap->prev_remote_ts = sap->remote_commit_ts; sap->remote_commit_lsn = str_to_lsn(PQgetvalue(res, rno, PP_REMOTE_COMMIT_LSN)); sap->remote_insert_lsn = str_to_lsn(PQgetvalue(res, rno, PP_REMOTE_INSERT_LSN)); /* * We don't actually receive a single WAL record - just assume we've got * the last commit only. Don't set it to the Invalid value in case someone * uses tracking data in state monitoring scripts. */ sap->received_lsn = sap->remote_commit_lsn; if (!PQgetisnull(res, rno, PP_LAST_UPDATED_TS)) { last_updated_ts = PQgetvalue(res, rno, PP_LAST_UPDATED_TS); sap->last_updated_ts = str_to_timestamptz(last_updated_ts); Assert(IS_VALID_TIMESTAMP(sap->last_updated_ts)); if (sap->last_updated_ts < sap->remote_commit_ts) /* * Complaining at the end of the sync we shouldn't flood the log */ ereport(WARNING, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("transaction apply time precedes its original commit time"), errdetail("Commit time %s (node ID %s), but it was applied at %s on the replica (node ID %d)", remote_commit_ts, remote_node_id, last_updated_ts, MySubscription->origin->id), errhint("usually it means that the server's clocks are out of sync."))); } sap->updated_by_decode = (PQgetvalue(res, rno, PP_UPDATED_BY_DECODE)[0] == 't'); return sap; } static List * read_provider_progress(PGconn *origin_conn) { StringInfoData query; PGresult *originRes; List *resultList = NIL; /* * Select the current content of the origin's spock.progress where the * origin is the target and this node is not the origin. In 6.0.0 that is * a view already limited to the current database; before 6.0.0 it is a * per-database table with the commit LSN under its old name. * * We use this information to update the target's spock.progress table so * that START REPLICATION can ask the walsender to skip all transactions * that have already been applied from other existing nodes and are * therefore part of the snapshot we are copying. */ initStringInfo(&query); appendStringInfo(&query, "SELECT " PEER_PROGRESS_COLUMNS " FROM spock.progress " "WHERE node_id = %u AND remote_node_id <> %u", provider_progress_has_commit_lsn(origin_conn) ? "remote_commit_lsn" : "remote_lsn", MySubscription->origin->id, MySubscription->target->id); originRes = PQexec(origin_conn, query.data); if (PQresultStatus(originRes) == PGRES_TUPLES_OK) { int rno; for (rno = 0; rno < PQntuples(originRes); rno++) { SpockApplyProgress *sap = peer_progress_from_row(originRes, rno); MemoryContext oldctx; oldctx = MemoryContextSwitchTo(CacheMemoryContext); resultList = lappend(resultList, sap); MemoryContextSwitchTo(oldctx); elog(LOG, "SPOCK: read provider progress %s->%d: " "remote_commit_ts='%s' " "remote_commit_lsn='%s' " "remote_insert_lsn='%s'", PQgetvalue(originRes, rno, PP_REMOTE_NODE_ID), MySubscription->target->id, PQgetvalue(originRes, rno, PP_REMOTE_COMMIT_TS), PQgetvalue(originRes, rno, PP_REMOTE_COMMIT_LSN), PQgetvalue(originRes, rno, PP_REMOTE_INSERT_LSN)); } } else { elog(WARNING, "SPOCK: Cannot adjust spock.progress - %s", PQresultErrorMessage(originRes)); } PQclear(originRes); pfree(query.data); return resultList; } /* * spock_create_slot_and_read_progress * * Pauses apply workers, creates a replication slot via the replication * protocol (which returns a snapshot consistent with the slot's WAL * position), reads peer progress while workers are paused, and resumes. * * Leaves conn in an open REPEATABLE READ transaction with the slot's * snapshot imported; caller must call spock_release_slot_snapshot() * after the COPY phase completes. */ static char * spock_create_slot_and_read_progress(PGconn *conn, PGconn *repl_conn, const char *slot_name, Oid origin_node_id, Oid subscriber_node_id, XLogRecPtr *lsn_out, List **progress_out) { StringInfoData query; PGresult *res; char *snapshot = NULL; List *progress_list = NIL; int nrows; int rno; bool have_read_peer_progress = false; bool can_pause_apply_workers; initStringInfo(&query); /* * Drop an existing inactive slot so we can re-create it cleanly. Ignore * errors (the slot may not exist, which is fine). */ appendStringInfo(&query, "SELECT pg_drop_replication_slot('%s') " "WHERE EXISTS (" " SELECT 1 FROM pg_replication_slots " " WHERE slot_name = '%s' AND NOT active)", slot_name, slot_name); res = PQexec(conn, query.data); PQclear(res); resetStringInfo(&query); elog(LOG, "SPOCK cswp slot=%s provider=%u subscriber=%u", slot_name, origin_node_id, subscriber_node_id); /* * Does the provider have spock.read_peer_progress()? It arrives with the * 6.0.0 extension version, so a provider on an older extension, or on the * 6.0 binary with ALTER EXTENSION spock UPDATE still pending, lacks it * and keeps its progress in the pre-6.0.0 spock.progress table. Ask the * catalog rather than the library version, and ask before pausing apply * workers so a failure here cannot leave them paused. */ res = PQexec(conn, "SELECT to_regprocedure('spock.read_peer_progress(text,oid,oid)') IS NOT NULL"); if (PQresultStatus(res) != PGRES_TUPLES_OK) elog(ERROR, "could not check for spock.read_peer_progress() on origin: %s", PQerrorMessage(conn)); have_read_peer_progress = (PQgetvalue(res, 0, 0)[0] == 't'); PQclear(res); elog(LOG, "SPOCK cswp provider has read_peer_progress: %s", have_read_peer_progress ? "yes" : "no"); /* * Pause apply workers so ros.remote_lsn reflects only fully committed * state when we read it below. * * pause_apply_workers() was added in 5.0.7; check for it first so an * older origin (a supported rolling-upgrade pairing) gets one clear log * line instead of a raw "does not exist" error. Sync proceeds unpaused * if it's missing, reopening the race above. */ can_pause_apply_workers = spock_remote_function_exists(conn, "spock", "pause_apply_workers", 0, NULL); if (!can_pause_apply_workers) elog(LOG, "SPOCK: origin does not support spock.pause_apply_workers() " "(Spock older than 5.0.7); proceeding without it. A " "concurrent write load on the origin during this sync " "could leave the subscriber missing rows; upgrade the " "origin to close this window."); else { res = PQexec(conn, "SELECT spock.pause_apply_workers()"); if (PQresultStatus(res) != PGRES_TUPLES_OK) elog(ERROR, "could not pause apply workers on origin: %s", PQerrorMessage(conn)); PQclear(res); } /* * Create the slot via the replication protocol. This returns a snapshot * consistent with the slot's WAL position — the correct snapshot for * COPY. * * Mark the slot with (FAILOVER) when the remote provider is PG17+. Detect * the version on the SQL connection (conn), held open anyway. */ appendStringInfo(&query, "CREATE_REPLICATION_SLOT \"%s\" LOGICAL %s", slot_name, "spock_output"); if (PQserverVersion(conn) >= 170000) appendStringInfo(&query, " (FAILOVER)"); res = PQexec(repl_conn, query.data); resetStringInfo(&query); if (PQresultStatus(res) != PGRES_TUPLES_OK) { const char *sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE); if (sqlstate && strcmp(sqlstate, "42710") == 0 && !spock_remote_slot_active(conn, slot_name)) { PQclear(res); spock_drop_remote_slot(conn, slot_name); appendStringInfo(&query, "CREATE_REPLICATION_SLOT \"%s\" LOGICAL %s", slot_name, "spock_output"); if (PQserverVersion(conn) >= 170000) appendStringInfo(&query, " (FAILOVER)"); res = PQexec(repl_conn, query.data); resetStringInfo(&query); } if (PQresultStatus(res) != PGRES_TUPLES_OK) elog(ERROR, "could not create replication slot on provider: %s", PQresultErrorMessage(res)); } *lsn_out = DatumGetLSN(DirectFunctionCall1Coll(pg_lsn_in, InvalidOid, CStringGetDatum(PQgetvalue(res, 0, 1)))); snapshot = pstrdup(PQgetvalue(res, 0, 2)); PQclear(res); elog(LOG, "SPOCK cswp slot=%s lsn=%X/%X snapshot=%s", slot_name, LSN_FORMAT_ARGS(*lsn_out), snapshot); /* * Import the slot's snapshot into a REPEATABLE READ transaction. This is * the snapshot the COPY will use. */ appendStringInfo(&query, "BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ; " "SET TRANSACTION SNAPSHOT '%s'", snapshot); res = PQexec(conn, query.data); if (PQresultStatus(res) != PGRES_COMMAND_OK) elog(ERROR, "could not import snapshot on origin: %s", PQerrorMessage(conn)); PQclear(res); resetStringInfo(&query); /* * Read peer progress: one row per peer subscription on the provider, with * the PP_ columns declared above. The resume LSN (remote_commit_lsn) is * the peer origin's remote_lsn from pg_replication_origin_status, exact * because apply workers are paused. * * Without spock.read_peer_progress() (see the probe above) the provider * keeps progress in a plain table, so read the same six values from its * catalog directly. Only the resume LSN is exact on that path: the table * is flushed at most once a second, so remote_commit_ts, * remote_insert_lsn, last_updated_ts and updated_by_decode can trail the * paused state by up to that. They seed the new node's lag tracking and * are replaced by the first replicated message, so that is acceptable; * remote_insert_lsn is clamped to the resume LSN so it never reads as * behind a commit we know has been applied. */ if (!have_read_peer_progress) { appendStringInfo(&query, "SELECT p.remote_node_id, p.remote_commit_ts, " " COALESCE(ros.remote_lsn, '0/0'::pg_lsn) AS remote_commit_lsn, " " GREATEST(p.remote_insert_lsn, COALESCE(ros.remote_lsn, '0/0'::pg_lsn)) AS remote_insert_lsn, " " p.last_updated_ts, p.updated_by_decode " "FROM spock.subscription sub " "JOIN spock.progress p ON p.remote_node_id = sub.sub_origin " " AND p.node_id = sub.sub_target " "JOIN pg_replication_origin o ON o.roname = sub.sub_slot_name " "LEFT JOIN pg_replication_origin_status ros ON ros.local_id = o.roident " "WHERE sub.sub_target = %u AND sub.sub_origin <> %u", origin_node_id, subscriber_node_id); } else { /* * read_peer_progress() also returns a header row (remote_node_id * NULL) and lsn/snapshot columns; neither is needed here because the * slot's LSN and snapshot were taken from CREATE_REPLICATION_SLOT. */ appendStringInfo(&query, "SELECT " PEER_PROGRESS_COLUMNS " " "FROM spock.read_peer_progress('%s', %u, %u) " "WHERE remote_node_id IS NOT NULL", "remote_commit_lsn", slot_name, origin_node_id, subscriber_node_id); } res = PQexec(conn, query.data); resetStringInfo(&query); if (PQresultStatus(res) != PGRES_TUPLES_OK) elog(ERROR, "could not create replication slot and get progress on " "origin: %s", PQerrorMessage(conn)); /* Zero rows is legitimate: the provider may have no other peers. */ nrows = PQntuples(res); elog(LOG, "SPOCK cswp slot=%s lsn=%X/%X snapshot=%s peers=%d", slot_name, LSN_FORMAT_ARGS(*lsn_out), snapshot, nrows); for (rno = 0; rno < nrows; rno++) { SpockApplyProgress *sap; MemoryContext oldctx; if (PQgetisnull(res, rno, PP_REMOTE_NODE_ID)) continue; /* shouldn't happen but be safe */ sap = peer_progress_from_row(res, rno); oldctx = MemoryContextSwitchTo(CacheMemoryContext); progress_list = lappend(progress_list, sap); MemoryContextSwitchTo(oldctx); elog(LOG, "SPOCK cswp peer=%s->%d commit_lsn=%s insert_lsn=%s", PQgetvalue(res, rno, PP_REMOTE_NODE_ID), MySubscription->target->id, PQgetvalue(res, rno, PP_REMOTE_COMMIT_LSN), PQgetvalue(res, rno, PP_REMOTE_INSERT_LSN)); } PQclear(res); /* * Resume apply workers now that slot and progress are captured. The * REPEATABLE READ transaction (and its snapshot) remain open for the COPY * phase; the workers can resume safely. Nothing was paused on an origin * that lacks the function, so there is nothing to resume there either. */ if (can_pause_apply_workers) { res = PQexec(conn, "SELECT spock.resume_apply_workers()"); if (PQresultStatus(res) != PGRES_TUPLES_OK) elog(WARNING, "could not resume apply workers on origin: %s", PQerrorMessage(conn)); PQclear(res); } pfree(query.data); *progress_out = progress_list; return snapshot; } /* * spock_release_slot_snapshot * * Rolls back the snapshot transaction opened by * spock_create_slot_and_read_progress and closes the connection. */ static void spock_release_slot_snapshot(PGconn *conn) { PGresult *res = PQexec(conn, "ROLLBACK"); PQclear(res); PQfinish(conn); } /* * Transaction management for COPY. */ static void start_copy_origin_tx(PGconn *conn, const char *snapshot) { PGresult *res; const char *session_setup = "SET DATESTYLE = ISO;\n" "SET INTERVALSTYLE = POSTGRES;\n" "SET extra_float_digits TO 3;\n" "SET statement_timeout = 0;\n" "SET lock_timeout = 0;\n"; StringInfoData query; /* * Set the session GUCs before BEGIN, in a command of their own. An * extension's ProcessUtility hook may run a query of its own for a plain * SET (pg_duckdb initializes its metadata cache that way), and one query * inside the transaction makes the import fail with "SET TRANSACTION * SNAPSHOT must be called before any query". pg_dump orders it the same * way, which is why structure sync survives such extensions. */ res = PQexec(conn, session_setup); if (PQresultStatus(res) != PGRES_COMMAND_OK) spock_raise_pqresult_error(res, "session setup on origin node failed"); PQclear(res); initStringInfo(&query); appendStringInfoString(&query, "BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ, READ ONLY;\n"); if (snapshot) { appendStringInfoString(&query, "SET TRANSACTION SNAPSHOT "); append_escaped_literal(&query, conn, snapshot); appendStringInfoString(&query, ";\n"); } res = PQexec(conn, query.data); if (PQresultStatus(res) != PGRES_COMMAND_OK) spock_raise_pqresult_error(res, "BEGIN on origin node failed"); PQclear(res); pfree(query.data); } static void start_copy_target_tx(PGconn *conn, const char *origin_name) { PGresult *res; const char *setup_query = "BEGIN TRANSACTION ISOLATION LEVEL READ COMMITTED;\n" "SET session_replication_role = 'replica';\n" "SET DATESTYLE = ISO;\n" "SET INTERVALSTYLE = POSTGRES;\n" "SET extra_float_digits TO 3;\n" "SET statement_timeout = 0;\n" "SET lock_timeout = 0;\n"; StringInfoData query; initStringInfo(&query); /* * Set correct origin if target db supports it. We must do this before * starting the transaction otherwise the status code below would get much * more complicated. */ if (PQserverVersion(conn) >= 90500) { appendStringInfoString(&query, "SELECT pg_catalog.pg_replication_origin_session_setup("); append_escaped_literal(&query, conn, origin_name); appendStringInfoString(&query, ");\n"); appendStringInfo(&query, "SELECT pg_catalog.pg_replication_origin_xact_setup('0/0', 'epoch');"); } appendStringInfoString(&query, setup_query); res = PQexec(conn, query.data); if (PQresultStatus(res) != PGRES_COMMAND_OK) spock_raise_pqresult_error(res, "BEGIN on target node failed"); PQclear(res); pfree(query.data); } static void finish_copy_origin_tx(PGconn *conn) { PGresult *res; /* Close the transaction and connection on origin node. */ res = PQexec(conn, "ROLLBACK"); if (PQresultStatus(res) != PGRES_COMMAND_OK) elog(WARNING, "ROLLBACK on origin node failed: %s", PQresultErrorMessage(res)); PQclear(res); PQfinish(conn); } static void finish_copy_target_tx(PGconn *conn) { PGresult *res; /* Close the transaction and connection on target node. */ res = PQexec(conn, "COMMIT"); if (PQresultStatus(res) != PGRES_COMMAND_OK) elog(ERROR, "COMMIT on target node failed: %s", PQresultErrorMessage(res)); PQclear(res); /* * Resetting the origin explicitly before the backend exits will help * prevent races with other accesses to the same replication origin. */ if (PQserverVersion(conn) >= 90500) { res = PQexec(conn, "SELECT pg_catalog.pg_replication_origin_xact_reset();\n" "SELECT pg_catalog.pg_replication_origin_session_reset();\n"); if (PQresultStatus(res) != PGRES_TUPLES_OK) elog(WARNING, "Resetting session origin on target node failed: %s", PQresultErrorMessage(res)); PQclear(res); } PQfinish(conn); } static int physatt_in_attmap(SpockRelation *rel, int attid) { AttrNumber i; for (i = 0; i < rel->natts; i++) if (rel->attmap[i] == attid) return i; return -1; } /* * Create list of columns for COPY based on logical relation mapping. */ static List * make_copy_attnamelist(SpockRelation *rel) { List *attnamelist = NIL; TupleDesc desc = RelationGetDescr(rel->rel); int attnum; for (attnum = 0; attnum < desc->natts; attnum++) { int remoteattnum = physatt_in_attmap(rel, attnum); /* Skip dropped and generated attributes. */ if (TupleDescAttr(desc, attnum)->attisdropped || TupleDescAttr(desc, attnum)->attgenerated) continue; if (remoteattnum < 0) continue; attnamelist = lappend(attnamelist, makeString(rel->attnames[remoteattnum])); } return attnamelist; } /* * COPY single table over wire. */ static void copy_table_data(PGconn *origin_conn, PGconn *target_conn, SpockRemoteRel *remoterel, List *replication_sets) { SpockRelation *rel; PGresult *res; int bytes; char *copybuf; List *attnamelist; ListCell *lc; bool first; StringInfoData query; StringInfoData attlist; MemoryContext curctx = CurrentMemoryContext, oldctx; /* Build the relation map. */ StartTransactionCommand(); oldctx = MemoryContextSwitchTo(curctx); spock_relation_cache_updater(remoterel); rel = spock_relation_open(remoterel->relid, AccessShareLock); if (rel == NULL) ereport(ERROR, (errcode(ERRCODE_UNDEFINED_TABLE), errmsg("relation \"%s.%s\" does not exist", remoterel->nspname, remoterel->relname))); attnamelist = make_copy_attnamelist(rel); initStringInfo(&attlist); first = true; foreach(lc, attnamelist) { char *attname = strVal(lfirst(lc)); if (first) first = false; else appendStringInfoString(&attlist, ","); append_escaped_identifier(&attlist, origin_conn, attname); } MemoryContextSwitchTo(oldctx); spock_relation_close(rel, AccessShareLock); CommitTransactionCommand(); /* Build COPY TO query. */ initStringInfo(&query); appendStringInfoString(&query, "COPY "); /* * If the table is row-filtered we need to run query over the table to * execute the filter. */ if (remoterel->hasRowFilter) { StringInfoData relname; StringInfoData repsetarr; StringInfoData relnamelit; initStringInfo(&relname); append_escaped_qualified_name(&relname, origin_conn, remoterel->nspname, remoterel->relname); initStringInfo(&repsetarr); first = true; foreach(lc, replication_sets) { char *repset_name = lfirst(lc); if (first) first = false; else appendStringInfoChar(&repsetarr, ','); append_escaped_literal(&repsetarr, origin_conn, repset_name); } initStringInfo(&relnamelit); append_escaped_literal(&relnamelit, origin_conn, relname.data); appendStringInfo(&query, "(SELECT %s FROM spock.table_data_filtered(NULL::%s, %s::regclass, ARRAY[%s])) ", list_length(attnamelist) ? attlist.data : "*", relname.data, relnamelit.data, repsetarr.data); } else { if (remoterel->relkind == RELKIND_PARTITIONED_TABLE) { /* use COPY(SELECT...) for partitioned tables. */ appendStringInfo(&query, "(SELECT %s FROM ", list_length(attnamelist) ? attlist.data : "*"); append_escaped_qualified_name(&query, origin_conn, remoterel->nspname, remoterel->relname); appendStringInfoString(&query, ") "); } else { /* Otherwise just copy the table. */ append_escaped_qualified_name(&query, origin_conn, remoterel->nspname, remoterel->relname); appendStringInfoChar(&query, ' '); if (list_length(attnamelist)) appendStringInfo(&query, "(%s) ", attlist.data); } } appendStringInfoString(&query, "TO stdout"); /* Execute COPY TO. */ res = PQexec(origin_conn, query.data); if (PQresultStatus(res) != PGRES_COPY_OUT) { ereport(ERROR, (errmsg("table copy failed"), errdetail("Query '%s': %s", query.data, PQerrorMessage(origin_conn)))); } /* Build COPY FROM query. */ resetStringInfo(&query); appendStringInfoString(&query, "COPY "); append_escaped_qualified_name(&query, origin_conn, remoterel->nspname, remoterel->relname); appendStringInfoChar(&query, ' '); if (list_length(attnamelist)) appendStringInfo(&query, "(%s) ", attlist.data); appendStringInfoString(&query, "FROM stdin"); /* Execute COPY FROM. */ res = PQexec(target_conn, query.data); if (PQresultStatus(res) != PGRES_COPY_IN) { ereport(ERROR, (errmsg("table copy failed"), errdetail("Query '%s': %s", query.data, PQerrorMessage(origin_conn)))); } while ((bytes = PQgetCopyData(origin_conn, ©buf, false)) > 0) { if (PQputCopyData(target_conn, copybuf, bytes) != 1) { ereport(ERROR, (errmsg("writing to target table failed"), errdetail("destination connection reported: %s", PQerrorMessage(target_conn)))); } PQfreemem(copybuf); CHECK_FOR_INTERRUPTS(); } if (bytes != -1) { ereport(ERROR, (errmsg("reading from origin table failed"), errdetail("source connection returned %d: %s", bytes, PQerrorMessage(origin_conn)))); } /* Send local finish */ if (PQputCopyEnd(target_conn, NULL) != 1) { ereport(ERROR, (errmsg("sending copy-completion to destination connection failed"), errdetail("destination connection reported: %s", PQerrorMessage(target_conn)))); } /* Clear the initial COPY_IN result. */ PQclear(res); /* * Retrieve the final COPY result. PQputCopyEnd only signals end-of-data * at the protocol level; the actual outcome (constraint violations, etc.) * is available only via PQgetResult. */ res = PQgetResult(target_conn); if (PQresultStatus(res) != PGRES_COMMAND_OK) { char *msg = pstrdup(PQerrorMessage(target_conn)); const char *sqlstate = PQresultErrorField(res, PG_DIAG_SQLSTATE); int sqlerrcode = ERRCODE_INTERNAL_ERROR; /* * Keep the destination's error code, e.g. a constraint violation. Not * a connection code though: the monitor takes one of those for a lost * provider connection, and this session is local. A SQLSTATE is five * digits or capitals; take it only in that exact shape. */ if (sqlstate != NULL) { int i; for (i = 0; i < 5; i++) if (!((sqlstate[i] >= '0' && sqlstate[i] <= '9') || (sqlstate[i] >= 'A' && sqlstate[i] <= 'Z'))) break; if (i == 5 && sqlstate[5] == '\0') sqlerrcode = MAKE_SQLSTATE(sqlstate[0], sqlstate[1], sqlstate[2], sqlstate[3], sqlstate[4]); } if (spock_monitor_is_connection_error(sqlerrcode)) sqlerrcode = ERRCODE_INTERNAL_ERROR; PQclear(res); ereport(ERROR, (errcode(sqlerrcode), errmsg("COPY to destination table failed"), errdetail("destination connection reported: %s", msg))); } PQclear(res); elog(INFO, "finished synchronization of data for table %s.%s", remoterel->nspname, remoterel->relname); } /* * Copy data from origin node to target node. * * Creates new connection to origin and target. */ static void copy_tables_data(SpockSubscription *sub, const char *origin_dsn, const char *target_dsn, const char *origin_snapshot, List *tables, List *replication_sets, const char *origin_name) { PGconn *origin_conn; PGconn *target_conn; List *progress_entries_list = NIL; ListCell *lc; /* Connect to origin node. */ origin_conn = spock_connect(origin_dsn, sub->name, "copy"); start_copy_origin_tx(origin_conn, origin_snapshot); /* Connect to target node. */ target_conn = spock_connect(target_dsn, sub->name, "copy"); start_copy_target_tx(target_conn, origin_name); /* Copy every table. */ foreach(lc, tables) { RangeVar *rv = lfirst(lc); SpockRemoteRel *remoterel; remoterel = spock_get_remote_repset_table(origin_conn, rv, replication_sets); /* * In case of table partitioning, we synchronize the partitioned * (parent) table and skip the partitions. Other tables are * synchronized normally. */ if (!remoterel->ispartition) copy_table_data(origin_conn, target_conn, remoterel, replication_sets); CHECK_FOR_INTERRUPTS(); } progress_entries_list = read_provider_progress(origin_conn); /* Finish the transactions and disconnect. */ finish_copy_origin_tx(origin_conn); finish_copy_target_tx(target_conn); /* * Update replication progress. We must do it after commit of the COPY. * * NOTE: It is not obvious we need to arrange progress in case of * accidental single-table re-sync. But while this machinery serves * information goals only we just follow the initial logic. */ spock_group_progress_update_list(progress_entries_list); } /* * Copy data from origin node to target node. * * Creates new connection to origin and target. * * This is basically same as the copy_tables_data, but it can't be easily * merged to single function because we need to get list of tables here after * the transaction is bound to a snapshot. */ static List * copy_replication_sets_data(SpockSubscription *sub, const char *origin_dsn, const char *target_dsn, const char *origin_snapshot, List *replication_sets, const char *origin_name, List **progress_out) { PGconn *origin_conn; PGconn *target_conn; List *tables; ListCell *lc; /* Connect to origin node. */ origin_conn = spock_connect(origin_dsn, sub->name, "copy"); start_copy_origin_tx(origin_conn, origin_snapshot); /* * Read progress info within the same snapshot used for COPY so that the * LSN values are consistent with the data we are about to copy. */ if (progress_out) *progress_out = read_provider_progress(origin_conn); /* Get tables to copy from origin node. */ tables = spock_get_remote_repset_tables(origin_conn, replication_sets); /* * Drop tables we must not data-sync: any block_in_repset schema (a table * may have joined a repset before its schema was reserved), plus the * subscription's own skip_schema. Matched by the table's own namespace; * this does not descend partition hierarchies, so a partition physically * in a reserved schema under a non-reserved parent is not caught here * (rare: the parent copy would target a partition the reserved schema's * dump-exclusion left off the subscriber, so it errors rather than * silently importing the data). */ { MemoryContext synccxt = CurrentMemoryContext; List *blocked = NIL; List *filtered_tables = NIL; ListCell *lc_filter; /* * Reading the catalog needs a transaction and this function runs with * none open; copy the names into the persistent context so they * survive the commit. */ StartTransactionCommand(); { List *reserved = spock_reserved_object_names(RESERVED_KIND_SCHEMA, RESERVED_PURPOSE_REPSET); ListCell *lc_res; MemoryContext txcxt = MemoryContextSwitchTo(synccxt); foreach(lc_res, reserved) blocked = lappend(blocked, pstrdup((char *) lfirst(lc_res))); MemoryContextSwitchTo(txcxt); } CommitTransactionCommand(); /* Commit reset the current context; restore the sync context. */ MemoryContextSwitchTo(synccxt); foreach(lc_filter, tables) { SpockRemoteRel *remoterel = lfirst(lc_filter); /* Skip if the schema is blocked or on the sub's own skip list. */ if (name_in_list(blocked, remoterel->nspname) || name_in_list(sub->skip_schema, remoterel->nspname)) continue; filtered_tables = lappend(filtered_tables, remoterel); } list_free_deep(blocked); tables = filtered_tables; } /* Connect to target node. */ target_conn = spock_connect(target_dsn, sub->name, "copy"); start_copy_target_tx(target_conn, origin_name); /* Copy every table. */ foreach(lc, tables) { SpockRemoteRel *remoterel = lfirst(lc); /* * In case of table partitioning, we synchronize the partitioned * (parent) table and skip the partitions. Other tables are * synchronized normally. */ if (!remoterel->ispartition) copy_table_data(origin_conn, target_conn, remoterel, replication_sets); CHECK_FOR_INTERRUPTS(); } /* Finish the transactions and disconnect. */ finish_copy_origin_tx(origin_conn); finish_copy_target_tx(target_conn); return tables; } static void spock_sync_worker_cleanup(SpockSubscription *sub) { PGconn *origin_conn; /* Drop the slot on the remote side. */ origin_conn = spock_connect(sub->origin_if->dsn, sub->name, "cleanup"); /* Wait for slot to be free. */ while (!got_SIGTERM) { int rc; if (!spock_remote_slot_active(origin_conn, sub->slot_name)) break; rc = WaitLatch(&MyProc->procLatch, WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, 1000L); ResetLatch(&MyProc->procLatch); /* emergency bailout if postmaster has died */ if (rc & WL_POSTMASTER_DEATH) proc_exit(1); CHECK_FOR_INTERRUPTS(); } spock_drop_remote_slot(origin_conn, sub->slot_name); PQfinish(origin_conn); /* Drop the origin tracking locally. */ if (replorigin_session_origin != InvalidRepOriginId) { replorigin_session_reset(); replorigin_drop_by_name(sub->slot_name, true, true); replorigin_session_origin = InvalidRepOriginId; } } static void spock_sync_worker_cleanup_error_cb(int code, Datum arg) { SpockSubscription *sub = (SpockSubscription *) DatumGetPointer(arg); spock_sync_worker_cleanup(sub); } static void spock_sync_tmpfile_cleanup_cb(int code, Datum arg) { const char *tmpfile = DatumGetCString(arg); if (unlink(tmpfile) != 0 && errno != ENOENT) elog(WARNING, "Failed to clean up spock temporary dump file \"%s\" on exit/error: %m", tmpfile); } void spock_sync_subscription(SpockSubscription *sub) { SpockSyncStatus *sync; XLogRecPtr lsn; char status; MemoryContext myctx, oldctx; RepOriginId originid; /* We need our own context for keeping things between transactions. */ myctx = AllocSetContextCreate(CurrentMemoryContext, "spock_sync_subscription cxt", ALLOCSET_DEFAULT_SIZES); StartTransactionCommand(); oldctx = MemoryContextSwitchTo(myctx); sync = get_subscription_sync_status(sub->id, false); MemoryContextSwitchTo(oldctx); CommitTransactionCommand(); status = sync->status; switch (status) { /* Already synced, nothing to do except cleanup. */ case SYNC_STATUS_READY: MemoryContextDelete(myctx); return; /* We can recover from crashes during these. */ case SYNC_STATUS_INIT: case SYNC_STATUS_CATCHUP: break; /* * If the 'apply worker' is found in a state where it can't * continue, it forcibly disables the subscription and logs a * complaint, providing as much information as possible. At this * stage, we only have the origin_id to display. May we add * something to the log to facilitate the resolution of the issue? * * NOTE: A transaction is required since the exception_log is a * database table. */ default: { int old_exception_behaviour = exception_behaviour; StartTransactionCommand(); originid = replorigin_by_name(sub->slot_name, true); exception_behaviour = SUB_DISABLE; spock_disable_subscription(sub, originid, InvalidTransactionId, InvalidXLogRecPtr, 0); exception_behaviour = old_exception_behaviour; CommitTransactionCommand(); ereport(FATAL, (errcode(ERRCODE_INTERNAL_ERROR), errmsg("Spock's subscriber %s initialization failed during" " nonrecoverable step (%c). Subscription disabled.", sub->name, status), errhint("Check server errors and 'exception_log', fix " "the issue and try the setup again"))); } break; } if (status == SYNC_STATUS_INIT) { PGconn *origin_conn; PGconn *origin_conn_repl; char *snapshot; List *progress_entries_list = NIL; elog(INFO, "initializing subscriber %s", sub->name); origin_conn = spock_connect(sub->origin_if->dsn, sub->name, "snap"); /* * An incompatible upstream is permanent, so disable the subscription * rather than let the worker crash-loop on every restart. Checked * before the replication connection and slot are created. No * transaction is open (the switch above committed its own), so * spock_disable_subscription() commits its own and the disable * survives the FATAL; origin_conn is left open to keep * remoteversion_str valid. */ if (SyncKindStructure(sync->kind) && !upstream_version_supports_structure_sync(origin_conn)) { const char *remoteversion_str = PQparameterStatus(origin_conn, "server_version"); Assert(!IsTransactionState()); exception_behaviour = SUB_DISABLE; spock_disable_subscription(sub, InvalidRepOriginId, InvalidTransactionId, InvalidXLogRecPtr, 0); ereport(FATAL, (errcode(ERRCODE_FEATURE_NOT_SUPPORTED), errmsg("SPOCK %s: cannot synchronize structure from upstream server version %s", sub->name, remoteversion_str ? remoteversion_str : "unknown"), errdetail("The subscriber's pg_dump/pg_restore (version %s) cannot dump from a newer major version.", PG_VERSION), errhint("Synchronize the schema another way and recreate the subscription with synchronize_structure = false, then re-enable the subscription."))); } origin_conn_repl = spock_connect_replica(sub->origin_if->dsn, sub->name, "snap"); /* * Pause apply workers, create slot via replication protocol (returns * snapshot consistent with slot's WAL position), read peer progress, * and resume workers. */ PG_TRY(); { snapshot = spock_create_slot_and_read_progress( origin_conn, origin_conn_repl, sub->slot_name, MySubscription->origin->id, MySubscription->target->id, &lsn, &progress_entries_list); } PG_CATCH(); { MemoryContext savecxt; ErrorData *edata; /* * CopyErrorData() requires that we are NOT running in * ErrorContext, otherwise its assertion in elog.c trips on * cassert builds and the apply worker dies with SIGABRT. Switch * into our long-lived sync context first. */ savecxt = MemoryContextSwitchTo(myctx); edata = CopyErrorData(); /* Do NOT call FlushErrorState() here. */ elog(LOG, "SPOCK cswp error sub=%s slot=%s: %s", sub->name, sub->slot_name, edata->message ? edata->message : ""); /* * Best-effort resume of apply workers on the remote node. If the * connection is broken this will fail silently — the workers' * CV timeout will recover them. */ if (origin_conn && PQstatus(origin_conn) == CONNECTION_OK) { PGresult *rres = PQexec(origin_conn, "SELECT spock.resume_apply_workers()"); if (rres) PQclear(rres); } FreeErrorData(edata); MemoryContextSwitchTo(savecxt); PG_RE_THROW(); } PG_END_TRY(); /* * origin_conn transaction remains open — snapshot held for COPY. * Keep origin_conn_repl open too — the exported snapshot file in * pg_snapshots/ is tied to this connection and needed by pg_dump * during structure sync. */ PG_ENSURE_ERROR_CLEANUP(spock_sync_worker_cleanup_error_cb, PointerGetDatum(sub)); { char tmpfile[MAXPGPATH]; snprintf(tmpfile, MAXPGPATH, "%s/spock-%d.dump", spock_temp_directory, MyProcPid); canonicalize_path(tmpfile); PG_ENSURE_ERROR_CLEANUP_SUFFIX(spock_sync_tmpfile_cleanup_cb, CStringGetDatum(tmpfile), _suf); { Relation replorigin_rel; StartTransactionCommand(); originid = ensure_replication_origin(sub->slot_name); elog(DEBUG3, "advancing origin with oid %u for forwarded row to %X/%X during subscription sync", originid, (uint32) (XactLastCommitEnd >> 32), (uint32) XactLastCommitEnd); replorigin_rel = table_open(ReplicationOriginRelationId, RowExclusiveLock); replorigin_advance(originid, lsn, XactLastCommitEnd, true, true); table_close(replorigin_rel, RowExclusiveLock); CommitTransactionCommand(); if (SyncKindStructure(sync->kind)) { elog(INFO, "synchronizing structure"); status = SYNC_STATUS_STRUCTURE; StartTransactionCommand(); set_subscription_sync_status(sub->id, status); CommitTransactionCommand(); /* Need catalog lookups, hence, transactional context */ StartTransactionCommand(); /* Dump structure to temp storage. */ dump_structure(sub, tmpfile, snapshot); /* Restore base pre-data structure (types, tables, etc). */ restore_structure(sub, tmpfile, "pre-data"); CommitTransactionCommand(); } /* Copy data. */ if (SyncKindData(sync->kind)) { List *tables; ListCell *lc; elog(INFO, "synchronizing data"); status = SYNC_STATUS_DATA; StartTransactionCommand(); set_subscription_sync_status(sub->id, status); CommitTransactionCommand(); tables = copy_replication_sets_data(sub, sub->origin_if->dsn, sub->target_if->dsn, snapshot, sub->replication_sets, sub->slot_name, NULL); /* * Arrange replication status according to the just copied * data. */ spock_group_progress_force_set_list(progress_entries_list); /* Store info about all the synchronized tables. */ StartTransactionCommand(); foreach(lc, tables) { SpockRemoteRel *remoterel = lfirst(lc); SpockSyncStatus *oldsync; oldsync = get_table_sync_status(sub->id, remoterel->nspname, remoterel->relname, true); if (oldsync) { set_table_sync_status(sub->id, remoterel->nspname, remoterel->relname, SYNC_STATUS_READY, lsn); } else { SpockSyncStatus newsync; newsync.kind = SYNC_KIND_FULL; newsync.subid = sub->id; namestrcpy(&newsync.nspname, remoterel->nspname); namestrcpy(&newsync.relname, remoterel->relname); newsync.status = SYNC_STATUS_READY; newsync.statuslsn = lsn; create_local_sync_status(&newsync); } } CommitTransactionCommand(); } /* Restore post-data structure (indexes, constraints, etc). */ if (SyncKindStructure(sync->kind)) { elog(INFO, "synchronizing constraints"); status = SYNC_STATUS_CONSTRAINTS; StartTransactionCommand(); set_subscription_sync_status(sub->id, status); CommitTransactionCommand(); restore_structure(sub, tmpfile, "post-data"); } } PG_END_ENSURE_ERROR_CLEANUP_SUFFIX(spock_sync_tmpfile_cleanup_cb, CStringGetDatum(tmpfile), _suf); spock_sync_tmpfile_cleanup_cb(0, CStringGetDatum(tmpfile)); } PG_END_ENSURE_ERROR_CLEANUP(spock_sync_worker_cleanup_error_cb, PointerGetDatum(sub)); PQfinish(origin_conn_repl); spock_release_slot_snapshot(origin_conn); status = SYNC_STATUS_CATCHUP; StartTransactionCommand(); set_subscription_sync_status(sub->id, status); CommitTransactionCommand(); } if (status == SYNC_STATUS_CATCHUP) { /* Nothing to do here yet. */ status = SYNC_STATUS_READY; StartTransactionCommand(); set_subscription_sync_status(sub->id, status); CommitTransactionCommand(); elog(INFO, "finished synchronization of subscriber %s, ready to enter normal replication", sub->name); } MemoryContextDelete(myctx); } char spock_sync_table(SpockSubscription *sub, RangeVar *table, XLogRecPtr *status_lsn) { PGconn *origin_conn_repl, *origin_conn; RepOriginId originid; char *snapshot; SpockSyncStatus *sync; StartTransactionCommand(); /* Sanity check. */ sync = get_subscription_sync_status(sub->id, false); if (sync->status != SYNC_STATUS_READY) { elog(ERROR, "subscriber %s is not ready, cannot synchronize individual tables", sub->name); } /* Check current state of the table. */ sync = get_table_sync_status(sub->id, table->schemaname, table->relname, false); *status_lsn = sync->statuslsn; /* Already synchronized, nothing to do here. */ if (sync->status == SYNC_STATUS_READY || sync->status == SYNC_STATUS_SYNCDONE) { CommitTransactionCommand(); return sync->status; } /* * If previous sync attempt failed, we need to start from beginning or * signal that we failed. */ if (sync->status != SYNC_STATUS_INIT) { /* * We don't use the if-branch at the moment. Just check to be * paranoid. */ Assert(sync->stop_on_error); /* * Do not continue if any error happened during a previous SYNC * attempt on this table. Just mark the status and exit. */ set_table_sync_status(sub->id, table->schemaname, table->relname, SYNC_STATUS_FAILED, InvalidXLogRecPtr); CommitTransactionCommand(); return SYNC_STATUS_FAILED; } /* Should be removed when we add stop_on_error as a sync option */ Assert(sync->status == SYNC_STATUS_INIT); /* We initiate sync procedure. Switch state to the next value */ set_table_sync_status(sub->id, table->schemaname, table->relname, SYNC_STATUS_STARTED, InvalidXLogRecPtr); CommitTransactionCommand(); origin_conn_repl = spock_connect_replica(sub->origin_if->dsn, sub->name, "copy"); origin_conn = spock_connect(sub->origin_if->dsn, sub->name, "copy_slot"); snapshot = ensure_replication_slot_snapshot(origin_conn, origin_conn_repl, sub->slot_name, false, status_lsn); PQfinish(origin_conn); /* Make sure we cleanup the slot if something goes wrong. */ PG_ENSURE_ERROR_CLEANUP(spock_sync_worker_cleanup_error_cb, PointerGetDatum(sub)); { Relation replorigin_rel; StartTransactionCommand(); originid = ensure_replication_origin(sub->slot_name); elog(DEBUG2, "advancing origin %s (oid %u) for forwarded row to %X/%X after sync error", MySubscription->slot_name, originid, (uint32) (XactLastCommitEnd >> 32), (uint32) XactLastCommitEnd); replorigin_rel = table_open(ReplicationOriginRelationId, RowExclusiveLock); replorigin_advance(originid, *status_lsn, XactLastCommitEnd, true, true); table_close(replorigin_rel, RowExclusiveLock); set_table_sync_status(sub->id, table->schemaname, table->relname, SYNC_STATUS_DATA, *status_lsn); CommitTransactionCommand(); /* Copy data. */ copy_tables_data(sub, sub->origin_if->dsn, sub->target_if->dsn, snapshot, list_make1(table), sub->replication_sets, sub->slot_name); } PG_END_ENSURE_ERROR_CLEANUP(spock_sync_worker_cleanup_error_cb, PointerGetDatum(sub)); PQfinish(origin_conn_repl); return SYNC_STATUS_SYNCWAIT; } void spock_sync_worker_finish(void) { SpockWorker *apply; /* * Commit any outstanding transaction. This is the usual case, unless * there was nothing to do for the table. */ if (IsTransactionState()) { CommitTransactionCommand(); pgstat_report_stat(false); } /* And flush all writes. */ XLogFlush(GetXLogWriteRecPtr()); StartTransactionCommand(); spock_sync_worker_cleanup(MySubscription); CommitTransactionCommand(); /* * In case there is apply process running, it might be waiting for the * table status change so tell it to check. */ LWLockAcquire(SpockCtx->lock, LW_EXCLUSIVE); apply = spock_apply_find(MySpockWorker->dboid, MyApplyWorker->subid); if (spock_worker_running(apply)) SetLatch(&apply->proc->procLatch); LWLockRelease(SpockCtx->lock); elog(LOG, "finished sync of table %s.%s for subscriber %s", NameStr(MySyncWorker->nspname), NameStr(MySyncWorker->relname), MySubscription->name); if (!sync_worker_failed) { spock_monitor_count(SPOCK_MONITOR_TABLES_SYNCED, 1); spock_monitor_record_event(SPOCK_EVENT_SYNC_FINISHED, MySubscription->id, InvalidXLogRecPtr, "table %s.%s", NameStr(MySyncWorker->nspname), NameStr(MySyncWorker->relname)); } } void spock_sync_main(Datum main_arg) { int slot = DatumGetInt32(main_arg); PGconn *streamConn; RepOriginId originid; XLogRecPtr lsn; XLogRecPtr status_lsn; StringInfoData slot_name; RangeVar *copytable = NULL; MemoryContext saved_ctx; char *tablename; char status; /* Setup shmem. */ spock_worker_attach(slot, SPOCK_WORKER_SYNC); MySyncWorker = &MySpockWorker->worker.sync; MyApplyWorker = &MySyncWorker->apply; /* Attach to dsm segment. */ Assert(CurrentResourceOwner == NULL); CurrentResourceOwner = ResourceOwnerCreate(NULL, "spock sync"); /* Setup synchronous commit according to the user's wishes */ SetConfigOption("synchronous_commit", spock_synchronous_commit ? "local" : "off", PGC_BACKEND, PGC_S_OVERRIDE); /* other context? */ /* Run as replica session replication role. */ SetConfigOption("session_replication_role", "replica", PGC_SUSET, PGC_S_OVERRIDE); /* other context? */ /* * Disable function body checks during replay. That's necessary because a) * the creator of the function might have had it disabled b) the function * might be search_path dependant and we don't fix the contents of * functions. */ SetConfigOption("check_function_bodies", "off", PGC_INTERNAL, PGC_S_OVERRIDE); StartTransactionCommand(); saved_ctx = MemoryContextSwitchTo(TopMemoryContext); MySubscription = get_subscription(MySyncWorker->apply.subid); MemoryContextSwitchTo(saved_ctx); CommitTransactionCommand(); copytable = makeRangeVar(NameStr(MySyncWorker->nspname), NameStr(MySyncWorker->relname), -1); tablename = quote_qualified_identifier(copytable->schemaname, copytable->relname); initStringInfo(&slot_name); appendStringInfo(&slot_name, "%s_%08x", MySubscription->slot_name, DatumGetUInt32(hash_any((unsigned char *) tablename, strlen(tablename)))); MySubscription->slot_name = slot_name.data; elog(LOG, "starting sync of table %s.%s for subscriber %s", copytable->schemaname, copytable->relname, MySubscription->name); spock_monitor_record_event(SPOCK_EVENT_SYNC_STARTED, MySubscription->id, InvalidXLogRecPtr, "table %s.%s", copytable->schemaname, copytable->relname); elog(DEBUG1, "connecting to provider %s, dsn %s", MySubscription->origin_if->name, MySubscription->origin_if->dsn); /* Do the initial sync first. */ status = spock_sync_table(MySubscription, copytable, &status_lsn); if (status == SYNC_STATUS_FAILED) { /* * spock_sync_table() gives up when an earlier attempt on this table * failed; that attempt recorded its own sync_failed with the error. */ sync_worker_failed = true; spock_monitor_record_event(SPOCK_EVENT_SYNC_FAILED, MySubscription->id, InvalidXLogRecPtr, "table %s.%s abandoned after an earlier failure", copytable->schemaname, copytable->relname); } if (status == SYNC_STATUS_SYNCDONE || status == SYNC_STATUS_READY || status == SYNC_STATUS_FAILED) { spock_sync_worker_finish(); proc_exit(0); } /* Wait for ack from the main apply thread. */ StartTransactionCommand(); set_table_sync_status(MySubscription->id, copytable->schemaname, copytable->relname, SYNC_STATUS_SYNCWAIT, status_lsn); CommitTransactionCommand(); wait_for_sync_status_change(MySubscription->id, copytable->schemaname, copytable->relname, SYNC_STATUS_CATCHUP, &lsn); Assert(lsn == status_lsn); /* Setup the origin and get the starting position for the replication. */ StartTransactionCommand(); originid = replorigin_by_name(MySubscription->slot_name, false); elog(DEBUG2, "setting origin %s (oid %u) for subscription sync", MySubscription->slot_name, originid); replorigin_session_setup(originid); replorigin_session_origin = originid; spock_session_origin_id = originid; Assert(status_lsn == replorigin_session_get_progress(false)); /* * In case there is nothing to catchup, finish immediately. Note * spock_sync_worker_finish() will commit. */ if (status_lsn >= MyApplyWorker->replay_stop_lsn) { /* Mark local table as done. */ set_table_sync_status(MySubscription->id, copytable->schemaname, copytable->relname, SYNC_STATUS_SYNCDONE, status_lsn); spock_sync_worker_finish(); proc_exit(0); } CommitTransactionCommand(); /* Start the replication. */ streamConn = spock_connect_replica(MySubscription->origin_if->dsn, MySubscription->name, "catchup"); /* * IDENTIFY_SYSTEM sets up some internal state on walsender so call it * even if we don't (yet) want to use any of the results. */ spock_identify_system(streamConn, NULL, NULL, NULL, NULL); spock_start_replication(streamConn, MySubscription->slot_name, status_lsn, "all", NULL, tablename, MySubscription->force_text_transfer); /* Leave it to standard apply code to do the replication. */ apply_work(streamConn); PQfinish(streamConn); /* * We should only get here if we received sigTERM, which in case of sync * worker is not expected. */ proc_exit(1); } /* Catalog access */ /* Create subscription sync status record in catalog. */ void create_local_sync_status(SpockSyncStatus *sync) { RangeVar *rv; Relation rel; TupleDesc tupDesc; HeapTuple tup; Datum values[Natts_local_sync_state]; bool nulls[Natts_local_sync_state]; Assert(IsTransactionState()); rv = makeRangeVar(EXTENSION_NAME, CATALOG_LOCAL_SYNC_STATUS, -1); rel = table_openrv(rv, RowExclusiveLock); tupDesc = RelationGetDescr(rel); /* Form a tuple. */ memset(nulls, false, sizeof(nulls)); values[Anum_sync_kind - 1] = CharGetDatum(sync->kind); values[Anum_sync_subid - 1] = ObjectIdGetDatum(sync->subid); if (sync->nspname.data[0]) values[Anum_sync_nspname - 1] = NameGetDatum(&sync->nspname); else nulls[Anum_sync_nspname - 1] = true; if (sync->relname.data[0]) values[Anum_sync_relname - 1] = NameGetDatum(&sync->relname); else nulls[Anum_sync_relname - 1] = true; values[Anum_sync_status - 1] = CharGetDatum(sync->status); values[Anum_sync_statuslsn - 1] = LSNGetDatum(sync->statuslsn); tup = heap_form_tuple(tupDesc, values, nulls); /* Insert the tuple to the catalog. */ CatalogTupleInsert(rel, tup); /* Cleanup. */ heap_freetuple(tup); table_close(rel, RowExclusiveLock); } /* Remove subscription sync status record from catalog. */ void drop_subscription_sync_status(Oid subid) { RangeVar *rv; Relation rel; SysScanDesc scan; HeapTuple tuple; ScanKeyData key[1]; Assert(IsTransactionState()); rv = makeRangeVar(EXTENSION_NAME, CATALOG_LOCAL_SYNC_STATUS, -1); rel = table_openrv(rv, RowExclusiveLock); ScanKeyInit(&key[0], Anum_sync_subid, BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(subid)); scan = systable_beginscan(rel, 0, true, NULL, 1, key); /* Remove the tuples. */ while (HeapTupleIsValid(tuple = systable_getnext(scan))) simple_heap_delete(rel, &tuple->t_self); /* Cleanup. */ systable_endscan(scan); table_close(rel, RowExclusiveLock); } static SpockSyncStatus * syncstatus_fromtuple(HeapTuple tuple, TupleDesc desc) { SpockSyncStatus *sync; Datum d; bool isnull; sync = (SpockSyncStatus *) palloc0(sizeof(SpockSyncStatus)); d = fastgetattr(tuple, Anum_sync_kind, desc, &isnull); Assert(!isnull); sync->kind = DatumGetChar(d); d = fastgetattr(tuple, Anum_sync_subid, desc, &isnull); Assert(!isnull); sync->subid = DatumGetObjectId(d); d = fastgetattr(tuple, Anum_sync_nspname, desc, &isnull); if (!isnull) namestrcpy(&sync->nspname, NameStr(*DatumGetName(d))); d = fastgetattr(tuple, Anum_sync_relname, desc, &isnull); if (!isnull) namestrcpy(&sync->relname, NameStr(*DatumGetName(d))); d = fastgetattr(tuple, Anum_sync_status, desc, &isnull); Assert(!isnull); sync->status = DatumGetChar(d); d = fastgetattr(tuple, Anum_sync_statuslsn, desc, &isnull); Assert(!isnull); sync->statuslsn = DatumGetLSN(d); if (strlen(sync->nspname.data) > 0 && strlen(sync->relname.data) > 0) { /* * XXX: Adding a parameter to a stable extension's UI and DB table * always needs careful coding. So, let's do it step-by-step and for * now change defaults and stop attempting single table * synchronisation on an error. Keep it false for subscriptions * syncing purposes. We don't really use it in subscription syncing * code now - it should be covered by the following commits. */ sync->stop_on_error = true; } return sync; } /* Get the sync status for a subscription. */ SpockSyncStatus * get_subscription_sync_status(Oid subid, bool missing_ok) { SpockSyncStatus *sync; RangeVar *rv; Relation rel; SysScanDesc scan; HeapTuple tuple; ScanKeyData key[1]; TupleDesc tupDesc; Assert(IsTransactionState()); rv = makeRangeVar(EXTENSION_NAME, CATALOG_LOCAL_SYNC_STATUS, -1); rel = table_openrv(rv, RowExclusiveLock); tupDesc = RelationGetDescr(rel); ScanKeyInit(&key[0], Anum_sync_subid, BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(subid)); scan = systable_beginscan(rel, 0, true, NULL, 1, key); while (HeapTupleIsValid(tuple = systable_getnext(scan))) { if (spk_heap_attisnull(tuple, Anum_sync_nspname, NULL) && spk_heap_attisnull(tuple, Anum_sync_relname, NULL)) break; } if (!HeapTupleIsValid(tuple)) { if (missing_ok) { systable_endscan(scan); table_close(rel, RowExclusiveLock); return NULL; } elog(ERROR, "subscription %u status not found", subid); } sync = syncstatus_fromtuple(tuple, tupDesc); systable_endscan(scan); table_close(rel, RowExclusiveLock); return sync; } /* Set the sync status for a subscription. */ void set_subscription_sync_status(Oid subid, char status) { RangeVar *rv; Relation rel; TupleDesc tupDesc; SysScanDesc scan; HeapTuple oldtup, newtup; ScanKeyData key[1]; Datum values[Natts_local_sync_state]; bool nulls[Natts_local_sync_state]; bool replaces[Natts_local_sync_state]; rv = makeRangeVar(EXTENSION_NAME, CATALOG_LOCAL_SYNC_STATUS, -1); rel = table_openrv(rv, RowExclusiveLock); tupDesc = RelationGetDescr(rel); ScanKeyInit(&key[0], Anum_sync_subid, BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(subid)); scan = systable_beginscan(rel, 0, true, NULL, 1, key); while (HeapTupleIsValid(oldtup = systable_getnext(scan))) { if (spk_heap_attisnull(oldtup, Anum_sync_nspname, NULL) && spk_heap_attisnull(oldtup, Anum_sync_relname, NULL)) break; } if (!HeapTupleIsValid(oldtup)) elog(ERROR, "subscription %u status not found", subid); memset(nulls, false, sizeof(nulls)); memset(replaces, false, sizeof(replaces)); values[Anum_sync_status - 1] = CharGetDatum(status); replaces[Anum_sync_status - 1] = true; values[Anum_sync_statuslsn - 1] = LSNGetDatum(InvalidXLogRecPtr); replaces[Anum_sync_statuslsn - 1] = true; newtup = heap_modify_tuple(oldtup, tupDesc, values, nulls, replaces); /* Update the tuple in catalog. */ CatalogTupleUpdate(rel, &oldtup->t_self, newtup); /* Cleanup. */ heap_freetuple(newtup); systable_endscan(scan); table_close(rel, RowExclusiveLock); } /* Remove table sync status record from catalog. */ void drop_table_sync_status(const char *nspname, const char *relname) { RangeVar *rv; Relation rel; SysScanDesc scan; HeapTuple tuple; ScanKeyData key[2]; Assert(IsTransactionState()); rv = makeRangeVar(EXTENSION_NAME, CATALOG_LOCAL_SYNC_STATUS, -1); rel = table_openrv(rv, RowExclusiveLock); ScanKeyInit(&key[0], Anum_sync_nspname, BTEqualStrategyNumber, F_NAMEEQ, CStringGetDatum(nspname)); ScanKeyInit(&key[1], Anum_sync_relname, BTEqualStrategyNumber, F_NAMEEQ, CStringGetDatum(relname)); scan = systable_beginscan(rel, 0, true, NULL, 2, key); /* Remove the tuples. */ while (HeapTupleIsValid(tuple = systable_getnext(scan))) simple_heap_delete(rel, &tuple->t_self); /* Cleanup. */ systable_endscan(scan); table_close(rel, RowExclusiveLock); } /* Remove table sync status record from catalog. */ void drop_table_sync_status_for_sub(Oid subid, const char *nspname, const char *relname) { RangeVar *rv; Relation rel; SysScanDesc scan; HeapTuple tuple; ScanKeyData key[3]; Assert(IsTransactionState()); rv = makeRangeVar(EXTENSION_NAME, CATALOG_LOCAL_SYNC_STATUS, -1); rel = table_openrv(rv, RowExclusiveLock); ScanKeyInit(&key[0], Anum_sync_subid, BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(subid)); ScanKeyInit(&key[1], Anum_sync_nspname, BTEqualStrategyNumber, F_NAMEEQ, CStringGetDatum(nspname)); ScanKeyInit(&key[2], Anum_sync_relname, BTEqualStrategyNumber, F_NAMEEQ, CStringGetDatum(relname)); scan = systable_beginscan(rel, 0, true, NULL, 3, key); /* Remove the tuples. */ while (HeapTupleIsValid(tuple = systable_getnext(scan))) simple_heap_delete(rel, &tuple->t_self); /* Cleanup. */ systable_endscan(scan); table_close(rel, RowExclusiveLock); } /* Get the sync status for a table. */ SpockSyncStatus * get_table_sync_status(Oid subid, const char *nspname, const char *relname, bool missing_ok) { SpockSyncStatus *sync; RangeVar *rv; Relation rel; SysScanDesc scan; HeapTuple tuple; ScanKeyData key[3]; TupleDesc tupDesc; Oid idxoid = InvalidOid; List *indexes; ListCell *l; Assert(IsTransactionState()); rv = makeRangeVar(EXTENSION_NAME, CATALOG_LOCAL_SYNC_STATUS, -1); rel = table_openrv(rv, RowExclusiveLock); /* Find an index we can use to scan this catalog. */ indexes = RelationGetIndexList(rel); foreach(l, indexes) { Relation idx = index_open(lfirst_oid(l), AccessShareLock); if (idx->rd_index->indkey.values[0] == Anum_sync_subid && idx->rd_index->indkey.values[1] == Anum_sync_nspname && idx->rd_index->indkey.values[2] == Anum_sync_relname) { idxoid = lfirst_oid(l); index_close(idx, AccessShareLock); break; } index_close(idx, AccessShareLock); } if (!OidIsValid(idxoid)) elog(ERROR, "could not find index on local_sync_status"); list_free(indexes); tupDesc = RelationGetDescr(rel); ScanKeyInit(&key[0], Anum_sync_subid, BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(subid)); ScanKeyInit(&key[1], Anum_sync_nspname, BTEqualStrategyNumber, F_NAMEEQ, CStringGetDatum(nspname)); ScanKeyInit(&key[2], Anum_sync_relname, BTEqualStrategyNumber, F_NAMEEQ, CStringGetDatum(relname)); scan = systable_beginscan(rel, idxoid, true, NULL, 3, key); tuple = systable_getnext(scan); if (!HeapTupleIsValid(tuple)) { if (missing_ok) { systable_endscan(scan); table_close(rel, RowExclusiveLock); return NULL; } elog(ERROR, "subscription %u table %s.%s status not found", subid, nspname, relname); } sync = syncstatus_fromtuple(tuple, tupDesc); systable_endscan(scan); table_close(rel, RowExclusiveLock); return sync; } /* Get the sync status for a table. */ List * get_unsynced_tables(Oid subid) { SpockSyncStatus *sync; RangeVar *rv; Relation rel; SysScanDesc scan; HeapTuple tuple; ScanKeyData key[1]; List *res = NIL; TupleDesc tupDesc; Assert(IsTransactionState()); rv = makeRangeVar(EXTENSION_NAME, CATALOG_LOCAL_SYNC_STATUS, -1); rel = table_openrv(rv, RowExclusiveLock); tupDesc = RelationGetDescr(rel); ScanKeyInit(&key[0], Anum_sync_subid, BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(subid)); scan = systable_beginscan(rel, 0, true, NULL, 1, key); while (HeapTupleIsValid(tuple = systable_getnext(scan))) { if (spk_heap_attisnull(tuple, Anum_sync_nspname, NULL) && spk_heap_attisnull(tuple, Anum_sync_relname, NULL)) continue; sync = syncstatus_fromtuple(tuple, tupDesc); if (sync->status != SYNC_STATUS_READY) res = lappend(res, sync); } systable_endscan(scan); table_close(rel, RowExclusiveLock); return res; } /* Get the sync status for all tables known to subscription. */ List * get_subscription_tables(Oid subid) { SpockSyncStatus *sync; RangeVar *rv; Relation rel; SysScanDesc scan; HeapTuple tuple; ScanKeyData key[1]; List *res = NIL; TupleDesc tupDesc; Assert(IsTransactionState()); rv = makeRangeVar(EXTENSION_NAME, CATALOG_LOCAL_SYNC_STATUS, -1); rel = table_openrv(rv, RowExclusiveLock); tupDesc = RelationGetDescr(rel); ScanKeyInit(&key[0], Anum_sync_subid, BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(subid)); scan = systable_beginscan(rel, 0, true, NULL, 1, key); while (HeapTupleIsValid(tuple = systable_getnext(scan))) { if (spk_heap_attisnull(tuple, Anum_sync_nspname, NULL) && spk_heap_attisnull(tuple, Anum_sync_relname, NULL)) continue; sync = syncstatus_fromtuple(tuple, tupDesc); res = lappend(res, sync); } systable_endscan(scan); table_close(rel, RowExclusiveLock); return res; } /* Set the sync status for a table. */ void set_table_sync_status(Oid subid, const char *nspname, const char *relname, char status, XLogRecPtr statuslsn) { RangeVar *rv; Relation rel; TupleDesc tupDesc; SysScanDesc scan; HeapTuple oldtup, newtup; ScanKeyData key[3]; Datum values[Natts_local_sync_state]; bool nulls[Natts_local_sync_state]; bool replaces[Natts_local_sync_state]; rv = makeRangeVar(EXTENSION_NAME, CATALOG_LOCAL_SYNC_STATUS, -1); rel = table_openrv(rv, RowExclusiveLock); tupDesc = RelationGetDescr(rel); ScanKeyInit(&key[0], Anum_sync_subid, BTEqualStrategyNumber, F_OIDEQ, ObjectIdGetDatum(subid)); ScanKeyInit(&key[1], Anum_sync_nspname, BTEqualStrategyNumber, F_NAMEEQ, CStringGetDatum(nspname)); ScanKeyInit(&key[2], Anum_sync_relname, BTEqualStrategyNumber, F_NAMEEQ, CStringGetDatum(relname)); scan = systable_beginscan(rel, 0, true, NULL, 3, key); oldtup = systable_getnext(scan); if (!HeapTupleIsValid(oldtup)) elog(ERROR, "subscription %u table %s.%s status not found", subid, nspname, relname); memset(nulls, false, sizeof(nulls)); memset(replaces, false, sizeof(replaces)); values[Anum_sync_status - 1] = CharGetDatum(status); replaces[Anum_sync_status - 1] = true; values[Anum_sync_statuslsn - 1] = LSNGetDatum(statuslsn); replaces[Anum_sync_statuslsn - 1] = true; newtup = heap_modify_tuple(oldtup, tupDesc, values, nulls, replaces); /* Update the tuple in catalog. */ CatalogTupleUpdate(rel, &oldtup->t_self, newtup); /* Cleanup. */ heap_freetuple(newtup); systable_endscan(scan); table_close(rel, RowExclusiveLock); } /* * Wait until the table sync status has changed desired one. * * We also exit if the worker is no longer recognized as sync worker as * that means something bad happened to it. * * Care with allocations is required here since it typically runs * in TopMemoryContext. */ bool wait_for_sync_status_change(Oid subid, const char *nspname, const char *relname, char desired_state, XLogRecPtr *lsn) { int rc; MemoryContext old_ctx = CurrentMemoryContext; bool ret = false; *lsn = InvalidXLogRecPtr; Assert(!IsTransactionState()); while (!got_SIGTERM) { SpockWorker *worker; SpockSyncStatus *sync; StartTransactionCommand(); sync = get_table_sync_status(subid, nspname, relname, true); if (!sync) { CommitTransactionCommand(); break; } if (sync->status == desired_state) { *lsn = sync->statuslsn; CommitTransactionCommand(); ret = true; break; } CommitTransactionCommand(); (void) MemoryContextSwitchTo(old_ctx); /* Check if the worker is still alive - no point waiting if it died. */ LWLockAcquire(SpockCtx->lock, LW_EXCLUSIVE); worker = spock_sync_find(MyDatabaseId, subid, nspname, relname); LWLockRelease(SpockCtx->lock); if (!worker) break; rc = WaitLatch(&MyProc->procLatch, WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH, 60000L); ResetLatch(&MyProc->procLatch); /* emergency bailout if postmaster has died */ if (rc & WL_POSTMASTER_DEATH) proc_exit(1); CHECK_FOR_INTERRUPTS(); } (void) MemoryContextSwitchTo(old_ctx); return ret; } /* * Truncates table if it exists. */ void truncate_table(char *nspname, char *relname) { RangeVar *rv; Oid relid; TruncateStmt *truncate; StringInfoData sql; rv = makeRangeVar(nspname, relname, -1); relid = RangeVarGetRelid(rv, AccessExclusiveLock, true); if (relid == InvalidOid) return; initStringInfo(&sql); appendStringInfo(&sql, "TRUNCATE TABLE %s", quote_qualified_identifier(rv->schemaname, rv->relname)); /* Truncate the table. */ truncate = makeNode(TruncateStmt); truncate->relations = list_make1(rv); truncate->restart_seqs = false; truncate->behavior = DROP_CASCADE; ExecuteTruncate(truncate); /* release memory allocated to create SQL statement */ pfree(sql.data); CommandCounterIncrement(); }