From f5c80bc06391a5d455549ef26f7d64b3206d7b0c Mon Sep 17 00:00:00 2001 From: Asif Rehman Date: Mon, 22 Jun 2026 14:06:38 +0500 Subject: [PATCH 1/7] =?UTF-8?q?Add=20spock=5Fcreate=5Fsubscriber=20?= =?UTF-8?q?=E2=80=94=20bootstrap=20a=20subscriber=20from=20a=20physical=20?= =?UTF-8?q?basebackup?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds a standalone frontend utility that: - Takes a 'pg_basebackup' from the provider and replays to a named restore point. - Creates per-database logical slots on the provider (optionally '--drop-slot-if-exists'). - Starts the new node to catch up with the provider to the restore point. - Installs the Spock extension, creates/advances the replication origin to the restore-point LSN, and creates the subscription to subscribe to the provider. --- .gitignore | 1 - Makefile | 1 + utils/spock_create_subscriber/Makefile | 15 + .../spock_create_subscriber.c | 1775 +++++++++++++++++ 4 files changed, 1791 insertions(+), 1 deletion(-) create mode 100644 utils/spock_create_subscriber/Makefile create mode 100644 utils/spock_create_subscriber/spock_create_subscriber.c diff --git a/.gitignore b/.gitignore index 71736e1d0..0d9c3d3cd 100644 --- a/.gitignore +++ b/.gitignore @@ -1,7 +1,6 @@ results regression_output tmp_check -spock_create_subscriber .vimrc *.o *.so diff --git a/Makefile b/Makefile index 000138d72..523d5e3a1 100644 --- a/Makefile +++ b/Makefile @@ -9,6 +9,7 @@ EXTENSION = spock PGFILEDESC = "spock - multi-master replication" MODULES = spock_output +SUBDIRS = utils/spock_create_subscriber # Lookup source directory vpath % src src/compat/$(PGVER) diff --git a/utils/spock_create_subscriber/Makefile b/utils/spock_create_subscriber/Makefile new file mode 100644 index 000000000..54e394462 --- /dev/null +++ b/utils/spock_create_subscriber/Makefile @@ -0,0 +1,15 @@ +# Makefile for spock_create_subscriber utility +PG_CONFIG ?= pg_config +PROGRAM = spock_create_subscriber + +PG_CPPFLAGS = -I../../include -I$(shell $(PG_CONFIG) --includedir) +PG_LDFLAGS = -lpq -L$(shell $(PG_CONFIG) --libdir) + +# create symlink to spock_fe.c here +spock_fe.c: ../../src/spock_fe.c + ln -sf $< $@ +OBJS = spock_create_subscriber.o spock_fe.o + +# PGXS +PGXS := $(shell $(PG_CONFIG) --pgxs) +include $(PGXS) diff --git a/utils/spock_create_subscriber/spock_create_subscriber.c b/utils/spock_create_subscriber/spock_create_subscriber.c new file mode 100644 index 000000000..8147c68b0 --- /dev/null +++ b/utils/spock_create_subscriber/spock_create_subscriber.c @@ -0,0 +1,1775 @@ +/* ------------------------------------------------------------------------- + * + * spock_create_subscriber.c + * Initialize a new spock subscriber from a physical base backup + * + * Copyright (c) 2022-2024, pgEdge, Inc. + * Portions Copyright (c) 1996-2021, PostgreSQL Global Development Group + * Portions Copyright (c) 1994, The Regents of the University of California + * + * ------------------------------------------------------------------------- + */ + +/* dirent.h on port/win32_msvc expects MAX_PATH to be defined */ +#if defined(_WIN32) +#define WIN32_LEAN_AND_MEAN +#include +#endif + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +/* Note the order is important for debian here. */ +#if !defined(pg_attribute_printf) + +/* GCC and XLC support format attributes */ +#if defined(__GNUC__) || defined(__IBMC__) +#define pg_attribute_format_arg(a) __attribute__((format_arg(a))) +#define pg_attribute_printf(f,a) __attribute__((format(PG_PRINTF_ATTRIBUTE, f, a))) +#else +#define pg_attribute_format_arg(a) +#define pg_attribute_printf(f,a) +#endif + +#endif + +#include "libpq-fe.h" +#include "postgres_fe.h" +#include "pqexpbuffer.h" + +#include "getopt_long.h" + +#include "miscadmin.h" + +#include "access/timeline.h" +#include "access/xlog_internal.h" +#include "catalog/pg_control.h" + +#include "spock_fe.h" + +#define MAX_APPLY_DELAY 86400 + +typedef struct RemoteInfo { + Oid nodeid; + char *node_name; + char *sysid; + char *dbname; + char *replication_sets; +} RemoteInfo; + +typedef enum { + VERBOSITY_NORMAL, + VERBOSITY_VERBOSE, + VERBOSITY_DEBUG +} VerbosityLevelEnum; + +static char *argv0 = NULL; +static const char *progname; +static char *data_dir = NULL; +static char pid_file[MAXPGPATH]; +static time_t start_time; +static VerbosityLevelEnum verbosity = VERBOSITY_NORMAL; + +/* defined as static so that die() can close them */ +static PGconn *subscriber_conn = NULL; +static PGconn *provider_conn = NULL; + +static void signal_handler(int sig); +static void usage(void); +static void die(const char *fmt,...) +pg_attribute_printf(1, 2); +static void print_msg(VerbosityLevelEnum level, const char *fmt,...) +pg_attribute_printf(2, 3); + +static int run_pg_ctl(const char *arg); +static void run_basebackup(const char *provider_connstr, const char *data_dir, + const char *extra_basebackup_args); +static void wait_postmaster_connection(const char *connstr); +static void wait_primary_connection(const char *connstr); +static void wait_postmaster_shutdown(void); + +static char *validate_replication_set_input(char *replication_sets); + +static void remove_unwanted_data(PGconn *conn); +static void initialize_replication_origin(PGconn *conn, char *origin_name, char *remote_lsn); +static char *create_restore_point(PGconn *conn, char *restore_point_name); +static char *initialize_replication_slot(PGconn *conn, char *dbname, + char *provider_node_name, char *subscription_name, + bool drop_slot_if_exists); +static void spock_subscribe(PGconn *conn, char *subscriber_name, + char *subscriber_dsn, + char *provider_connstr, + char *replication_sets, + int apply_delay, + bool force_text_transfer); + +static RemoteInfo *get_remote_info(PGconn* conn); + +static bool extension_exists(PGconn *conn, const char *extname); +static void install_extension(PGconn *conn, const char *extname); + +static void initialize_data_dir(char *data_dir, char *connstr, + char *postgresql_conf, char *pg_hba_conf, + char *extra_basebackup_args); +static bool check_data_dir(char *data_dir, RemoteInfo *remoteinfo); + +static char *read_sysid(const char *data_dir); + +static void WriteRecoveryConf(PQExpBuffer contents); +static void CopyConfFile(char *fromfile, char *tofile, bool append); + +static char *get_connstr_dbname(char *connstr); +static char *get_connstr(char *connstr, char *dbname); +static char *PQconninfoParamsToConnstr(const char *const * keywords, const char *const * values); +static void appendPQExpBufferConnstrValue(PQExpBuffer buf, const char *str); + +static bool file_exists(const char *path); +static bool is_pg_dir(const char *path); +static void copy_file(char *fromfile, char *tofile, bool append); +static char *find_other_exec_or_die(const char *argv0, const char *target); +static bool postmaster_is_alive(pid_t pid); +static long get_pgpid(void); +static char **get_database_list(char *databases, int *n_databases); +static char *generate_restore_point_name(void); + +static PGconn * +connectdb(const char *connstr) +{ + PGconn *conn; + + conn = PQconnectdb(connstr); + if (PQstatus(conn) != CONNECTION_OK) + die(_("Connection to database failed: %s, connection string was: %s\n"), PQerrorMessage(conn), connstr); + + return conn; +} + +void signal_handler(int sig) +{ + if (sig == SIGINT) + { + die(_("\nCanceling...\n")); + } +} + + +int +main(int argc, char **argv) +{ + int i; + int c; + PQExpBuffer recoveryconfcontents = createPQExpBuffer(); + RemoteInfo *remote_info; + char *remote_lsn; + bool stop = false; + bool drop_slot_if_exists = false; + int optindex; + char *subscriber_name = NULL; + char *base_sub_connstr = NULL; + char *base_prov_connstr = NULL; + char *replication_sets = NULL; + char *databases = NULL; + char *postgresql_conf = NULL, + *pg_hba_conf = NULL, + *recovery_conf = NULL; + int apply_delay = 0; + bool force_text_transfer = false; + char **slot_names; + char *sub_connstr; + char *prov_connstr; + char **database_list = { NULL }; + int n_databases = 1; + int dbnum; + bool use_existing_data_dir = false; + int pg_ctl_ret, + logfd; + char *restore_point_name = NULL; + char *extra_basebackup_args = NULL; + + static struct option long_options[] = { + {"subscriber-name", required_argument, NULL, 'n'}, + {"pgdata", required_argument, NULL, 'D'}, + {"provider-dsn", required_argument, NULL, 1}, + {"subscriber-dsn", required_argument, NULL, 2}, + {"replication-sets", required_argument, NULL, 3}, + {"postgresql-conf", required_argument, NULL, 4}, + {"hba-conf", required_argument, NULL, 5}, + {"recovery-conf", required_argument, NULL, 6}, + {"stop", no_argument, NULL, 's'}, + {"drop-slot-if-exists", no_argument, NULL, 7}, + {"apply-delay", required_argument, NULL, 8}, + {"databases", required_argument, NULL, 9}, + {"extra-basebackup-args", required_argument, NULL, 10}, + {"text-types", no_argument, NULL, 11}, + {NULL, 0, NULL, 0} + }; + + argv0 = argv[0]; + progname = get_progname(argv[0]); + start_time = time(NULL); + signal(SIGINT, signal_handler); + + /* check for --help */ + if (argc > 1) + { + for (i = 1; i < argc; i++) + { + if (strcmp(argv[i], "--help") == 0 || strcmp(argv[i], "-?") == 0) + { + usage(); + exit(0); + } + } + } + + /* Option parsing and validation */ + while ((c = getopt_long(argc, argv, "D:n:sv", long_options, &optindex)) != -1) + { + switch (c) + { + case 'D': + data_dir = pg_strdup(optarg); + break; + case 'n': + subscriber_name = pg_strdup(optarg); + break; + case 1: + base_prov_connstr = pg_strdup(optarg); + break; + case 2: + base_sub_connstr = pg_strdup(optarg); + break; + case 3: + replication_sets = validate_replication_set_input(pg_strdup(optarg)); + break; + case 4: + { + postgresql_conf = pg_strdup(optarg); + if (postgresql_conf != NULL && !file_exists(postgresql_conf)) + die(_("The specified postgresql.conf file does not exist.")); + break; + } + case 5: + { + pg_hba_conf = pg_strdup(optarg); + if (pg_hba_conf != NULL && !file_exists(pg_hba_conf)) + die(_("The specified pg_hba.conf file does not exist.")); + break; + } + case 6: + { + recovery_conf = pg_strdup(optarg); + if (recovery_conf != NULL && !file_exists(recovery_conf)) + die(_("The specified recovery configuration file does not exist.")); + break; + } + case 'v': + verbosity++; + break; + case 's': + stop = true; + break; + case 7: + drop_slot_if_exists = true; + break; + case 8: + apply_delay = atoi(optarg); + break; + case 9: + databases = pg_strdup(optarg); + break; + case 10: + extra_basebackup_args = pg_strdup(optarg); + break; + case 11: + force_text_transfer = true; + break; + default: + fprintf(stderr, _("Unknown option\n")); + fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname); + exit(1); + } + } + + /* + * Sanity checks + */ + + if (data_dir == NULL) + { + fprintf(stderr, _("No data directory specified\n")); + fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname); + exit(1); + } + else if (subscriber_name == NULL) + { + fprintf(stderr, _("No subscriber name specified\n")); + fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname); + exit(1); + } + + if (!base_prov_connstr || !strlen(base_prov_connstr)) + die(_("Provider connection string must be specified.\n")); + if (!base_sub_connstr || !strlen(base_sub_connstr)) + die(_("Subscriber connection string must be specified.\n")); + + if (apply_delay < 0) + die(_("Apply delay cannot be negative.\n")); + + if (apply_delay > MAX_APPLY_DELAY) + die(_("Apply delay cannot be more than %d.\n"), MAX_APPLY_DELAY); + + if (!replication_sets || !strlen(replication_sets)) + replication_sets = "default,default_insert_only,ddl_sql"; + + /* Init random numbers used for slot suffixes, etc */ + srand(time(NULL)); + + /* Parse database list or connection string. */ + if (databases != NULL) + { + database_list = get_database_list(databases, &n_databases); + } + else + { + char *dbname = get_connstr_dbname(base_prov_connstr); + + if (!dbname) + die(_("Either provider connection string must contain database " + "name or --databases option must be specified.\n")); + + n_databases = 1; + database_list = palloc(n_databases * sizeof(char *)); + database_list[0] = dbname; + } + + slot_names = palloc(n_databases * sizeof(char *)); + + /* + * Check connection strings for validity before doing anything + * expensive. + */ + for (dbnum = 0; dbnum < n_databases; dbnum++) + { + char *db = database_list[dbnum]; + + prov_connstr = get_connstr(base_prov_connstr, db); + if (!prov_connstr || !strlen(prov_connstr)) + die(_("Provider connection string is not valid.\n")); + + sub_connstr = get_connstr(base_sub_connstr, db); + if (!sub_connstr || !strlen(sub_connstr)) + die(_("Subscriber connection string is not valid.\n")); + } + + /* + * Create log file where new postgres instance will log to while being + * initialized. + */ + logfd = open("spock_create_subscriber_postgres.log", O_CREAT | O_RDWR, + S_IRUSR | S_IWUSR); + if (logfd == -1) + { + die(_("Creating spock_create_subscriber_postgres.log failed: %s"), + strerror(errno)); + } + /* Safe to close() unchecked, we didn't write */ + (void) close(logfd); + + /* Let's start the real work... */ + print_msg(VERBOSITY_NORMAL, _("%s: starting ...\n"), progname); + + for (dbnum = 0; dbnum < n_databases; dbnum++) + { + char *db = database_list[dbnum]; + + prov_connstr = get_connstr(base_prov_connstr, db); + if (!prov_connstr || !strlen(prov_connstr)) + die(_("Provider connection string is not valid.\n")); + + /* Read the remote server indetification. */ + print_msg(VERBOSITY_NORMAL, + _("Getting information for database %s ...\n"), db); + provider_conn = connectdb(prov_connstr); + remote_info = get_remote_info(provider_conn); + + /* only need to do this piece once */ + + if (dbnum == 0) + { + use_existing_data_dir = check_data_dir(data_dir, remote_info); + + if (use_existing_data_dir && + strcmp(remote_info->sysid, read_sysid(data_dir)) != 0) + die(_("Subscriber data directory is not basebackup of remote node.\n")); + } + + /* + * Create replication slots on remote node. + */ + print_msg(VERBOSITY_NORMAL, + _("Creating replication slot in database %s ...\n"), db); + slot_names[dbnum] = initialize_replication_slot(provider_conn, + remote_info->dbname, + remote_info->node_name, + subscriber_name, + drop_slot_if_exists); + PQfinish(provider_conn); + provider_conn = NULL; + } + + /* + * Create basebackup or use existing one + */ + prov_connstr = get_connstr(base_prov_connstr, database_list[0]); + sub_connstr = get_connstr(base_sub_connstr, database_list[0]); + + initialize_data_dir(data_dir, + use_existing_data_dir ? NULL : prov_connstr, + postgresql_conf, pg_hba_conf, + extra_basebackup_args); + snprintf(pid_file, MAXPGPATH, "%s/postmaster.pid", data_dir); + + restore_point_name = generate_restore_point_name(); + + print_msg(VERBOSITY_NORMAL, _("Creating restore point \"%s\" on remote node ...\n"), + restore_point_name); + provider_conn = connectdb(prov_connstr); + remote_lsn = create_restore_point(provider_conn, restore_point_name); + PQfinish(provider_conn); + provider_conn = NULL; + + /* + * Get subscriber db to consistent state (for lsn after slot creation). + */ + print_msg(VERBOSITY_NORMAL, + _("Bringing subscriber node to the restore point ...\n")); + if (recovery_conf) + { + CopyConfFile(recovery_conf, "postgresql.auto.conf", true); + } + else + { + appendPQExpBuffer(recoveryconfcontents, "primary_conninfo = '%s'\n", + escape_single_quotes_ascii(prov_connstr)); + } + appendPQExpBuffer(recoveryconfcontents, "recovery_target_name = '%s'\n", restore_point_name); + appendPQExpBuffer(recoveryconfcontents, "recovery_target_inclusive = true\n"); + appendPQExpBuffer(recoveryconfcontents, "recovery_target_action = promote\n"); + WriteRecoveryConf(recoveryconfcontents); + + free(restore_point_name); + restore_point_name = NULL; + + /* + * Start subscriber node with spock disabled, and wait until it starts + * accepting connections which means it has caught up to the restore point. + */ + pg_ctl_ret = run_pg_ctl("start -l \"spock_create_subscriber_postgres.log\" -o \"-c shared_preload_libraries=''\""); + if (pg_ctl_ret != 0) + die(_("Postgres startup for restore point catchup failed with %d. See spock_create_subscriber_postgres.log."), pg_ctl_ret); + + wait_primary_connection(sub_connstr); + + /* + * Clean any per-node data that were copied by pg_basebackup. + */ + print_msg(VERBOSITY_VERBOSE, + _("Removing old spock configuration ...\n")); + + for (dbnum = 0; dbnum < n_databases; dbnum++) + { + char *db = database_list[dbnum]; + + sub_connstr = get_connstr(base_sub_connstr, db); + + if (!sub_connstr || !strlen(sub_connstr)) + die(_("Subscriber connection string is not valid.\n")); + + subscriber_conn = connectdb(sub_connstr); + remove_unwanted_data(subscriber_conn); + PQfinish(subscriber_conn); + subscriber_conn = NULL; + } + + /* Stop Postgres so we can reset system id and start it with spock loaded. */ + pg_ctl_ret = run_pg_ctl("stop"); + if (pg_ctl_ret != 0) + die(_("Postgres stop after restore point catchup failed with %d. See spock_create_subscriber_postgres.log."), pg_ctl_ret); + wait_postmaster_shutdown(); + + /* + * Start the node again, now with spock active so that we can start the + * logical replication. This is final start, so don't log to to special log + * file anymore. + */ + print_msg(VERBOSITY_NORMAL, + _("Initializing spock on the subscriber node:\n")); + + pg_ctl_ret = run_pg_ctl("start"); + if (pg_ctl_ret != 0) + die(_("Postgres restart with spock enabled failed with %d."), pg_ctl_ret); + wait_postmaster_connection(base_sub_connstr); + + for (dbnum = 0; dbnum < n_databases; dbnum++) + { + char *db = database_list[dbnum]; + + sub_connstr = get_connstr(base_sub_connstr, db); + prov_connstr = get_connstr(base_prov_connstr, db); + + subscriber_conn = connectdb(sub_connstr); + + /* Create the extension. */ + print_msg(VERBOSITY_VERBOSE, + _("Creating spock extension for database %s...\n"), db); + if (PQserverVersion(subscriber_conn) < 90500) + install_extension(subscriber_conn, "spock_origin"); + install_extension(subscriber_conn, "spock"); + + /* + * Create the identifier which is setup with the position to which we + * already caught up using physical replication. + */ + print_msg(VERBOSITY_VERBOSE, + _("Creating replication origin for database %s...\n"), db); + initialize_replication_origin(subscriber_conn, slot_names[dbnum], remote_lsn); + + /* + * And finally add the node to the cluster. + */ + print_msg(VERBOSITY_NORMAL, _("Creating subscriber %s for database %s...\n"), + subscriber_name, db); + print_msg(VERBOSITY_VERBOSE, _("Replication sets: %s\n"), replication_sets); + + spock_subscribe(subscriber_conn, subscriber_name, sub_connstr, + prov_connstr, replication_sets, apply_delay, + force_text_transfer); + + PQfinish(subscriber_conn); + subscriber_conn = NULL; + } + + /* If user does not want the node to be running at the end, stop it. */ + if (stop) + { + print_msg(VERBOSITY_NORMAL, _("Stopping the subscriber node ...\n")); + pg_ctl_ret = run_pg_ctl("stop"); + if (pg_ctl_ret != 0) + die(_("Stopping postgres after successful subscribtion failed with %d."), pg_ctl_ret); + wait_postmaster_shutdown(); + } + + print_msg(VERBOSITY_NORMAL, _("All done\n")); + + return 0; +} + + +/* + * Print help. + */ +static void +usage(void) +{ + printf(_("%s create new spock subscriber from basebackup of provider.\n\n"), progname); + printf(_("Usage:\n")); + printf(_(" %s [OPTION]...\n"), progname); + printf(_("\nGeneral options:\n")); + printf(_(" -D, --pgdata=DIRECTORY data directory to be used for new node,\n")); + printf(_(" can be either empty/non-existing directory,\n")); + printf(_(" or directory populated using\n")); + printf(_(" pg_basebackup -X stream command\n")); + printf(_(" --databases optional list of databases to replicate\n")); + printf(_(" -n, --subscriber-name=NAME name of the newly created subscriber\n")); + printf(_(" --subscriber-dsn=CONNSTR connection string to the newly created subscriber\n")); + printf(_(" --provider-dsn=CONNSTR connection string to the provider\n")); + printf(_(" --replication-sets=SETS comma separated list of replication set names\n")); + printf(_(" --apply-delay=DELAY apply delay in seconds (by default 0)\n")); + printf(_(" --drop-slot-if-exists drop replication slot of conflicting name\n")); + printf(_(" -s, --stop stop the server once the initialization is done\n")); + printf(_(" -v increase logging verbosity\n")); + printf(_(" --extra-basebackup-args additional arguments to pass to pg_basebackup.\n")); + printf(_(" Safe options: -T, -c, --xlogdir/--waldir\n")); + printf(_("\nConfiguration files override:\n")); + printf(_(" --hba-conf path to the new pg_hba.conf\n")); + printf(_(" --postgresql-conf path to the new postgresql.conf\n")); + printf(_(" --recovery-conf path to the template recovery configuration\n")); +} + +/* + * Print error and exit. + */ +static void +die(const char *fmt,...) +{ + va_list argptr; + va_start(argptr, fmt); + vfprintf(stderr, fmt, argptr); + va_end(argptr); + + if (subscriber_conn) + PQfinish(subscriber_conn); + if (provider_conn) + PQfinish(provider_conn); + + if (get_pgpid()) + { + if (!run_pg_ctl("stop -s")) + { + fprintf(stderr, _("WARNING: postgres seems to be running, but could not be stopped\n")); + } + } + + exit(1); +} + +/* + * Print message to stdout and flush + */ +static void +print_msg(VerbosityLevelEnum level, const char *fmt,...) +{ + if (verbosity >= level) + { + va_list argptr; + va_start(argptr, fmt); + vfprintf(stdout, fmt, argptr); + va_end(argptr); + fflush(stdout); + } +} + + +/* + * Start pg_ctl with given argument(s) - used to start/stop postgres + * + * Returns the exit code reported by pg_ctl. If pg_ctl exits due to a + * signal this call will die and not return. + */ +static int +run_pg_ctl(const char *arg) +{ + int ret; + PQExpBuffer cmd = createPQExpBuffer(); + char *exec_path = find_other_exec_or_die(argv0, "pg_ctl"); + + appendPQExpBuffer(cmd, "%s %s -D \"%s\"", exec_path, arg, data_dir); + + /* Run pg_ctl in silent mode unless we run in debug mode. */ + if (verbosity < VERBOSITY_DEBUG) + appendPQExpBuffer(cmd, " -s"); + + print_msg(VERBOSITY_DEBUG, _("Running pg_ctl: %s.\n"), cmd->data); + ret = system(cmd->data); + + destroyPQExpBuffer(cmd); + + if (WIFEXITED(ret)) + return WEXITSTATUS(ret); + else if (WIFSIGNALED(ret)) + die(_("pg_ctl exited with signal %d"), WTERMSIG(ret)); + else + die(_("pg_ctl exited for an unknown reason (system() returned %d)"), ret); + + return -1; +} + + +/* + * Run pg_basebackup to create the copy of the origin node. + */ +static void +run_basebackup(const char *provider_connstr, const char *data_dir, + const char *extra_basebackup_args) +{ + int ret; + PQExpBuffer cmd = createPQExpBuffer(); + char *exec_path = find_other_exec_or_die(argv0, "pg_basebackup"); + + appendPQExpBuffer(cmd, "%s -D \"%s\" -d \"%s\" -X s -P", exec_path, data_dir, provider_connstr); + + /* Run pg_basebackup in verbose mode if we are running in verbose mode. */ + if (verbosity >= VERBOSITY_VERBOSE) + appendPQExpBuffer(cmd, " -v"); + + if (extra_basebackup_args != NULL) + appendPQExpBuffer(cmd, "%s", extra_basebackup_args); + + print_msg(VERBOSITY_DEBUG, _("Running pg_basebackup: %s.\n"), cmd->data); + ret = system(cmd->data); + + destroyPQExpBuffer(cmd); + + if (WIFEXITED(ret) && WEXITSTATUS(ret) == 0) + return; + if (WIFEXITED(ret)) + die(_("pg_basebackup failed with exit status %d, cannot continue.\n"), WEXITSTATUS(ret)); + else if (WIFSIGNALED(ret)) + die(_("pg_basebackup exited with signal %d, cannot continue"), WTERMSIG(ret)); + else + die(_("pg_basebackup exited for an unknown reason (system() returned %d)"), ret); +} + +/* + * Init the datadir + * + * This function can either ensure provided datadir is a postgres datadir, + * or create it using pg_basebackup. + * + * In any case, new postresql.conf and pg_hba.conf will be copied to the + * datadir if they are provided. + */ +static void +initialize_data_dir(char *data_dir, char *connstr, + char *postgresql_conf, char *pg_hba_conf, + char *extra_basebackup_args) +{ + if (connstr) + { + print_msg(VERBOSITY_NORMAL, + _("Creating base backup of the remote node...\n")); + run_basebackup(connstr, data_dir, extra_basebackup_args); + } + + if (postgresql_conf) + CopyConfFile(postgresql_conf, "postgresql.conf", false); + if (pg_hba_conf) + CopyConfFile(pg_hba_conf, "pg_hba.conf", false); +} + +/* + * This function checks if provided datadir is clone of the remote node + * described by the remote info, or if it's emtpy directory that can be used + * as new datadir. + */ +static bool +check_data_dir(char *data_dir, RemoteInfo *remoteinfo) +{ + /* Run basebackup as needed. */ + switch (pg_check_dir(data_dir)) + { + case 0: /* Does not exist */ + case 1: /* Exists, empty */ + return false; + case 2: + case 3: /* Exists, not empty */ + case 4: + { + if (!is_pg_dir(data_dir)) + die(_("Directory \"%s\" exists but is not valid postgres data directory.\n"), + data_dir); + return true; + } + case -1: /* Access problem */ + die(_("Could not access directory \"%s\": %s.\n"), + data_dir, strerror(errno)); + } + + /* Unreachable */ + die(_("Unexpected result from pg_check_dir() call")); + return false; +} + +/* + * Initialize replication slots + */ +static char * +initialize_replication_slot(PGconn *conn, char *dbname, + char *provider_node_name, char *subscription_name, + bool drop_slot_if_exists) +{ + PQExpBufferData query; + char *slot_name; + PGresult *res; + + /* Generate the slot name. */ + initPQExpBuffer(&query); + printfPQExpBuffer(&query, + "SELECT spock.spock_gen_slot_name(%s, %s, %s)", + PQescapeLiteral(conn, dbname, strlen(dbname)), + PQescapeLiteral(conn, provider_node_name, + strlen(provider_node_name)), + PQescapeLiteral(conn, subscription_name, + strlen(subscription_name))); + + res = PQexec(conn, query.data); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + die(_("Could generate slot name: %s"), PQerrorMessage(conn)); + + slot_name = pstrdup(PQgetvalue(res, 0, 0)); + + PQclear(res); + resetPQExpBuffer(&query); + + /* Check if the current slot exists. */ + printfPQExpBuffer(&query, + "SELECT 1 FROM pg_catalog.pg_replication_slots WHERE slot_name = %s", + PQescapeLiteral(conn, slot_name, strlen(slot_name))); + + res = PQexec(conn, query.data); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + die(_("Could not fetch existing slot information: %s"), PQerrorMessage(conn)); + + /* Drop the existing slot when asked for it or error if it already exists. */ + if (PQntuples(res) > 0) + { + PQclear(res); + resetPQExpBuffer(&query); + + if (!drop_slot_if_exists) + die(_("Slot %s already exists, drop it or use --drop-slot-if-exists to drop it automatically.\n"), + slot_name); + + print_msg(VERBOSITY_VERBOSE, + _("Droping existing slot %s ...\n"), slot_name); + + printfPQExpBuffer(&query, + "SELECT pg_catalog.pg_drop_replication_slot(%s)", + PQescapeLiteral(conn, slot_name, strlen(slot_name))); + + res = PQexec(conn, query.data); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + die(_("Could not drop existing slot %s: %s"), slot_name, + PQerrorMessage(conn)); + } + + PQclear(res); + resetPQExpBuffer(&query); + + /* And finally, create the slot. */ + appendPQExpBuffer(&query, "SELECT pg_create_logical_replication_slot(%s, '%s');", + PQescapeLiteral(conn, slot_name, strlen(slot_name)), + "spock_output"); + + res = PQexec(conn, query.data); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + die(_("Could not create replication slot, status %s: %s\n"), + PQresStatus(PQresultStatus(res)), PQresultErrorMessage(res)); + } + + PQclear(res); + termPQExpBuffer(&query); + + return slot_name; +} + +/* + * Read replication info about remote connection + * + * TODO: unify with spock_remote_node_info in spock_rpc + */ +static RemoteInfo * +get_remote_info(PGconn* conn) +{ + RemoteInfo *ri = (RemoteInfo *)pg_malloc0(sizeof(RemoteInfo)); + PGresult *res; + + if (!extension_exists(conn, "spock")) + die(_("The remote node is not configured as a spock provider.\n")); + + res = PQexec(conn, "SELECT node_id, node_name, sysid, dbname, replication_sets FROM spock.node_info()"); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + die(_("could not fetch remote node info: %s\n"), PQerrorMessage(conn)); + + /* No nodes found? */ + if (PQntuples(res) == 0) + die(_("The remote database is not configured as a spock node.\n")); + + if (PQntuples(res) > 1) + die(_("The remote database has multiple nodes configured. That is not supported with current version of spock.\n")); + +#define atooid(x) ((Oid) strtoul((x), NULL, 10)) + + ri->nodeid = atooid(PQgetvalue(res, 0, 0)); + ri->node_name = pstrdup(PQgetvalue(res, 0, 1)); + ri->sysid = pstrdup(PQgetvalue(res, 0, 2)); + ri->dbname = pstrdup(PQgetvalue(res, 0, 3)); + ri->replication_sets = pstrdup(PQgetvalue(res, 0, 4)); + + PQclear(res); + + return ri; +} + +/* + * Check if extension exists. + */ +static bool +extension_exists(PGconn *conn, const char *extname) +{ + PQExpBuffer query = createPQExpBuffer(); + PGresult *res; + bool ret; + + printfPQExpBuffer(query, "SELECT 1 FROM pg_catalog.pg_extension WHERE extname = %s;", + PQescapeLiteral(conn, extname, strlen(extname))); + res = PQexec(conn, query->data); + + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + PQclear(res); + die(_("Could not read extension info: %s\n"), PQerrorMessage(conn)); + } + + ret = PQntuples(res) == 1; + + PQclear(res); + destroyPQExpBuffer(query); + + return ret; +} + +/* + * Create extension. + */ +static void +install_extension(PGconn *conn, const char *extname) +{ + PQExpBuffer query = createPQExpBuffer(); + PGresult *res; + + printfPQExpBuffer(query, "CREATE EXTENSION IF NOT EXISTS %s;", + PQescapeIdentifier(conn, extname, strlen(extname))); + res = PQexec(conn, query->data); + + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + PQclear(res); + die(_("Could not install %s extension: %s\n"), extname, PQerrorMessage(conn)); + } + + PQclear(res); + destroyPQExpBuffer(query); +} + +/* + * Clean all the data that was copied from remote node but we don't + * want it here (currently shared security labels and replication identifiers). + */ +static void +remove_unwanted_data(PGconn *conn) +{ + PGresult *res; + + /* + * Remove replication identifiers (9.4 will get them removed by dropping + * the extension later as we emulate them there). + */ + if (PQserverVersion(conn) >= 90500) + { + res = PQexec(conn, "SELECT pg_replication_origin_drop(external_id) FROM pg_replication_origin_status;"); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + PQclear(res); + die(_("Could not remove existing replication origins: %s\n"), PQerrorMessage(conn)); + } + PQclear(res); + } + + res = PQexec(conn, "DROP EXTENSION spock CASCADE;"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + die(_("Could not clean the spock extension, status %s: %s\n"), + PQresStatus(PQresultStatus(res)), PQresultErrorMessage(res)); + } + PQclear(res); +} + +/* + * Initialize new remote identifier to specific position. + */ +static void +initialize_replication_origin(PGconn *conn, char *origin_name, char *remote_lsn) +{ + PGresult *res; + PQExpBuffer query = createPQExpBuffer(); + + if (PQserverVersion(conn) >= 90500) + { + printfPQExpBuffer(query, "SELECT pg_replication_origin_create(%s)", + PQescapeLiteral(conn, origin_name, strlen(origin_name))); + + res = PQexec(conn, query->data); + + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + die(_("Could not create replication origin \"%s\": status %s: %s\n"), + query->data, + PQresStatus(PQresultStatus(res)), PQresultErrorMessage(res)); + } + PQclear(res); + + if (remote_lsn) + { + printfPQExpBuffer(query, "SELECT pg_replication_origin_advance(%s, '%s')", + PQescapeLiteral(conn, origin_name, strlen(origin_name)), + remote_lsn); + + res = PQexec(conn, query->data); + + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + die(_("Could not advance replication origin \"%s\": status %s: %s\n"), + query->data, + PQresStatus(PQresultStatus(res)), PQresultErrorMessage(res)); + } + PQclear(res); + } + } + else + { + printfPQExpBuffer(query, "INSERT INTO spock_origin.replication_origin (roident, roname, roremote_lsn) SELECT COALESCE(MAX(roident::int), 0) + 1, %s, %s FROM spock_origin.replication_origin", + PQescapeLiteral(conn, origin_name, strlen(origin_name)), + remote_lsn ? PQescapeLiteral(conn, remote_lsn, strlen(remote_lsn)) : "0"); + + res = PQexec(conn, query->data); + + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + die(_("Could not create replication origin \"%s\": status %s: %s\n"), + query->data, + PQresStatus(PQresultStatus(res)), PQresultErrorMessage(res)); + } + PQclear(res); + } + + destroyPQExpBuffer(query); +} + + +/* + * Create remote restore point which will be used to get into synchronized + * state through physical replay. + */ +static char * +create_restore_point(PGconn *conn, char *restore_point_name) +{ + PQExpBuffer query = createPQExpBuffer(); + PGresult *res; + char *remote_lsn = NULL; + + printfPQExpBuffer(query, "SELECT pg_create_restore_point('%s')", restore_point_name); + res = PQexec(conn, query->data); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + die(_("Could not create restore point, status %s: %s\n"), + PQresStatus(PQresultStatus(res)), PQresultErrorMessage(res)); + } + remote_lsn = pstrdup(PQgetvalue(res, 0, 0)); + + PQclear(res); + destroyPQExpBuffer(query); + + return remote_lsn; +} + +static void +spock_subscribe(PGconn *conn, char *subscriber_name, char *subscriber_dsn, + char *provider_dsn, char *replication_sets, + int apply_delay, bool force_text_transfer) +{ + PQExpBufferData query; + PQExpBufferData repsets; + PGresult *res; + + initPQExpBuffer(&query); + printfPQExpBuffer(&query, + "SELECT spock.node_create(node_name := %s, dsn := %s);", + PQescapeLiteral(conn, subscriber_name, strlen(subscriber_name)), + PQescapeLiteral(conn, subscriber_dsn, strlen(subscriber_dsn))); + + res = PQexec(conn, query.data); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + die(_("Could not create local node, status %s: %s\n"), + PQresStatus(PQresultStatus(res)), PQresultErrorMessage(res)); + } + PQclear(res); + + resetPQExpBuffer(&query); + initPQExpBuffer(&repsets); + + printfPQExpBuffer(&repsets, "{%s}", replication_sets); + printfPQExpBuffer(&query, + "SELECT spock.sub_create(" + "subscription_name := %s, provider_dsn := %s, " + "replication_sets := %s, " + "apply_delay := '%d seconds'::interval, " + "synchronize_structure := false, " + "synchronize_data := false, " + "force_text_transfer := '%s');", + PQescapeLiteral(conn, subscriber_name, strlen(subscriber_name)), + PQescapeLiteral(conn, provider_dsn, strlen(provider_dsn)), + PQescapeLiteral(conn, repsets.data, repsets.len), + apply_delay, (force_text_transfer ? "t" : "f")); + + res = PQexec(conn, query.data); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + die(_("Could not create subscription, status %s: %s\n"), + PQresStatus(PQresultStatus(res)), PQresultErrorMessage(res)); + } + PQclear(res); + + /* TODO */ + res = PQexec(conn, "UPDATE spock.local_sync_status SET sync_status = 'r'"); + if (PQresultStatus(res) != PGRES_COMMAND_OK) + { + die(_("Could not update subscription, status %s: %s\n"), + PQresStatus(PQresultStatus(res)), PQresultErrorMessage(res)); + } + + PQclear(res); + + termPQExpBuffer(&repsets); + termPQExpBuffer(&query); +} + + +/* + * Validates input of the replication sets and returns normalized data. + */ +static char * +validate_replication_set_input(char *replication_sets) +{ + char *name; + PQExpBuffer retbuf = createPQExpBuffer(); + char *ret; + bool first = true; + + if (!replication_sets) + return NULL; + + name = strtok(replication_sets, " ,"); + while (name != NULL) + { + const char *cp; + + if (strlen(name) == 0) + die(_("Replication set name \"%s\" is too short\n"), name); + + if (strlen(name) > NAMEDATALEN) + die(_("Replication set name \"%s\" is too long\n"), name); + + for (cp = name; *cp; cp++) + { + if (!((*cp >= 'a' && *cp <= 'z') + || (*cp >= '0' && *cp <= '9') + || (*cp == '_') + || (*cp == '-'))) + { + die(_("Replication set name \"%s\" contains invalid character\n"), + name); + } + } + + if (first) + first = false; + else + appendPQExpBufferStr(retbuf, ", "); + appendPQExpBufferStr(retbuf, name); + + name = strtok(NULL, " ,"); + } + + ret = pg_strdup(retbuf->data); + destroyPQExpBuffer(retbuf); + + return ret; +} + +static char * +get_connstr_dbname(char *connstr) +{ + PQconninfoOption *conn_opts = NULL; + PQconninfoOption *conn_opt; + char *err_msg = NULL; + char *ret = NULL; + + conn_opts = PQconninfoParse(connstr, &err_msg); + if (conn_opts == NULL) + { + die(_("Invalid connection string: %s\n"), err_msg); + } + + for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++) + { + if (strcmp(conn_opt->keyword, "dbname") == 0) + { + ret = pstrdup(conn_opt->val); + break; + } + } + + PQconninfoFree(conn_opts); + + return ret; +} + + +/* + * Build connection string from individual parameter. + * + * dbname can be specified in connstr parameter + */ +static char * +get_connstr(char *connstr, char *dbname) +{ + char *ret; + int argcount = 4; /* dbname, host, user, port */ + int i; + const char **keywords; + const char **values; + PQconninfoOption *conn_opts = NULL; + PQconninfoOption *conn_opt; + char *err_msg = NULL; + + /* + * Merge the connection info inputs given in form of connection string + * and options + */ + i = 0; + if (connstr && + (strncmp(connstr, "postgresql://", 13) == 0 || + strncmp(connstr, "postgres://", 11) == 0 || + strchr(connstr, '=') != NULL)) + { + conn_opts = PQconninfoParse(connstr, &err_msg); + if (conn_opts == NULL) + { + die(_("Invalid connection string: %s\n"), err_msg); + } + + for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++) + { + if (conn_opt->val != NULL && conn_opt->val[0] != '\0') + argcount++; + } + + keywords = pg_malloc0((argcount + 1) * sizeof(*keywords)); + values = pg_malloc0((argcount + 1) * sizeof(*values)); + + for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++) + { + /* If db* parameters were provided, we'll fill them later. */ + if (dbname && strcmp(conn_opt->keyword, "dbname") == 0) + continue; + + if (conn_opt->val != NULL && conn_opt->val[0] != '\0') + { + keywords[i] = conn_opt->keyword; + values[i] = conn_opt->val; + i++; + } + } + } + else + { + keywords = pg_malloc0((argcount + 1) * sizeof(*keywords)); + values = pg_malloc0((argcount + 1) * sizeof(*values)); + + /* + * If connstr was provided but it's not in connection string format and + * the dbname wasn't provided then connstr is actually dbname. + */ + if (connstr && !dbname) + dbname = connstr; + } + + if (dbname) + { + keywords[i] = "dbname"; + values[i] = dbname; + i++; + } + + ret = PQconninfoParamsToConnstr(keywords, values); + + /* Connection ok! */ + pg_free(values); + pg_free(keywords); + if (conn_opts) + PQconninfoFree(conn_opts); + + return ret; +} + + +/* + * Reads the pg_control file of the existing data dir. + */ +static char * +read_sysid(const char *data_dir) +{ + ControlFileData ControlFile; + int fd; + char ControlFilePath[MAXPGPATH]; + char *res = (char *) pg_malloc0(33); + + snprintf(ControlFilePath, MAXPGPATH, "%s/global/pg_control", data_dir); + + if ((fd = open(ControlFilePath, O_RDONLY | PG_BINARY, 0)) == -1) + die(_("%s: could not open file \"%s\" for reading: %s\n"), + progname, ControlFilePath, strerror(errno)); + + if (read(fd, &ControlFile, sizeof(ControlFileData)) != sizeof(ControlFileData)) + die(_("%s: could not read file \"%s\": %s\n"), + progname, ControlFilePath, strerror(errno)); + + close(fd); + + snprintf(res, 33, UINT64_FORMAT, ControlFile.system_identifier); + return res; +} + +/* + * Write contents of recovery.conf or postgresql.auto.conf + */ +static void +WriteRecoveryConf(PQExpBuffer contents) +{ + char filename[MAXPGPATH]; + FILE *cf; + + sprintf(filename, "%s/postgresql.auto.conf", data_dir); + + cf = fopen(filename, "a"); + if (cf == NULL) + { + die(_("%s: could not create file \"%s\": %s\n"), progname, filename, strerror(errno)); + } + + if (fwrite(contents->data, contents->len, 1, cf) != 1) + { + die(_("%s: could not write to file \"%s\": %s\n"), + progname, filename, strerror(errno)); + } + + fclose(cf); + + { + sprintf(filename, "%s/standby.signal", data_dir); + cf = fopen(filename, "w"); + if (cf == NULL) + { + die(_("%s: could not create file \"%s\": %s\n"), progname, filename, strerror(errno)); + } + + fclose(cf); + } +} + +/* + * Copy file to data + */ +static void +CopyConfFile(char *fromfile, char *tofile, bool append) +{ + char filename[MAXPGPATH]; + + sprintf(filename, "%s/%s", data_dir, tofile); + + print_msg(VERBOSITY_DEBUG, _("Copying \"%s\" to \"%s\".\n"), + fromfile, filename); + copy_file(fromfile, filename, append); +} + + +/* + * Convert PQconninfoOption array into conninfo string + */ +static char * +PQconninfoParamsToConnstr(const char *const * keywords, const char *const * values) +{ + PQExpBuffer retbuf = createPQExpBuffer(); + char *ret; + int i = 0; + + for (i = 0; keywords[i] != NULL; i++) + { + if (i > 0) + appendPQExpBufferChar(retbuf, ' '); + appendPQExpBuffer(retbuf, "%s=", keywords[i]); + appendPQExpBufferConnstrValue(retbuf, values[i]); + } + + ret = pg_strdup(retbuf->data); + destroyPQExpBuffer(retbuf); + + return ret; +} + +/* + * Escape connection info value + */ +static void +appendPQExpBufferConnstrValue(PQExpBuffer buf, const char *str) +{ + const char *s; + bool needquotes; + + /* + * If the string consists entirely of plain ASCII characters, no need to + * quote it. This is quite conservative, but better safe than sorry. + */ + needquotes = false; + for (s = str; *s; s++) + { + if (!((*s >= 'a' && *s <= 'z') || (*s >= 'A' && *s <= 'Z') || + (*s >= '0' && *s <= '9') || *s == '_' || *s == '.')) + { + needquotes = true; + break; + } + } + + if (needquotes) + { + appendPQExpBufferChar(buf, '\''); + while (*str) + { + /* ' and \ must be escaped by to \' and \\ */ + if (*str == '\'' || *str == '\\') + appendPQExpBufferChar(buf, '\\'); + + appendPQExpBufferChar(buf, *str); + str++; + } + appendPQExpBufferChar(buf, '\''); + } + else + appendPQExpBufferStr(buf, str); +} + + +/* + * Find the pgport and try a connection + */ +static void +wait_postmaster_connection(const char *connstr) +{ + PGPing res; + long pmpid = 0; + + print_msg(VERBOSITY_VERBOSE, "Waiting for PostgreSQL to accept connections ..."); + + /* First wait for Postmaster to come up. */ + for (;;) + { + if ((pmpid = get_pgpid()) != 0 && + postmaster_is_alive((pid_t) pmpid)) + break; + + pg_usleep(1000000); /* 1 sec */ + print_msg(VERBOSITY_VERBOSE, "."); + } + + /* Now wait for Postmaster to either accept connections or die. */ + for (;;) + { + res = PQping(connstr); + if (res == PQPING_OK) + break; + else if (res == PQPING_NO_ATTEMPT) + break; + + /* + * Check if the process is still alive. This covers cases where the + * postmaster successfully created the pidfile but then crashed without + * removing it. + */ + if (!postmaster_is_alive((pid_t) pmpid)) + break; + + /* No response; wait */ + pg_usleep(1000000); /* 1 sec */ + print_msg(VERBOSITY_VERBOSE, "."); + } + + print_msg(VERBOSITY_VERBOSE, "\n"); +} + + +/* + * Wait for PostgreSQL to leave recovery/standby mode + */ +static void +wait_primary_connection(const char *connstr) +{ + bool ispri = false; + PGconn *conn = NULL; + PGresult *res; + + wait_postmaster_connection(connstr); + + print_msg(VERBOSITY_VERBOSE, "Waiting for PostgreSQL to become primary..."); + + while (!ispri) + { + if (!conn || PQstatus(conn) != CONNECTION_OK) + { + if (conn) + PQfinish(conn); + wait_postmaster_connection(connstr); + conn = connectdb(connstr); + } + + res = PQexec(conn, "SELECT pg_is_in_recovery()"); + if (PQresultStatus(res) == PGRES_TUPLES_OK && PQntuples(res) == 1 && *PQgetvalue(res, 0, 0) == 'f') + ispri = true; + else + { + pg_usleep(1000000); /* 1 sec */ + print_msg(VERBOSITY_VERBOSE, "."); + } + + PQclear(res); + } + + PQfinish(conn); + print_msg(VERBOSITY_VERBOSE, "\n"); +} + +/* + * Wait for postmaster to die + */ +static void +wait_postmaster_shutdown(void) +{ + long pid; + + print_msg(VERBOSITY_VERBOSE, "Waiting for PostgreSQL to shutdown ..."); + + for (;;) + { + if ((pid = get_pgpid()) != 0) + { + pg_usleep(1000000); /* 1 sec */ + print_msg(VERBOSITY_NORMAL, "."); + } + else + break; + } + + print_msg(VERBOSITY_VERBOSE, "\n"); +} + +static bool +file_exists(const char *path) +{ + struct stat statbuf; + + if (stat(path, &statbuf) != 0) + return false; + + return true; +} + +static bool +is_pg_dir(const char *path) +{ + struct stat statbuf; + char version_file[MAXPGPATH]; + + if (stat(path, &statbuf) != 0) + return false; + + snprintf(version_file, MAXPGPATH, "%s/PG_VERSION", data_dir); + if (stat(version_file, &statbuf) != 0 && errno == ENOENT) + { + return false; + } + + return true; +} + +/* + * copy one file + */ +static void +copy_file(char *fromfile, char *tofile, bool append) +{ + char *buffer; + int srcfd; + int dstfd; + int nbytes; + +#define COPY_BUF_SIZE (8 * BLCKSZ) + + buffer = malloc(COPY_BUF_SIZE); + + /* + * Open the files + */ + srcfd = open(fromfile, O_RDONLY | PG_BINARY, 0); + if (srcfd < 0) + die(_("could not open file \"%s\""), fromfile); + + dstfd = open(tofile, O_RDWR | O_CREAT | (append ? O_APPEND : O_TRUNC) | PG_BINARY, + S_IRUSR | S_IWUSR); + if (dstfd < 0) + die(_("could not create file \"%s\""), tofile); + + /* + * Do the data copying. + */ + for (;;) + { + nbytes = read(srcfd, buffer, COPY_BUF_SIZE); + if (nbytes < 0) + die(_("could not read file \"%s\""), fromfile); + if (nbytes == 0) + break; + errno = 0; + if ((int) write(dstfd, buffer, nbytes) != nbytes) + { + /* if write didn't set errno, assume problem is no disk space */ + if (errno == 0) + errno = ENOSPC; + die(_("could not write to file \"%s\""), tofile); + } + } + + if (close(dstfd)) + die(_("could not close file \"%s\""), tofile); + + /* we don't care about errors here */ + close(srcfd); + + free(buffer); +} + + +static char * +find_other_exec_or_die(const char *argv0, const char *target) +{ + int ret; + char *found_path; + uint32 bin_version; + + found_path = pg_malloc(MAXPGPATH); + + ret = find_other_exec_version(argv0, target, &bin_version, found_path); + + if (ret < 0) + { + char full_path[MAXPGPATH]; + + if (find_my_exec(argv0, full_path) < 0) + strlcpy(full_path, progname, sizeof(full_path)); + + if (ret == -1) + die(_("The program \"%s\" is needed by %s " + "but was not found in the\n" + "same directory as \"%s\".\n" + "Check your installation.\n"), + target, progname, full_path); + else + die(_("The program \"%s\" was found by \"%s\"\n" + "but was not the same version as %s.\n" + "Check your installation.\n"), + target, full_path, progname); + } + else + { + char full_path[MAXPGPATH]; + + if (find_my_exec(argv0, full_path) < 0) + strlcpy(full_path, progname, sizeof(full_path)); + + if (bin_version / 100 != PG_VERSION_NUM / 100) + die(_("The program \"%s\" was found by \"%s\"\n" + "but was not the same version as %s.\n" + "Check your installation.\n"), + target, full_path, progname); + + } + + return found_path; +} + +static bool +postmaster_is_alive(pid_t pid) +{ + /* + * Test to see if the process is still there. Note that we do not + * consider an EPERM failure to mean that the process is still there; + * EPERM must mean that the given PID belongs to some other userid, and + * considering the permissions on $PGDATA, that means it's not the + * postmaster we are after. + * + * Don't believe that our own PID or parent shell's PID is the postmaster, + * either. (Windows hasn't got getppid(), though.) + */ + if (pid == getpid()) + return false; +#ifndef WIN32 + if (pid == getppid()) + return false; +#endif + if (kill(pid, 0) == 0) + return true; + return false; +} + +static long +get_pgpid(void) +{ + FILE *pidf; + long pid; + + pidf = fopen(pid_file, "r"); + if (pidf == NULL) + { + return 0; + } + if (fscanf(pidf, "%ld", &pid) != 1) + { + return 0; + } + fclose(pidf); + return pid; +} + +static char ** +get_database_list(char *databases, int *n_databases) +{ + char *c; + char **result; + int num = 1; + for (c = databases; *c; c++ ) + if (*c == ',') + num++; + *n_databases = num; + result = palloc(num * sizeof(char *)); + num = 0; + /* clone the argument so we don't destroy it with strtok*/ + databases = pstrdup(databases); + c = strtok(databases, ","); + while (c != NULL) + { + result[num] = pstrdup(c); + num++; + c = strtok(NULL,","); + } + pfree(databases); + return result; +} + +static char * +generate_restore_point_name(void) +{ + char *rpn = malloc(NAMEDATALEN); + snprintf(rpn, NAMEDATALEN-1, "spock_create_subscriber_%lx", random()); + return rpn; +} From d688e21fff4e9a57c3554956e61250eabd09b698 Mon Sep 17 00:00:00 2001 From: Asif Rehman Date: Mon, 22 Jun 2026 14:06:38 +0500 Subject: [PATCH 2/7] spock_create_subscriber: drop legacy paths and extension usage --- .../spock_create_subscriber.c | 64 ++++++------------- 1 file changed, 20 insertions(+), 44 deletions(-) diff --git a/utils/spock_create_subscriber/spock_create_subscriber.c b/utils/spock_create_subscriber/spock_create_subscriber.c index 8147c68b0..fbf85fdfe 100644 --- a/utils/spock_create_subscriber/spock_create_subscriber.c +++ b/utils/spock_create_subscriber/spock_create_subscriber.c @@ -531,8 +531,6 @@ main(int argc, char **argv) /* Create the extension. */ print_msg(VERBOSITY_VERBOSE, _("Creating spock extension for database %s...\n"), db); - if (PQserverVersion(subscriber_conn) < 90500) - install_extension(subscriber_conn, "spock_origin"); install_extension(subscriber_conn, "spock"); /* @@ -965,16 +963,13 @@ remove_unwanted_data(PGconn *conn) * Remove replication identifiers (9.4 will get them removed by dropping * the extension later as we emulate them there). */ - if (PQserverVersion(conn) >= 90500) + res = PQexec(conn, "SELECT pg_replication_origin_drop(external_id) FROM pg_replication_origin_status;"); + if (PQresultStatus(res) != PGRES_TUPLES_OK) { - res = PQexec(conn, "SELECT pg_replication_origin_drop(external_id) FROM pg_replication_origin_status;"); - if (PQresultStatus(res) != PGRES_TUPLES_OK) - { - PQclear(res); - die(_("Could not remove existing replication origins: %s\n"), PQerrorMessage(conn)); - } PQclear(res); + die(_("Could not remove existing replication origins: %s\n"), PQerrorMessage(conn)); } + PQclear(res); res = PQexec(conn, "DROP EXTENSION spock CASCADE;"); if (PQresultStatus(res) != PGRES_COMMAND_OK) @@ -994,49 +989,30 @@ initialize_replication_origin(PGconn *conn, char *origin_name, char *remote_lsn) PGresult *res; PQExpBuffer query = createPQExpBuffer(); - if (PQserverVersion(conn) >= 90500) - { - printfPQExpBuffer(query, "SELECT pg_replication_origin_create(%s)", - PQescapeLiteral(conn, origin_name, strlen(origin_name))); - - res = PQexec(conn, query->data); + printfPQExpBuffer(query, "SELECT pg_replication_origin_create(%s)", + PQescapeLiteral(conn, origin_name, strlen(origin_name))); - if (PQresultStatus(res) != PGRES_TUPLES_OK) - { - die(_("Could not create replication origin \"%s\": status %s: %s\n"), - query->data, - PQresStatus(PQresultStatus(res)), PQresultErrorMessage(res)); - } - PQclear(res); - - if (remote_lsn) - { - printfPQExpBuffer(query, "SELECT pg_replication_origin_advance(%s, '%s')", - PQescapeLiteral(conn, origin_name, strlen(origin_name)), - remote_lsn); - - res = PQexec(conn, query->data); + res = PQexec(conn, query->data); - if (PQresultStatus(res) != PGRES_TUPLES_OK) - { - die(_("Could not advance replication origin \"%s\": status %s: %s\n"), - query->data, - PQresStatus(PQresultStatus(res)), PQresultErrorMessage(res)); - } - PQclear(res); - } + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + die(_("Could not create replication origin \"%s\": status %s: %s\n"), + query->data, + PQresStatus(PQresultStatus(res)), PQresultErrorMessage(res)); } - else + PQclear(res); + + if (remote_lsn) { - printfPQExpBuffer(query, "INSERT INTO spock_origin.replication_origin (roident, roname, roremote_lsn) SELECT COALESCE(MAX(roident::int), 0) + 1, %s, %s FROM spock_origin.replication_origin", - PQescapeLiteral(conn, origin_name, strlen(origin_name)), - remote_lsn ? PQescapeLiteral(conn, remote_lsn, strlen(remote_lsn)) : "0"); + printfPQExpBuffer(query, "SELECT pg_replication_origin_advance(%s, '%s')", + PQescapeLiteral(conn, origin_name, strlen(origin_name)), + remote_lsn); res = PQexec(conn, query->data); - if (PQresultStatus(res) != PGRES_COMMAND_OK) + if (PQresultStatus(res) != PGRES_TUPLES_OK) { - die(_("Could not create replication origin \"%s\": status %s: %s\n"), + die(_("Could not advance replication origin \"%s\": status %s: %s\n"), query->data, PQresStatus(PQresultStatus(res)), PQresultErrorMessage(res)); } From 16821b2f79947368a9dd52a57e02aa549660f391 Mon Sep 17 00:00:00 2001 From: Asif Rehman Date: Mon, 22 Jun 2026 14:06:38 +0500 Subject: [PATCH 3/7] Added usage document for spock_create_subscriber --- docs/creating_subscriber_nodes.md | 37 +++++++++++++++++++++++++++++++ 1 file changed, 37 insertions(+) create mode 100644 docs/creating_subscriber_nodes.md diff --git a/docs/creating_subscriber_nodes.md b/docs/creating_subscriber_nodes.md new file mode 100644 index 000000000..bc4622394 --- /dev/null +++ b/docs/creating_subscriber_nodes.md @@ -0,0 +1,37 @@ +## Creating a Subscriber Node with pg_basebackup + +Spock supports creating a subscriber node by cloning the provider with [`pg_basebackup`](https://www.postgresql.org/docs/current/app-pgbasebackup.html) and starting it as a Spock subscriber. Use the `spock_create_subscriber` utility (located in the `bin` directory of your pgEdge platform installation) to register the node. + +#### Synopsis: + + `spock_create_subscriber [OPTION]...` + +**Options** + +Specify the following options as needed. + +| Option | Description +|----------|------------- +| `-D`, `--pgdata=DIRECTORY` | The `data` directory to be used for new node. This can be either an empty/non-existing directory, or a directory populated using the `pg_basebackup -X stream` command. +| `--databases` | An optional list of databases to replicate. +| `-n`, `--subscriber-name=NAME` | The name of the newly created subscriber. +| `--subscriber-dsn=CONNSTR` | A connection string to the newly created subscriber. +| `--provider-dsn=CONNSTR` | A connection string to the provider. +| `--replication-sets=SETS` | A comma separated list of replication set names. +| `--apply-delay=DELAY` | The apply delay in seconds (by default 0). +| `--drop-slot-if-exists` | Drop replication slot of conflicting name. +| `-s`, `--stop` | Stop the server once the initialization is done. +| `-v` | Increase logging verbosity. +| `--extra-basebackup-args` | Additional arguments to pass to `pg_basebackup`. Safe options are: `-T`, `-c`, `--xlogdir`/`--waldir` + +**Configuration files overrides** + +You can use the following options to override the location of the configuration files. + +| Option | Description +|----------|------------- +|`--hba-conf` | path to the new `pg_hba.conf` +| `--postgresql-conf` | path to the new `postgresql.conf` +| `--recovery-conf` | path to the template recovery configuration + +Unlike `spock.sub_create`'s other data sync options, this method of cloning ignores replication sets and copies all tables on all databases. However, it's often much faster, especially over high-bandwidth connections. From 051450c57204ae094eb17474a9cf4f90ee7e2c25 Mon Sep 17 00:00:00 2001 From: Asif Rehman Date: Wed, 22 Jul 2026 10:34:35 +0500 Subject: [PATCH 4/7] spock_apply: advance forwarded origin using pre-created disabled subscription Any node that subscribes with forward_origins='all' receives transactions originally sourced from other peer nodes, forwarded through its immediate provider. Each forwarded transaction's ORIGIN wire message carries the original peer's Spock node OID, not the immediate provider's. A peer's transactions can reach this node forwarded this way well before -- or without -- a direct subscription to that peer ever being created here. If one has been (or later is) created, tracking the forwarded progress on its origin lets it start from the right position whenever it is enabled, instead of a full resync. A replication origin only exists where a subscription entry does, so this requires resolving the peer's node id to the local subscription (if any) that names it as provider. handle_origin() resolves the peer id to a local origin via track_forward_peer_origin(), excluding the "no origin" and direct-provider cases (the provider tags its own commits with an ORIGIN too). Resolution is cached in a hash table keyed by peer node id, not a single last-seen scalar, since one worker can forward transactions from many interleaved peers; a failed lookup is left uncached so a subscription created later is still picked up without a worker restart. get_node_subscriptions() can return more than one match, which is a hard error rather than an arbitrary pick. maybe_advance_forwarded_origin(), called from handle_commit(), advances the resolved local origin using XactLastCommitEnd -- this node's own commit LSN -- as the local position, not the provider's end_lsn, a different WAL space entirely. The advance is WAL-logged, making the peer's origin itself the durable resume position with no separate ledger. One narrow residual gap remains: the advance is a separate WAL record after the data commit, so a crash there re-delivers at most the last forwarded transaction (at-least-once, matching ordinary Spock apply); delta-apply idempotency under such re-delivery is a separate, tracked follow-up. t/015 redesigned to test this path: - n1->n2->n3 cascade with forward_origins='all' - disabled sub_n3_n1 pre-created on n3 before data arrives - verifies origin is advanced beyond 0/0 after n1 inserts propagate - verifies origin is stable after n2->n1 is disabled (gap detection) - verifies sub_enable('sub_n3_n1') starts the apply worker from the advanced position (not 0/0), so rows already received via the cascade are not re-sent, and new inserts arrive exactly once via the direct sub --- src/spock_apply.c | 284 ++++++++++++++------- tests/tap/t/015_forward_origin_advance.pl | 296 ++++++++++++++-------- 2 files changed, 381 insertions(+), 199 deletions(-) diff --git a/src/spock_apply.c b/src/spock_apply.c index a05dd86f1..26fec6fb7 100644 --- a/src/spock_apply.c +++ b/src/spock_apply.c @@ -57,7 +57,9 @@ #include "utils/builtins.h" #include "utils/acl.h" +#include "utils/catcache.h" #include "utils/fmgroids.h" +#include "utils/hsearch.h" #include "utils/jsonb.h" #include "utils/lsyscache.h" #include "utils/memutils.h" @@ -102,12 +104,24 @@ static TimeOffset apply_delay = 0; static TimestampTz required_commit_ts = 0; /* - * Cache for forwarded origin lookup. The remote_origin_id (Spock node ID) - * is consistent across the cluster, so we can use it as a cache key to - * avoid repeated slot name generation and origin lookups. + * Cache mapping a forwarding peer's Spock node id to the local RepOriginId + * of this node's own subscription to that peer, if any. Keyed by node id + * (not a single last-seen scalar) because one apply worker can forward + * transactions originally from many different peers, interleaved in + * arbitrary commit order -- a single-entry cache would be invalidated by + * nearly every transaction as soon as more than one peer is in play. Only + * peers that successfully resolved to a local subscription are entered; a + * peer with none is looked up again next time (see resolve_forward_peer_origin() + * callers), so a subscription created later is picked up without a worker + * restart. */ -static RepOriginId cached_forward_remote_id = InvalidRepOriginId; -static RepOriginId cached_forward_local_id = InvalidRepOriginId; +typedef struct ForwardOriginCacheEntry +{ + RepOriginId remote_node_id; /* hash key: peer's Spock node id */ + RepOriginId local_origin_id; /* this node's origin for that peer */ +} ForwardOriginCacheEntry; + +static HTAB *ForwardOriginCache = NULL; static Oid QueueRelid = InvalidOid; @@ -274,7 +288,10 @@ static void append_feedback_position(XLogRecPtr local_commit_lsn, static void get_feedback_position(XLogRecPtr *recvpos, XLogRecPtr *writepos, XLogRecPtr *flushpos, XLogRecPtr *max_recvpos); static void UpdateWorkerStats(XLogRecPtr last_received, XLogRecPtr last_inserted); -static void maybe_advance_forwarded_origin(XLogRecPtr end_lsn, bool xact_had_exception); +static void forward_origin_cache_init(void); +static RepOriginId resolve_forward_peer_origin(RepOriginId remote_origin_id); +static void track_forward_peer_origin(RepOriginId remote_origin_id); +static void maybe_advance_forwarded_origin(XLogRecPtr local_lsn, bool xact_had_exception); static ApplyReplayEntry *apply_replay_queue_next_entry(void); static bool apply_replay_queue_append_entry(ApplyReplayEntry **entry_p, StringInfo *msg_p); @@ -836,6 +853,7 @@ handle_commit(StringInfo s) XLogRecPtr end_lsn; TimestampTz commit_time; XLogRecPtr remote_insert_lsn; + XLogRecPtr local_commit_lsn = InvalidXLogRecPtr; errcallback_arg.action_name = "COMMIT"; xact_action_counter++; @@ -1026,6 +1044,7 @@ handle_commit(StringInfo s) flushpos = (SPKFlushPosition *) palloc(sizeof(SPKFlushPosition)); flushpos->local_end = XactLastCommitEnd; flushpos->remote_end = end_lsn; + local_commit_lsn = XactLastCommitEnd; dlist_push_tail(&lsn_mapping, &flushpos->node); MemoryContextSwitchTo(MessageContext); @@ -1050,10 +1069,12 @@ handle_commit(StringInfo s) /* * For forwarded transactions, advance the replication origin for the - * original source node. This is done outside the IsTransactionState() - * block because it starts its own transaction. + * original source node. The local LSN passed here must be OUR commit + * position (XactLastCommitEnd), not end_lsn — end_lsn is the provider's + * WAL position, a different WAL space entirely. For an empty forwarded + * transaction there is no local commit, so InvalidXLogRecPtr is passed. */ - maybe_advance_forwarded_origin(end_lsn, xact_had_exception); + maybe_advance_forwarded_origin(local_commit_lsn, xact_had_exception); /* Update the entry in the progress table. */ elog(DEBUG1, "SPOCK %s: updating progress table for node_id %d" \ @@ -1175,6 +1196,60 @@ handle_commit(StringInfo s) pgstat_report_activity(STATE_IDLE, NULL); } +static void +forward_origin_cache_init(void) +{ + HASHCTL ctl; + + if (CacheMemoryContext == NULL) + CreateCacheMemoryContext(); + + MemSet(&ctl, 0, sizeof(ctl)); + ctl.keysize = sizeof(RepOriginId); + ctl.entrysize = sizeof(ForwardOriginCacheEntry); + ctl.hcxt = CacheMemoryContext; + + ForwardOriginCache = hash_create("spock forwarded origin cache", + 16, &ctl, + HASH_ELEM | HASH_BLOBS | HASH_CONTEXT); +} + +/* + * Resolve the local subscription to a forwarding peer, if one exists. + * Called from handle_origin() with a transaction already open. + * + * A peer's transactions can reach this node forwarded through some other + * provider (cascade replication, or forward_origins) well before -- or + * without -- a direct subscription to that peer ever being created here. + * If one has been created, tracking the forwarded progress on its origin + * lets it start from the right position whenever it is enabled, instead of + * a full resync. get_node_subscriptions() finds it by the peer's Spock node + * id, returning InvalidRepOriginId when no such subscription exists (the + * ordinary case for a peer this node has no plans to subscribe to + * directly). + * + * get_node_subscriptions() returns a list rather than a single row, so more + * than one local subscription to the same peer is possible; picking one + * arbitrarily would advance the wrong one's origin. That is a hard error + * rather than a pick. + */ +static RepOriginId +resolve_forward_peer_origin(RepOriginId remote_origin_id) +{ + List *subs; + + subs = get_node_subscriptions((Oid) remote_origin_id, true); + if (subs == NIL) + return InvalidRepOriginId; + + if (list_length(subs) > 1) + elog(ERROR, "SPOCK %s: ambiguous forwarded origin: more than one " + "local subscription matches peer node (origin id %u)", + MySubscription->name, remote_origin_id); + + return replorigin_by_name(((SpockSubscription *) linitial(subs))->slot_name, true); +} + /* * Handle ORIGIN message. */ @@ -1211,6 +1286,70 @@ handle_origin(StringInfo s) */ remote_origin_id = spock_read_origin(s, &remote_origin_lsn, &remote_origin_name); replorigin_session_origin = remote_origin_id; + + track_forward_peer_origin(remote_origin_id); +} + +/* + * Ensure the forward-origin cache has an entry for remote_origin_id, if a + * local subscription to that peer exists. Called from handle_origin() for + * every ORIGIN message. + * + * An ORIGIN message is NOT exclusive to forwarded transactions: with + * forwarding output enabled the direct provider also tags its OWN + * locally-originated transactions with an ORIGIN carrying its own node + * id/name. Excluding InvalidRepOriginId and the direct provider + * (MySubscription->origin->id) keeps the provider's own commit from ever + * being mistaken for a forwarded peer transaction. + * + * hash_search(HASH_FIND) is a cheap in-memory lookup, so it is fine to do + * unconditionally on every ORIGIN message, regardless of how many distinct + * peers are being forwarded or how their transactions interleave -- unlike + * a single-entry "last seen" cache, this pays the catalog-lookup cost only + * once per peer, ever. That catalog lookup runs here, outside any active + * transaction, so handle_commit()/maybe_advance_forwarded_origin() stay + * free of any transaction overhead and just call replorigin_advance() + * directly. + * + * Only insert when a local subscription to this peer was actually found. A + * subscription can be created to a peer at any time, independent of when + * its transactions start arriving forwarded through some other provider — + * nothing here controls that ordering. Leaving a failed lookup out of the + * cache means the next occurrence of this peer's origin id retries, so a + * subscription created after this worker already saw (and missed on) that + * peer is still picked up, without requiring an apply worker restart. + */ +static void +track_forward_peer_origin(RepOriginId remote_origin_id) +{ + bool found; + + if (remote_origin_id == InvalidRepOriginId || + remote_origin_id == (RepOriginId) MySubscription->origin->id) + return; + + if (ForwardOriginCache == NULL) + forward_origin_cache_init(); + + hash_search(ForwardOriginCache, &remote_origin_id, HASH_FIND, &found); + if (!found) + { + RepOriginId local_id; + + StartTransactionCommand(); + local_id = resolve_forward_peer_origin(remote_origin_id); + CommitTransactionCommand(); + MemoryContextSwitchTo(MessageContext); + + if (local_id != InvalidRepOriginId) + { + ForwardOriginCacheEntry *entry; + + entry = hash_search(ForwardOriginCache, &remote_origin_id, + HASH_ENTER, NULL); + entry->local_origin_id = local_id; + } + } } /* @@ -4867,101 +5006,60 @@ apply_replay_queue_start_replay(void) /* * Advance the replication origin for forwarded transactions. * - * In cascade replication (A -> B -> C with forward_origins='all'), when C - * receives transactions that originated on A (forwarded through B), we track - * C's position relative to A by maintaining a separate replication origin. + * Called from handle_commit() for forwarded transactions (those carrying an + * ORIGIN message from a peer node, not our direct provider). * - * This enables seamless switchover: if C later subscribes directly to A, - * the origin will already exist with the correct LSN, so C knows where to - * start receiving from A. + * handle_origin() already resolved the cache: a hash_search(HASH_FIND) below + * finds the RepOriginId of this node's own subscription to the forwarding + * peer, keyed by the peer's node id, if one exists. There is no entry when + * no such subscription exists (pure cascade topology, the ordinary case), + * or when this specific peer has not yet been resolved -- either way, + * nothing to advance. * - * The origin is named using slot name format (spk___) - * for consistency with direct subscriptions. + * The origin records two LSNs in two different WAL spaces: the remote LSN is + * the peer's commit position (remote_origin_lsn, peer space) and the local + * LSN is our own commit position for the applied transaction (local_lsn = + * XactLastCommitEnd, or InvalidXLogRecPtr for an empty transaction). Never + * pass the provider's end_lsn as the local value — it poisons the origin's + * crash-recovery ordering. * - * We cache the remote_origin_id -> local_origin_id mapping since the Spock - * node ID is stable across the cluster (set by commit f60484e). + * wal_log=true makes this advance durable through PG's normal + * replication-origin machinery: the peer's origin becomes the position that + * subscription resumes from once/if it is later enabled -- no separate + * ledger or reconciliation step, the origin is the single source of truth, + * advanced in place as forwarded transactions commit. One residual gap is + * inherent: this advance is a separate WAL record *after* the data commit + * (PG allows only one session origin per commit, and that one belongs to + * our actual provider's subscription), so a crash in that narrow window + * loses only the last forwarded transaction's advance, which is simply + * re-delivered once forwarding resumes -- at-least-once, matching ordinary + * Spock apply. last_update_wins absorbs the duplicate; delta-apply + * idempotency under such re-delivery is a separate, tracked follow-up. */ static void -maybe_advance_forwarded_origin(XLogRecPtr end_lsn, bool xact_had_exception) +maybe_advance_forwarded_origin(XLogRecPtr local_lsn, bool xact_had_exception) { - RepOriginId forwarded_origin; + ForwardOriginCacheEntry *entry; + bool found; - /* - * Only advance for forwarded transactions (origin differs from our direct - * provider) that completed without exceptions. - */ if (xact_had_exception || remote_origin_id == InvalidRepOriginId || - remote_origin_id == MySubscription->origin->id || - remote_origin_name == NULL) + remote_origin_id == (RepOriginId) MySubscription->origin->id || + remote_origin_name == NULL || + ForwardOriginCache == NULL) return; - /* - * Check cache first. The remote_origin_id (Spock node ID) is stable - * for a given source node, so we can reuse the local origin ID. - */ - if (remote_origin_id == cached_forward_remote_id && - cached_forward_local_id != InvalidRepOriginId) - { - forwarded_origin = cached_forward_local_id; - - elog(DEBUG2, "SPOCK %s: advancing forwarded origin (cached, oid %u) " - "remote_lsn %X/%X end_lsn %X/%X", - MySubscription->name, - forwarded_origin, - (uint32) (remote_origin_lsn >> 32), (uint32) remote_origin_lsn, - (uint32) (end_lsn >> 32), (uint32) end_lsn); - } - else - { - /* - * Cache miss - look up or create the origin. Use slot name format - * (spk___) for consistency with direct - * subscriptions. - */ - Relation replorigin_rel; - NameData slot_name; - char *dbname; - - StartTransactionCommand(); - - dbname = get_database_name(MyDatabaseId); - gen_slot_name(&slot_name, dbname, remote_origin_name, - MySubscription->name); - - elog(DEBUG2, "SPOCK %s: advancing forwarded origin '%s' (from node '%s') " - "remote_lsn %X/%X end_lsn %X/%X", - MySubscription->name, - NameStr(slot_name), - remote_origin_name, - (uint32) (remote_origin_lsn >> 32), (uint32) remote_origin_lsn, - (uint32) (end_lsn >> 32), (uint32) end_lsn); - - replorigin_rel = table_open(ReplicationOriginRelationId, RowExclusiveLock); - forwarded_origin = replorigin_by_name(NameStr(slot_name), true); - - if (forwarded_origin == InvalidRepOriginId) - { - forwarded_origin = replorigin_create(NameStr(slot_name)); - elog(DEBUG2, "SPOCK %s: created replication origin '%s' (oid %u) " - "for forwarded transactions from node '%s'", - MySubscription->name, NameStr(slot_name), forwarded_origin, - remote_origin_name); - } - - table_close(replorigin_rel, RowExclusiveLock); - CommitTransactionCommand(); - MemoryContextSwitchTo(MessageContext); + entry = hash_search(ForwardOriginCache, &remote_origin_id, HASH_FIND, &found); + if (!found) + return; - /* Update cache */ - cached_forward_remote_id = remote_origin_id; - cached_forward_local_id = forwarded_origin; - } + elog(DEBUG2, "SPOCK %s: advancing forwarded origin (oid %u) " + "remote_lsn %X/%X local_lsn %X/%X", + MySubscription->name, + entry->local_origin_id, + (uint32) (remote_origin_lsn >> 32), (uint32) remote_origin_lsn, + (uint32) (local_lsn >> 32), (uint32) local_lsn); - /* Advance the origin */ - StartTransactionCommand(); - replorigin_advance(forwarded_origin, remote_origin_lsn, - end_lsn, false, false); - CommitTransactionCommand(); - MemoryContextSwitchTo(MessageContext); + replorigin_advance(entry->local_origin_id, remote_origin_lsn, + local_lsn, false, true); } diff --git a/tests/tap/t/015_forward_origin_advance.pl b/tests/tap/t/015_forward_origin_advance.pl index d2ceea047..278471de4 100755 --- a/tests/tap/t/015_forward_origin_advance.pl +++ b/tests/tap/t/015_forward_origin_advance.pl @@ -2,29 +2,40 @@ # ============================================================================= # Test: 015_forward_origin_advance.pl - Verify Forward Origin Tracking # ============================================================================= -# This test verifies that when forward_origins='all' is set, the subscriber -# creates and advances a replication origin for the original source node. +# This test verifies that when forward_origins='all' is set and a disabled +# subscription is pre-created on the new node to a peer, the apply worker +# on the new node correctly advances that pre-created origin as the peer's +# forwarded transactions arrive during catchup. # # Topology: -# A (n1) -> B (n2) -> C (n3) -# forward_origins='all' on both subscriptions +# n2 -> n1 -> n3 +# forward_origins='all' only on n3's subscription to n1 -- that is +# what tells n1 to relay transactions it received from n2, rather +# than only its own # -# Expected behavior: -# - A inserts data -# - B receives from A and forwards to C -# - C should create a replication origin using slot name format: -# spk___ (e.g., "spk_regression_n1_n3") -# - C should advance that origin's LSN as it receives forwarded transactions +# Roles: +# n1 = source — the node n3 actively subscribes to for catchup (enabled, +# forward_origins='all'); forwards n2's changes to n3 +# n2 = peer — existing cluster member whose changes are forwarded +# through n1 to n3 +# n3 = new_node — the node being added to the cluster # -# This test FAILS on 'main' branch (origin not created) -# This test PASSES on 'task/SPOC-228/physical-to-logical-replica' branch +# Expected behavior: +# - Before catchup: n3 pre-creates a disabled subscription to n2 (sub_n3_n2). +# This creates a named replication origin at LSN 0/0. +# - n2 inserts data; n1 receives it (via sub_n1_n2) and forwards it to n3 +# (via the enabled sub_n3_n1, forward_origins='all') +# - n3 advances the pre-created sub_n3_n2 origin LSN as forwarded n2 +# transactions arrive +# - When sub_enable('sub_n3_n2') fires, the apply worker starts from +# the correct position — no duplicates, no gaps # ============================================================================= use strict; use warnings; -use Test::More tests => 22; +use Test::More tests => 31; use lib '.'; -use SpockTest qw(create_cluster destroy_cluster system_or_bail system_maybe command_ok get_test_config scalar_query psql_or_bail); +use SpockTest qw(create_cluster destroy_cluster system_or_bail get_test_config scalar_query psql_or_bail wait_for_sub_status); # ============================================================================= # SETUP: Create 3-node cluster @@ -32,15 +43,13 @@ create_cluster(3, 'Create 3-node cluster'); -my $config = get_test_config(); +my $config = get_test_config(); my $node_ports = $config->{node_ports}; -my $pg_bin = $config->{pg_bin}; -my $dbname = $config->{db_name}; -my $host = $config->{host}; +my $dbname = $config->{db_name}; +my $host = $config->{host}; -# Connection strings -my $conn_a = "host=$host port=$node_ports->[0] dbname=$dbname"; -my $conn_b = "host=$host port=$node_ports->[1] dbname=$dbname"; +my $conn_n1 = "host=$host port=$node_ports->[0] dbname=$dbname"; +my $conn_n2 = "host=$host port=$node_ports->[1] dbname=$dbname"; # ============================================================================= # TEST: Forward Origin Advance @@ -64,116 +73,191 @@ psql_or_bail(3, "SELECT spock.repset_add_table('cascade_set', 'test_origin')"); pass('Added table to replication sets'); -# Create cascade: A -> B -> C -# B subscribes to A with forward_origins='all' -psql_or_bail(2, "SELECT spock.sub_create('sub_a_to_b', '$conn_a', ARRAY['cascade_set'], false, false, ARRAY['all'])"); -pass('Created subscription B->A with forward_origins=all'); - -# C subscribes to B with forward_origins='all' -psql_or_bail(3, "SELECT spock.sub_create('sub_b_to_c', '$conn_b', ARRAY['cascade_set'], false, false, ARRAY['all'])"); -pass('Created subscription C->B with forward_origins=all'); - -# Wait for subscriptions to be ready -system_or_bail 'sleep', '5'; - -# Verify subscriptions are replicating -my $sub_b = scalar_query(2, "SELECT 1 FROM spock.sub_show_status() WHERE subscription_name = 'sub_a_to_b' AND status = 'replicating'"); -is($sub_b, '1', 'Subscription A->B is replicating'); - -my $sub_c = scalar_query(3, "SELECT 1 FROM spock.sub_show_status() WHERE subscription_name = 'sub_b_to_c' AND status = 'replicating'"); -is($sub_c, '1', 'Subscription B->C is replicating'); - -# Insert data on A - this will be forwarded through B to C -psql_or_bail(1, "INSERT INTO test_origin (val) VALUES ('from_node_a')"); +# n1 subscribes to n2, an ordinary subscription. n2's own commits carry no +# origin (InvalidRepOriginId, since n2 has no upstream peer of its own in +# this test) and are never filtered by forward_origins regardless of its +# setting, so this leg does not need it. +psql_or_bail(1, "SELECT spock.sub_create('sub_n1_n2', '$conn_n2', ARRAY['cascade_set'], false, false)"); +pass('Created subscription n1->n2'); + +# n3 subscribes to n1 (the source) for catchup. forward_origins='all' here +# is what tells n1's output plugin to also include transactions n1 itself +# received from n2 (tagged with n2's origin), rather than only n1's own. +psql_or_bail(3, "SELECT spock.sub_create('sub_n3_n1', '$conn_n1', ARRAY['cascade_set'], false, false, ARRAY['all'])"); +pass('Created subscription n3->n1 with forward_origins=all'); + +# Pre-create a disabled subscription on n3 to peer n2 before catchup begins. +# sub_create(enabled=false) creates a named replication origin on n3 at LSN 0/0 +# without starting an apply worker. track_forward_peer_origin()/ +# maybe_advance_forwarded_origin() advance it as forwarded n2 transactions +# arrive, so that sub_enable('sub_n3_n2') starts from the correct LSN. +psql_or_bail(3, "SELECT spock.sub_create( + subscription_name := 'sub_n3_n2', + provider_dsn := '$conn_n2', + replication_sets := ARRAY['cascade_set'], + synchronize_structure := false, + synchronize_data := false, + enabled := false +)"); +pass('Pre-created disabled subscription n3->n2'); + +# sub_create(enabled=false) creates the catalog entry and replication origin +# but not a slot on the provider. In the real join flow, spock_create_subscriber +# handles slot creation. Create it directly here so sub_enable() can start the +# apply worker. +my $sub_n3_n2_slot = scalar_query(3, + "SELECT spock.spock_gen_slot_name(current_database()::name, 'n2'::name, 'sub_n3_n2'::name)"); +psql_or_bail(2, "SELECT pg_create_logical_replication_slot('$sub_n3_n2_slot', 'spock_output')"); + +ok(wait_for_sub_status(1, 'sub_n1_n2', 'replicating', 30), + 'Subscription n1->n2 is replicating'); +ok(wait_for_sub_status(3, 'sub_n3_n1', 'replicating', 30), + 'Subscription n3->n1 is replicating'); + +my $sub_disabled = scalar_query(3, "SELECT 1 FROM spock.sub_show_status() WHERE subscription_name = 'sub_n3_n2' AND status = 'disabled'"); +is($sub_disabled, '1', 'Subscription sub_n3_n2 is disabled on n3'); + +# Insert data on n2 — will be forwarded through n1 to n3 +psql_or_bail(2, "INSERT INTO test_origin (val) VALUES ('from_n2')"); system_or_bail 'sleep', '5'; -# Verify data reached C -my $count_c = scalar_query(3, "SELECT COUNT(*) FROM test_origin WHERE val = 'from_node_a'"); -is($count_c, '1', 'Data from A reached C via B'); +my $count_n3 = scalar_query(3, "SELECT COUNT(*) FROM test_origin WHERE val = 'from_n2'"); +is($count_n3, '1', 'Data from n2 reached n3 via n1'); # ============================================================================= -# KEY TEST: Check if C has created a replication origin for node A +# KEY TEST: Check that n3's pre-created origin for n2 has been advanced # ============================================================================= -# On main branch: This origin will NOT exist (test fails) -# On fix branch: This origin WILL exist (test passes) -# -# The origin uses slot name format: spk___ -# e.g., "spk_regression_n1_sub_b_to_c" for forwarded transactions from n1 via sub_b_to_c - -my $expected_origin = scalar_query(3, "SELECT spock.spock_gen_slot_name(current_database()::name, 'n1'::name, 'sub_b_to_c'::name)"); -diag("Expected forwarded origin name on C: $expected_origin"); - -my $origin_exists = scalar_query(3, "SELECT COUNT(*) FROM pg_replication_origin WHERE roname = '$expected_origin'"); -diag("Origin '$expected_origin' exists on C: $origin_exists"); -is($origin_exists, '1', "C has replication origin for forwarded source ($expected_origin)"); - -# Also verify the origin has a valid LSN (not 0/0) -my $origin_lsn = scalar_query(3, "SELECT COALESCE(s.remote_lsn::text, 'NULL') FROM pg_replication_origin o LEFT JOIN pg_replication_origin_status s ON o.roident = s.local_id WHERE o.roname = '$expected_origin'"); -diag("Origin '$expected_origin' LSN on C: $origin_lsn"); -ok($origin_lsn ne '0/0' && $origin_lsn ne 'NULL' && $origin_lsn ne '', "Origin $expected_origin has been advanced (LSN is valid)"); +# sub_create(enabled=false) created the origin at 0/0. +# As forwarded n2 transactions arrive on n3, maybe_advance_forwarded_origin() +# looks up sub_n3_n2 by n2's node OID and advances its named origin. +# When sub_enable('sub_n3_n2') fires, the apply worker reads this origin +# and starts replication from the already-advanced position. + +my $expected_origin = scalar_query(3, + "SELECT spock.spock_gen_slot_name(current_database()::name, 'n2'::name, 'sub_n3_n2'::name)"); +diag("Expected forwarded origin name on n3: $expected_origin"); + +my $origin_exists = scalar_query(3, + "SELECT COUNT(*) FROM pg_replication_origin WHERE roname = '$expected_origin'"); +diag("Origin '$expected_origin' exists on n3: $origin_exists"); +is($origin_exists, '1', "n3 has replication origin for forwarded n2 ($expected_origin)"); + +my $origin_lsn = scalar_query(3, + "SELECT COALESCE(s.remote_lsn::text, 'NULL') + FROM pg_replication_origin o + LEFT JOIN pg_replication_origin_status s ON o.roident = s.local_id + WHERE o.roname = '$expected_origin'"); +diag("Origin '$expected_origin' LSN on n3: $origin_lsn"); +ok($origin_lsn ne '0/0' && $origin_lsn ne 'NULL' && $origin_lsn ne '', + "Origin $expected_origin has been advanced (LSN is valid)"); # ============================================================================= # GAP DETECTION TEST: Demonstrate origin tracking enables gap detection # ============================================================================= -# This test demonstrates that forwarded origin tracking enables detection of -# unreplicated data during cascade switchover scenarios. -# -# With the fix (forwarded origin tracking): -# - We can query C's position relative to A via the forwarded origin LSN -# - We can query A's current position: pg_current_wal_lsn() -# - If A's LSN > C's tracked LSN, there's unreplicated data (a "gap") -# - This enables tooling to make informed switchover decisions -# -# Without the fix (main branch): -# - C has no forwarded origin, so we cannot detect gaps -# - Switchover is "blind" - we don't know what C has received from A diag("=== GAP DETECTION TEST: Origin tracking enables gap detection ==="); -# Insert more data and let it propagate -psql_or_bail(1, "INSERT INTO test_origin (val) VALUES ('batch2_row1')"); -psql_or_bail(1, "INSERT INTO test_origin (val) VALUES ('batch2_row2')"); +psql_or_bail(2, "INSERT INTO test_origin (val) VALUES ('batch2_row1')"); +psql_or_bail(2, "INSERT INTO test_origin (val) VALUES ('batch2_row2')"); system_or_bail 'sleep', '3'; -# Verify data reached C my $count_after_batch2 = scalar_query(3, "SELECT COUNT(*) FROM test_origin"); -diag("Row count on C after batch 2: $count_after_batch2"); -is($count_after_batch2, '3', 'C has 3 rows after batch 2'); +diag("Row count on n3 after batch 2: $count_after_batch2"); +is($count_after_batch2, '3', 'n3 has 3 rows after batch 2'); -# KEY TEST: Query C's tracked position for forwarded origin (only works with fix) -my $c_origin_lsn = scalar_query(3, "SELECT COALESCE(s.remote_lsn::text, 'not_tracked') FROM pg_replication_origin o LEFT JOIN pg_replication_origin_status s ON o.roident = s.local_id WHERE o.roname = '$expected_origin'"); -diag("C's tracked LSN for A (origin '$expected_origin'): $c_origin_lsn"); +my $c_origin_lsn = scalar_query(3, + "SELECT COALESCE(s.remote_lsn::text, 'not_tracked') + FROM pg_replication_origin o + LEFT JOIN pg_replication_origin_status s ON o.roident = s.local_id + WHERE o.roname = '$expected_origin'"); +diag("n3's tracked LSN for n2 (origin '$expected_origin'): $c_origin_lsn"); -# Query A's current WAL position -my $a_wal_lsn = scalar_query(1, "SELECT pg_current_wal_lsn()::text"); -diag("A's current WAL LSN: $a_wal_lsn"); - -# On fix branch: c_origin_lsn should be a valid LSN (not 'not_tracked') -# On main branch: c_origin_lsn would be 'not_tracked' (query returns nothing) ok($c_origin_lsn ne 'not_tracked' && $c_origin_lsn ne '', - 'C can track its position relative to A (gap detection enabled)'); + 'n3 can track its position relative to n2 (gap detection enabled)'); -# Simulate gap: disable B->A, insert on A, check that we can detect the gap -diag("Creating gap: disabling B->A subscription..."); -psql_or_bail(2, "SELECT spock.sub_disable('sub_a_to_b')"); +# Simulate gap: disable n1->n2, insert on n2, check origin does not advance +diag("Creating gap: disabling n1->n2 subscription..."); +psql_or_bail(1, "SELECT spock.sub_disable('sub_n1_n2')"); system_or_bail 'sleep', '2'; -# Insert data on A that creates a gap (can't reach C) -psql_or_bail(1, "INSERT INTO test_origin (val) VALUES ('gap_data')"); -my $a_wal_after_gap = scalar_query(1, "SELECT pg_current_wal_lsn()::text"); -diag("A's WAL LSN after gap data: $a_wal_after_gap"); +psql_or_bail(2, "INSERT INTO test_origin (val) VALUES ('gap_data')"); +system_or_bail 'sleep', '5'; -# C's origin LSN should still be at the old position (gap exists) -my $c_origin_after_gap = scalar_query(3, "SELECT COALESCE(s.remote_lsn::text, 'not_tracked') FROM pg_replication_origin o LEFT JOIN pg_replication_origin_status s ON o.roident = s.local_id WHERE o.roname = '$expected_origin'"); -diag("C's origin LSN (unchanged, gap detected): $c_origin_after_gap"); +my $c_origin_after_gap = scalar_query(3, + "SELECT COALESCE(s.remote_lsn::text, 'not_tracked') + FROM pg_replication_origin o + LEFT JOIN pg_replication_origin_status s ON o.roident = s.local_id + WHERE o.roname = '$expected_origin'"); +diag("n3's origin LSN (unchanged, gap detected): $c_origin_after_gap"); -# Verify the LSNs show a gap (A advanced, C's tracking hasn't) -# This is the key insight: with origin tracking, we KNOW there's unreplicated data -is($c_origin_after_gap, $c_origin_lsn, 'Gap detected: C origin unchanged while A advanced'); +is($c_origin_after_gap, $c_origin_lsn, 'Gap detected: n3 origin unchanged while n2 advanced'); -# Clean test: verify C still has 3 rows (gap data didn't arrive) my $count_with_gap = scalar_query(3, "SELECT COUNT(*) FROM test_origin"); -is($count_with_gap, '3', 'C still has 3 rows (gap data not received)'); +is($count_with_gap, '3', 'n3 still has 3 rows (gap data not received)'); + +# ============================================================================= +# GAP RECOVERY TEST: Re-enable n1->n2 and verify gap data arrives on n3 +# ============================================================================= + +psql_or_bail(1, "SELECT spock.sub_enable('sub_n1_n2')"); +system_or_bail 'sleep', '5'; + +my $count_after_reenable = scalar_query(3, "SELECT COUNT(*) FROM test_origin"); +is($count_after_reenable, '4', 'n3 received gap_data after n1->n2 re-enabled'); + +my $c_origin_after_reenable = scalar_query(3, + "SELECT COALESCE(s.remote_lsn::text, 'not_tracked') + FROM pg_replication_origin o + LEFT JOIN pg_replication_origin_status s ON o.roident = s.local_id + WHERE o.roname = '$expected_origin'"); +diag("n3's origin LSN after re-enable: $c_origin_after_reenable"); +ok($c_origin_after_reenable ne $c_origin_lsn, + 'n3 forwarded origin LSN advanced after gap closed'); + +# ============================================================================= +# SUB_ENABLE TEST: Verify no duplicates when sub_n3_n2 is enabled +# ============================================================================= +# The forwarded origin on n3 for n2 has been advanced during cascade catchup. +# When sub_enable('sub_n3_n2') fires, the apply worker calls +# replorigin_session_get_progress() on the pre-created origin and receives +# the forwarded LSN — so it starts replication from that position, skipping +# rows n3 already received via the n1 cascade. This is the end-to-end proof +# that maybe_advance_forwarded_origin() prevents duplicates. + +# Disable the cascade leg so that new rows on n2 can only reach n3 via the +# direct subscription we are about to enable. Without this, data arriving +# via sub_n3_n1 would mask whether sub_n3_n2 is actually running. +psql_or_bail(3, "SELECT spock.sub_disable('sub_n3_n1')"); + +psql_or_bail(3, "SELECT spock.sub_enable('sub_n3_n2')"); +pass('Enabled direct subscription sub_n3_n2 on n3'); + +# No-duplicates check: if the apply worker started from 0/0 instead of the +# forwarded LSN, it would re-send all 4 rows already on n3, making the count +# jump to 8. A stable count of 4 proves the correct start position was used. +system_or_bail 'sleep', '3'; +my $count_no_dup = scalar_query(3, "SELECT COUNT(*) FROM test_origin"); +is($count_no_dup, '4', 'No duplicates: existing rows not re-sent after sub_enable'); + +# Insert a new row on n2 and poll for it to arrive on n3. With sub_n3_n1 +# disabled, the only path is sub_n3_n2, so arrival proves that subscription +# is live and replicating. +psql_or_bail(2, "INSERT INTO test_origin (val) VALUES ('via_direct_sub')"); +my $direct_arrived = 0; +for my $attempt (1..60) { + my $cnt = scalar_query(3, + "SELECT COUNT(*) FROM test_origin WHERE val = 'via_direct_sub'"); + if (defined $cnt && $cnt >= 1) { $direct_arrived = 1; last; } + sleep 1; +} +ok($direct_arrived, 'sub_n3_n2 is replicating directly from n2 (cascade disabled)'); + +my $count_after_direct = scalar_query(3, "SELECT COUNT(*) FROM test_origin"); +is($count_after_direct, '5', 'n3 has exactly 5 rows via direct sub_n3_n2'); + +my $direct_row = scalar_query(3, + "SELECT COUNT(*) FROM test_origin WHERE val = 'via_direct_sub'"); +is($direct_row, '1', 'Direct row arrived exactly once on n3'); # ============================================================================= # CLEANUP From d3b5370366ab6751f98c462bf708c7b184d2bab3 Mon Sep 17 00:00:00 2001 From: Asif Rehman Date: Wed, 22 Jul 2026 10:35:51 +0500 Subject: [PATCH 5/7] spock_create_subscriber address review feedback Harden option parsing and cleanup unrelated to bidirectional join: - Reject --apply-delay values that aren't a clean integer instead of silently taking atoi()'s partial parse. - Validate --extra-basebackup-args against shell metacharacters before it is appended to a system() command string; the args are otherwise a command-injection vector. - Free the read_sysid() allocation on the existing-data-dir sysid check. - Fix a missing separator when appending --extra-basebackup-args to the pg_basebackup command line. - Replace sprintf with snprintf when writing postgresql.auto.conf. - Scope the post-sub_create sync_status fixup with a WHERE clause instead of unconditionally rewriting every row. - Document --text-types and a docs formatting nit. --- docs/creating_subscriber_nodes.md | 5 +- .../spock_create_subscriber.c | 62 +++++++++++++++---- 2 files changed, 53 insertions(+), 14 deletions(-) diff --git a/docs/creating_subscriber_nodes.md b/docs/creating_subscriber_nodes.md index bc4622394..64528071f 100644 --- a/docs/creating_subscriber_nodes.md +++ b/docs/creating_subscriber_nodes.md @@ -2,7 +2,7 @@ Spock supports creating a subscriber node by cloning the provider with [`pg_basebackup`](https://www.postgresql.org/docs/current/app-pgbasebackup.html) and starting it as a Spock subscriber. Use the `spock_create_subscriber` utility (located in the `bin` directory of your pgEdge platform installation) to register the node. -#### Synopsis: +### Synopsis: `spock_create_subscriber [OPTION]...` @@ -17,12 +17,13 @@ Specify the following options as needed. | `-n`, `--subscriber-name=NAME` | The name of the newly created subscriber. | `--subscriber-dsn=CONNSTR` | A connection string to the newly created subscriber. | `--provider-dsn=CONNSTR` | A connection string to the provider. -| `--replication-sets=SETS` | A comma separated list of replication set names. +| `--replication-sets=SETS` | A comma-separated list of replication set names. | `--apply-delay=DELAY` | The apply delay in seconds (by default 0). | `--drop-slot-if-exists` | Drop replication slot of conflicting name. | `-s`, `--stop` | Stop the server once the initialization is done. | `-v` | Increase logging verbosity. | `--extra-basebackup-args` | Additional arguments to pass to `pg_basebackup`. Safe options are: `-T`, `-c`, `--xlogdir`/`--waldir` +| `--text-types` | Transfer all column values as text rather than binary during initial sync. Use this when provider and subscriber differ in endianness or type representation. **Configuration files overrides** diff --git a/utils/spock_create_subscriber/spock_create_subscriber.c b/utils/spock_create_subscriber/spock_create_subscriber.c index fbf85fdfe..076280386 100644 --- a/utils/spock_create_subscriber/spock_create_subscriber.c +++ b/utils/spock_create_subscriber/spock_create_subscriber.c @@ -90,6 +90,7 @@ static void print_msg(VerbosityLevelEnum level, const char *fmt,...) pg_attribute_printf(2, 3); static int run_pg_ctl(const char *arg); +static void validate_extra_basebackup_args(const char *args); static void run_basebackup(const char *provider_connstr, const char *data_dir, const char *extra_basebackup_args); static void wait_postmaster_connection(const char *connstr); @@ -281,13 +282,19 @@ main(int argc, char **argv) drop_slot_if_exists = true; break; case 8: - apply_delay = atoi(optarg); + { + char *endptr; + apply_delay = (int) strtol(optarg, &endptr, 10); + if (*endptr != '\0' || endptr == optarg) + die(_("--apply-delay requires an integer value\n")); + } break; case 9: databases = pg_strdup(optarg); break; case 10: extra_basebackup_args = pg_strdup(optarg); + validate_extra_basebackup_args(extra_basebackup_args); break; case 11: force_text_transfer = true; @@ -407,9 +414,14 @@ main(int argc, char **argv) { use_existing_data_dir = check_data_dir(data_dir, remote_info); - if (use_existing_data_dir && - strcmp(remote_info->sysid, read_sysid(data_dir)) != 0) - die(_("Subscriber data directory is not basebackup of remote node.\n")); + if (use_existing_data_dir) + { + char *local_sysid = read_sysid(data_dir); + bool mismatch = strcmp(remote_info->sysid, local_sysid) != 0; + free(local_sysid); + if (mismatch) + die(_("Subscriber data directory is not basebackup of remote node.\n")); + } } /* @@ -597,6 +609,9 @@ usage(void) printf(_(" -v increase logging verbosity\n")); printf(_(" --extra-basebackup-args additional arguments to pass to pg_basebackup.\n")); printf(_(" Safe options: -T, -c, --xlogdir/--waldir\n")); + printf(_(" --text-types transfer column values as text rather than binary\n")); + printf(_(" (use when provider and subscriber differ in type\n")); + printf(_(" representation or endianness)\n")); printf(_("\nConfiguration files override:\n")); printf(_(" --hba-conf path to the new pg_hba.conf\n")); printf(_(" --postgresql-conf path to the new postgresql.conf\n")); @@ -682,6 +697,26 @@ run_pg_ctl(const char *arg) } +/* + * Reject --extra-basebackup-args values that contain shell control characters. + * The args are appended to a system() command string, so semicolons, pipes, + * backticks, and similar metacharacters would allow arbitrary command injection. + */ +static void +validate_extra_basebackup_args(const char *args) +{ + const char *p; + + for (p = args; *p; p++) + { + if (*p == ';' || *p == '|' || *p == '&' || *p == '`' || + *p == '$' || *p == '(' || *p == ')' || + *p == '<' || *p == '>' || *p == '{' || *p == '}' || + *p == '\n' || *p == '\r') + die(_("--extra-basebackup-args contains unsafe shell characters\n")); + } +} + /* * Run pg_basebackup to create the copy of the origin node. */ @@ -700,7 +735,7 @@ run_basebackup(const char *provider_connstr, const char *data_dir, appendPQExpBuffer(cmd, " -v"); if (extra_basebackup_args != NULL) - appendPQExpBuffer(cmd, "%s", extra_basebackup_args); + appendPQExpBuffer(cmd, " %s", extra_basebackup_args); print_msg(VERBOSITY_DEBUG, _("Running pg_basebackup: %s.\n"), cmd->data); ret = system(cmd->data); @@ -1097,8 +1132,8 @@ spock_subscribe(PGconn *conn, char *subscriber_name, char *subscriber_dsn, } PQclear(res); - /* TODO */ - res = PQexec(conn, "UPDATE spock.local_sync_status SET sync_status = 'r'"); + res = PQexec(conn, "UPDATE spock.local_sync_status SET sync_status = 'r'" + " WHERE sync_status != 'r'"); if (PQresultStatus(res) != PGRES_COMMAND_OK) { die(_("Could not update subscription, status %s: %s\n"), @@ -1317,7 +1352,7 @@ WriteRecoveryConf(PQExpBuffer contents) char filename[MAXPGPATH]; FILE *cf; - sprintf(filename, "%s/postgresql.auto.conf", data_dir); + snprintf(filename, sizeof(filename), "%s/postgresql.auto.conf", data_dir); cf = fopen(filename, "a"); if (cf == NULL) @@ -1334,7 +1369,7 @@ WriteRecoveryConf(PQExpBuffer contents) fclose(cf); { - sprintf(filename, "%s/standby.signal", data_dir); + snprintf(filename, sizeof(filename), "%s/standby.signal", data_dir); cf = fopen(filename, "w"); if (cf == NULL) { @@ -1353,7 +1388,7 @@ CopyConfFile(char *fromfile, char *tofile, bool append) { char filename[MAXPGPATH]; - sprintf(filename, "%s/%s", data_dir, tofile); + snprintf(filename, sizeof(filename), "%s/%s", data_dir, tofile); print_msg(VERBOSITY_DEBUG, _("Copying \"%s\" to \"%s\".\n"), fromfile, filename); @@ -1560,7 +1595,7 @@ is_pg_dir(const char *path) if (stat(path, &statbuf) != 0) return false; - snprintf(version_file, MAXPGPATH, "%s/PG_VERSION", data_dir); + snprintf(version_file, MAXPGPATH, "%s/PG_VERSION", path); if (stat(version_file, &statbuf) != 0 && errno == ENOENT) { return false; @@ -1711,6 +1746,7 @@ get_pgpid(void) } if (fscanf(pidf, "%ld", &pid) != 1) { + fclose(pidf); return 0; } fclose(pidf); @@ -1746,6 +1782,8 @@ static char * generate_restore_point_name(void) { char *rpn = malloc(NAMEDATALEN); - snprintf(rpn, NAMEDATALEN-1, "spock_create_subscriber_%lx", random()); + if (rpn == NULL) + die(_("out of memory\n")); + snprintf(rpn, NAMEDATALEN, "spock_create_subscriber_%lx", random()); return rpn; } From 80578d289002271dad5a49e8a70927fd17ecd6fb Mon Sep 17 00:00:00 2001 From: Asif Rehman Date: Thu, 9 Jul 2026 15:43:36 +0500 Subject: [PATCH 6/7] spock_create_subscriber: add bidirectional join plumbing (PR2) Add --bidirectional, --stall-timeout, --max-wait, and --cleanup options to support the bidirectional node-join procedure defined in the SPOC-601 design. In --bidirectional mode the tool connects to the source cluster, discovers all peer nodes via spock.subscription/node/node_interface, verifies preconditions (Spock >= 6.0.0, track_commit_timestamp on on all nodes, no pending DDL, full-mesh topology, per-peer connectivity), then writes a JSON manifest to /spock_bidirectional_manifest.json and exits. The manifest records peer names, DSNs, slot names, and sub names so that later phases (PR3+) can resume idempotently. In --cleanup mode the tool reads the manifest and idempotently removes any partial state left by a prior attempt: drops replication slots on the source and each peer, drops reverse subscriptions, and removes the manifest file. Connectivity failures during cleanup are logged as warnings rather than being fatal. No replication behavior is changed in this commit; the subscriber DSN is not required in --bidirectional mode since the subscriber database does not exist yet at this stage. --- .../spock_create_subscriber.c | 791 +++++++++++++++++- 1 file changed, 785 insertions(+), 6 deletions(-) diff --git a/utils/spock_create_subscriber/spock_create_subscriber.c b/utils/spock_create_subscriber/spock_create_subscriber.c index 076280386..6ccc5bf7e 100644 --- a/utils/spock_create_subscriber/spock_create_subscriber.c +++ b/utils/spock_create_subscriber/spock_create_subscriber.c @@ -52,6 +52,8 @@ #include "access/timeline.h" #include "access/xlog_internal.h" #include "catalog/pg_control.h" +#include "common/jsonapi.h" +#include "mb/pg_wchar.h" #include "spock_fe.h" @@ -65,6 +67,30 @@ typedef struct RemoteInfo { char *replication_sets; } RemoteInfo; +typedef struct PeerNodeInfo +{ + char *node_name; + char *dsn; + char *slot_name; /* from spock.spock_gen_slot_name() */ + char *sub_name; /* "sub__" */ + bool disabled_sub_created; + bool slot_created; + bool reverse_sub_created; +} PeerNodeInfo; + +typedef struct BidirectionalState +{ + bool enabled; + int num_peers; + PeerNodeInfo *peers; + int stall_timeout; /* default 600s */ + int max_wait; /* default 0 = unbounded */ + char *source_slot_name; + char *source_origin_name; + bool cleanup_mode; + char *manifest_path; +} BidirectionalState; + typedef enum { VERBOSITY_NORMAL, VERBOSITY_VERBOSE, @@ -141,6 +167,20 @@ static long get_pgpid(void); static char **get_database_list(char *databases, int *n_databases); static char *generate_restore_point_name(void); +static int discover_peer_nodes(PGconn *source_conn, const char *source_node_name, + const char *subscriber_name, const char *dbname, + PeerNodeInfo **peers_out); +static void check_preconditions(PGconn *source_conn, PeerNodeInfo *peers, int num_peers); +static void write_manifest(BidirectionalState *state, const char *subscriber_name, + const char *dbname, const char *source_dsn); +static bool read_manifest(const char *manifest_path, BidirectionalState *state, + char **subscriber_name_out, char **dbname_out, + char **source_dsn_out); +static void cleanup_partial_state(BidirectionalState *state, const char *subscriber_name, + const char *dbname, const char *source_dsn, + bool force_rm_datadir); +static void append_json_string(PQExpBuffer buf, const char *str); + static PGconn * connectdb(const char *connstr) { @@ -161,6 +201,667 @@ void signal_handler(int sig) } } +/* + * append_json_string + * Append str to buf with JSON string escaping applied (no surrounding + * quotes — caller wraps in "..."). Follows the same convention as + * pg_combinebackup's local escape_json: control characters below 0x20 + * are emitted as \uXXXX. + * + * jsonapi.h provides a parser but no encoder; this local helper is the + * standard frontend pattern (see src/bin/pg_combinebackup/write_manifest.c). + */ +static void +append_json_string(PQExpBuffer buf, const char *str) +{ + const char *p; + + for (p = str; *p; p++) + { + switch (*p) + { + case '\b': appendPQExpBufferStr(buf, "\\b"); break; + case '\f': appendPQExpBufferStr(buf, "\\f"); break; + case '\n': appendPQExpBufferStr(buf, "\\n"); break; + case '\r': appendPQExpBufferStr(buf, "\\r"); break; + case '\t': appendPQExpBufferStr(buf, "\\t"); break; + case '"': appendPQExpBufferStr(buf, "\\\""); break; + case '\\': appendPQExpBufferStr(buf, "\\\\"); break; + default: + if ((unsigned char) *p < 0x20) + appendPQExpBuffer(buf, "\\u%04x", (unsigned char) *p); + else + appendPQExpBufferChar(buf, *p); + break; + } + } +} + +/* + * discover_peer_nodes + * Query source for all peer nodes in the multi-master cluster. + * Returns the peer count; *peers_out is set to a pg_malloc0'd array. + * + * For each peer, sub_name is derived as "sub__" + * and slot_name is obtained via spock.spock_gen_slot_name() on the source. + */ +static int +discover_peer_nodes(PGconn *source_conn, const char *source_node_name, + const char *subscriber_name, const char *dbname, + PeerNodeInfo **peers_out) +{ + static const char *discover_sql = + "SELECT DISTINCT n.node_name, ni.if_dsn" + " FROM spock.subscription s" + " JOIN spock.node n ON s.sub_origin = n.node_id" + " JOIN spock.node_interface ni ON n.node_id = ni.if_nodeid" + " WHERE n.node_name != $1" + " ORDER BY n.node_name"; + const char *paramValues[3]; + PGresult *res; + PGresult *slot_res; + int npeers; + PeerNodeInfo *peers; + int i; + + paramValues[0] = source_node_name; + res = PQexecParams(source_conn, discover_sql, + 1, NULL, paramValues, NULL, NULL, 0); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + die(_("could not discover peer nodes: %s"), + PQerrorMessage(source_conn)); + + npeers = PQntuples(res); + if (npeers == 0) + { + PQclear(res); + die(_("no peer nodes found; source does not appear to be part of a " + "multi-master cluster")); + } + + peers = pg_malloc0(npeers * sizeof(PeerNodeInfo)); + + for (i = 0; i < npeers; i++) + { + PQExpBuffer sub_name_buf = createPQExpBuffer(); + + peers[i].node_name = pg_strdup(PQgetvalue(res, i, 0)); + peers[i].dsn = pg_strdup(PQgetvalue(res, i, 1)); + + appendPQExpBuffer(sub_name_buf, "sub_%s_%s", + subscriber_name, peers[i].node_name); + peers[i].sub_name = pg_strdup(sub_name_buf->data); + destroyPQExpBuffer(sub_name_buf); + + paramValues[0] = dbname; + paramValues[1] = peers[i].node_name; + paramValues[2] = peers[i].sub_name; + slot_res = PQexecParams(source_conn, + "SELECT spock.spock_gen_slot_name" + "($1::name, $2::name, $3::name)", + 3, NULL, paramValues, NULL, NULL, 0); + if (PQresultStatus(slot_res) != PGRES_TUPLES_OK) + die(_("could not generate slot name for peer \"%s\": %s"), + peers[i].node_name, PQerrorMessage(source_conn)); + + peers[i].slot_name = pg_strdup(PQgetvalue(slot_res, 0, 0)); + PQclear(slot_res); + + print_msg(VERBOSITY_VERBOSE, + _(" discovered peer: %s (slot: %s)\n"), + peers[i].node_name, peers[i].slot_name); + } + + PQclear(res); + *peers_out = peers; + return npeers; +} + +/* + * check_preconditions + * Verify that the source cluster and all peers meet the requirements + * for a bidirectional join: Spock >= 6.0.0, track_commit_timestamp on, + * no pending DDL, full-mesh topology, and peer connectivity. + */ +static void +check_preconditions(PGconn *source_conn, PeerNodeInfo *peers, int num_peers) +{ + PGresult *res; + int i; + + /* Spock version gate: require >= 6.0.0 */ + res = PQexec(source_conn, + "SELECT extversion FROM pg_extension WHERE extname = 'spock'"); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + die(_("could not query Spock extension version: %s"), + PQerrorMessage(source_conn)); + if (PQntuples(res) == 0) + die(_("Spock extension is not installed on the source node")); + { + const char *ver = PQgetvalue(res, 0, 0); + int major = 0; + + if (sscanf(ver, "%d.", &major) < 1) + die(_("could not parse Spock version \"%s\""), ver); + if (major < 6) + die(_("Spock version %s on source is too old for bidirectional " + "join; require >= 6.0.0"), ver); + } + PQclear(res); + + /* track_commit_timestamp must be on at the source */ + res = PQexec(source_conn, "SHOW track_commit_timestamp"); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + die(_("could not check track_commit_timestamp: %s"), + PQerrorMessage(source_conn)); + if (strcmp(PQgetvalue(res, 0, 0), "on") != 0) + die(_("track_commit_timestamp must be on for bidirectional join (source)")); + PQclear(res); + + /* No pending DDL in spock.queue */ + res = PQexec(source_conn, "SELECT COUNT(*) FROM spock.queue"); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + die(_("could not check spock.queue: %s"), + PQerrorMessage(source_conn)); + if (strcmp(PQgetvalue(res, 0, 0), "0") != 0) + die(_("pending DDL in spock.queue; wait for replication to drain " + "before joining")); + PQclear(res); + + /* Full-mesh assertion: subscriptions on source == num_peers */ + res = PQexec(source_conn, "SELECT COUNT(*) FROM spock.subscription"); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + die(_("could not count subscriptions: %s"), + PQerrorMessage(source_conn)); + { + int sub_count = atoi(PQgetvalue(res, 0, 0)); + + if (sub_count != num_peers) + die(_("source node has %d active subscription(s) but %d peer(s) " + "discovered; partial-mesh topologies are not supported"), + sub_count, num_peers); + } + PQclear(res); + + /* + * Per-peer: connectivity and track_commit_timestamp. + * + * Spock version is not checked on peers here; peer version checking + * is deferred to the subscription-setup phase (PR3+). + */ + for (i = 0; i < num_peers; i++) + { + PGconn *peer_conn; + + print_msg(VERBOSITY_VERBOSE, + _(" checking peer %s ...\n"), peers[i].node_name); + + peer_conn = PQconnectdb(peers[i].dsn); + if (PQstatus(peer_conn) != CONNECTION_OK) + die(_("cannot connect to peer \"%s\": %s"), + peers[i].node_name, PQerrorMessage(peer_conn)); + + res = PQexec(peer_conn, "SHOW track_commit_timestamp"); + if (PQresultStatus(res) != PGRES_TUPLES_OK) + { + PQclear(res); + PQfinish(peer_conn); + die(_("could not check track_commit_timestamp on peer \"%s\": %s"), + peers[i].node_name, PQerrorMessage(peer_conn)); + } + if (strcmp(PQgetvalue(res, 0, 0), "on") != 0) + { + PQclear(res); + PQfinish(peer_conn); + die(_("track_commit_timestamp must be on for bidirectional join " + "(peer \"%s\")"), peers[i].node_name); + } + PQclear(res); + PQfinish(peer_conn); + } + + print_msg(VERBOSITY_NORMAL, _("Preconditions verified.\n")); +} + +/* + * write_manifest + * Write the bidirectional state manifest to state->manifest_path + * atomically (write to .tmp then rename). + * + * The manifest is a simple hand-formatted JSON file; no parser library + * is required. String values are escaped with append_json_string(). + */ +static void +write_manifest(BidirectionalState *state, const char *subscriber_name, + const char *dbname, const char *source_dsn) +{ + PQExpBuffer buf = createPQExpBuffer(); + char tmp_path[MAXPGPATH]; + FILE *f; + int i; + + snprintf(tmp_path, MAXPGPATH, "%s.tmp", state->manifest_path); + + appendPQExpBufferStr(buf, "{\n"); + appendPQExpBufferStr(buf, " \"version\": 1,\n"); + + appendPQExpBufferStr(buf, " \"subscriber_name\": \""); + append_json_string(buf, subscriber_name); + appendPQExpBufferStr(buf, "\",\n"); + + appendPQExpBufferStr(buf, " \"dbname\": \""); + append_json_string(buf, dbname); + appendPQExpBufferStr(buf, "\",\n"); + + appendPQExpBufferStr(buf, " \"source_dsn\": \""); + append_json_string(buf, source_dsn); + appendPQExpBufferStr(buf, "\",\n"); + + appendPQExpBufferStr(buf, " \"source_slot_name\": \""); + if (state->source_slot_name) + append_json_string(buf, state->source_slot_name); + appendPQExpBufferStr(buf, "\",\n"); + + appendPQExpBufferStr(buf, " \"source_origin_name\": \""); + if (state->source_origin_name) + append_json_string(buf, state->source_origin_name); + appendPQExpBufferStr(buf, "\",\n"); + + appendPQExpBufferStr(buf, " \"peers\": [\n"); + for (i = 0; i < state->num_peers; i++) + { + PeerNodeInfo *p = &state->peers[i]; + bool last = (i == state->num_peers - 1); + + appendPQExpBufferStr(buf, " {\n"); + + appendPQExpBufferStr(buf, " \"node_name\": \""); + append_json_string(buf, p->node_name); + appendPQExpBufferStr(buf, "\",\n"); + + appendPQExpBufferStr(buf, " \"peer_dsn\": \""); + append_json_string(buf, p->dsn); + appendPQExpBufferStr(buf, "\",\n"); + + appendPQExpBufferStr(buf, " \"sub_name_on_n3\": \""); + append_json_string(buf, p->sub_name); + appendPQExpBufferStr(buf, "\",\n"); + + appendPQExpBufferStr(buf, " \"peer_slot_name\": \""); + append_json_string(buf, p->slot_name); + appendPQExpBufferStr(buf, "\"\n"); + + appendPQExpBufferStr(buf, last ? " }\n" : " },\n"); + } + appendPQExpBufferStr(buf, " ]\n"); + appendPQExpBufferStr(buf, "}\n"); + + f = fopen(tmp_path, "w"); + if (f == NULL) + die(_("could not create manifest file \"%s\": %s"), + tmp_path, strerror(errno)); + + if (fwrite(buf->data, 1, buf->len, f) != buf->len) + { + fclose(f); + unlink(tmp_path); + die(_("could not write manifest file \"%s\": %s"), + tmp_path, strerror(errno)); + } + if (fclose(f) != 0) + { + unlink(tmp_path); + die(_("could not close manifest file \"%s\": %s"), + tmp_path, strerror(errno)); + } + if (rename(tmp_path, state->manifest_path) != 0) + die(_("could not rename manifest to \"%s\": %s"), + state->manifest_path, strerror(errno)); + + destroyPQExpBuffer(buf); +} + +/* + * Semantic-action state for read_manifest(). Passed as void *semstate to all + * pg_parse_json callbacks; tracks nesting depth and accumulates field values. + */ +typedef struct ManifestParseState +{ + /* outputs written by scalar callback */ + char **subscriber_name_out; + char **dbname_out; + char **source_dsn_out; + BidirectionalState *bidir; + + /* parser context */ + int depth; /* object/array nesting depth */ + bool in_peers; /* inside the top-level "peers" array */ + bool in_peer_obj; /* inside one peer object */ + char *cur_field; /* current object field name (owned by us) */ + + /* per-peer accumulator, flushed on each object_end inside peers */ + char *peer_node_name; + char *peer_dsn; + char *peer_sub_name; + char *peer_slot_name; + int peer_capacity; +} ManifestParseState; + +static JsonParseErrorType +manifest_object_start(void *st) +{ + ManifestParseState *s = (ManifestParseState *) st; + + s->depth++; + if (s->in_peers && s->depth == 3) + s->in_peer_obj = true; + return JSON_SUCCESS; +} + +static JsonParseErrorType +manifest_object_end(void *st) +{ + ManifestParseState *s = (ManifestParseState *) st; + + if (s->in_peer_obj && s->depth == 3) + { + int i = s->bidir->num_peers; + + if (i >= s->peer_capacity) + { + s->peer_capacity = (s->peer_capacity > 0) ? s->peer_capacity * 2 : 4; + s->bidir->peers = pg_realloc(s->bidir->peers, + s->peer_capacity * sizeof(PeerNodeInfo)); + } + s->bidir->peers[i].node_name = s->peer_node_name; + s->bidir->peers[i].dsn = s->peer_dsn; + s->bidir->peers[i].sub_name = s->peer_sub_name; + s->bidir->peers[i].slot_name = s->peer_slot_name; + s->bidir->num_peers++; + s->peer_node_name = s->peer_dsn = s->peer_sub_name = s->peer_slot_name = NULL; + s->in_peer_obj = false; + } + s->depth--; + return JSON_SUCCESS; +} + +static JsonParseErrorType +manifest_array_start(void *st) +{ + ManifestParseState *s = (ManifestParseState *) st; + + s->depth++; + if (s->depth == 2 && s->cur_field != NULL && + strcmp(s->cur_field, "peers") == 0) + s->in_peers = true; + return JSON_SUCCESS; +} + +static JsonParseErrorType +manifest_array_end(void *st) +{ + ManifestParseState *s = (ManifestParseState *) st; + + if (s->in_peers && s->depth == 2) + s->in_peers = false; + s->depth--; + return JSON_SUCCESS; +} + +static JsonParseErrorType +manifest_ofield_start(void *st, char *fname, bool isnull) +{ + ManifestParseState *s = (ManifestParseState *) st; + + (void) isnull; + pg_free(s->cur_field); + s->cur_field = pg_strdup(fname); + pg_free(fname); /* callback owns the token */ + return JSON_SUCCESS; +} + +static JsonParseErrorType +manifest_scalar(void *st, char *token, JsonTokenType tokentype) +{ + ManifestParseState *s = (ManifestParseState *) st; + + if (s->cur_field == NULL || tokentype != JSON_TOKEN_STRING) + { + pg_free(token); + return JSON_SUCCESS; + } + + if (!s->in_peer_obj) + { + /* top-level scalar fields */ + if (strcmp(s->cur_field, "subscriber_name") == 0) + *s->subscriber_name_out = token; + else if (strcmp(s->cur_field, "dbname") == 0) + *s->dbname_out = token; + else if (strcmp(s->cur_field, "source_dsn") == 0) + *s->source_dsn_out = token; + else if (strcmp(s->cur_field, "source_slot_name") == 0) + s->bidir->source_slot_name = token; + else if (strcmp(s->cur_field, "source_origin_name") == 0) + s->bidir->source_origin_name = token; + else + pg_free(token); + } + else + { + /* per-peer scalar fields */ + if (strcmp(s->cur_field, "node_name") == 0) + s->peer_node_name = token; + else if (strcmp(s->cur_field, "peer_dsn") == 0) + s->peer_dsn = token; + else if (strcmp(s->cur_field, "sub_name_on_n3") == 0) + s->peer_sub_name = token; + else if (strcmp(s->cur_field, "peer_slot_name") == 0) + s->peer_slot_name = token; + else + pg_free(token); + } + return JSON_SUCCESS; +} + +/* + * read_manifest + * Read the bidirectional manifest from manifest_path. + * + * Returns false if the file does not exist (nothing to clean up). + * Dies if the file exists but cannot be read or is malformed. + * On success, sets *subscriber_name_out, *dbname_out, *source_dsn_out, + * and populates state->peers[]. + * + * Uses pg_parse_json (common/jsonapi.h) for correct JSON lexing, which + * handles string quoting, escape sequences, and nesting transparently. + */ +static bool +read_manifest(const char *manifest_path, BidirectionalState *state, + char **subscriber_name_out, char **dbname_out, + char **source_dsn_out) +{ + struct stat st; + char *content; + FILE *f; + JsonLexContext *lex; + JsonSemAction sem; + ManifestParseState pstate; + JsonParseErrorType result; + + if (stat(manifest_path, &st) != 0) + return false; + + content = pg_malloc(st.st_size + 1); + f = fopen(manifest_path, "r"); + if (f == NULL) + die(_("could not open manifest file \"%s\": %s"), + manifest_path, strerror(errno)); + + if ((size_t) fread(content, 1, st.st_size, f) != (size_t) st.st_size) + { + fclose(f); + die(_("could not read manifest file \"%s\": %s"), + manifest_path, strerror(errno)); + } + content[st.st_size] = '\0'; + fclose(f); + + memset(&pstate, 0, sizeof(pstate)); + pstate.subscriber_name_out = subscriber_name_out; + pstate.dbname_out = dbname_out; + pstate.source_dsn_out = source_dsn_out; + pstate.bidir = state; + + memset(&sem, 0, sizeof(sem)); + sem.semstate = &pstate; + sem.object_start = manifest_object_start; + sem.object_end = manifest_object_end; + sem.array_start = manifest_array_start; + sem.array_end = manifest_array_end; + sem.object_field_start = manifest_ofield_start; + sem.scalar = manifest_scalar; + + lex = makeJsonLexContextCstringLen(NULL, content, st.st_size, + PG_UTF8, true); + result = pg_parse_json(lex, &sem); + pg_free(content); + pg_free(pstate.cur_field); + + if (result != JSON_SUCCESS) + { + char *detail = json_errdetail(result, lex); + + freeJsonLexContext(lex); + die(_("manifest file \"%s\" is malformed: %s"), manifest_path, detail); + } + freeJsonLexContext(lex); + + if (!*subscriber_name_out || !*dbname_out || !*source_dsn_out) + die(_("manifest file \"%s\" is malformed or missing required fields"), + manifest_path); + + return true; +} + +/* + * cleanup_partial_state + * Idempotently remove bidirectional join state from all reachable nodes. + * + * Connects to the source and each peer; drops replication slots and + * reverse subscriptions that were created during a previous join attempt. + * All operations are best-effort: connectivity failures are logged as + * warnings rather than being fatal. + */ +static void +cleanup_partial_state(BidirectionalState *state, const char *subscriber_name, + const char *dbname, const char *source_dsn, + bool force_rm_datadir) +{ + PGconn *source_conn; + PGresult *res; + PQExpBuffer query = createPQExpBuffer(); + int i; + + print_msg(VERBOSITY_NORMAL, + _("Cleaning up partial bidirectional join state ...\n")); + + source_conn = PQconnectdb(source_dsn); + if (PQstatus(source_conn) != CONNECTION_OK) + { + print_msg(VERBOSITY_NORMAL, + _("warning: cannot connect to source node; skipping " + "source-side cleanup: %s\n"), + PQerrorMessage(source_conn)); + PQfinish(source_conn); + source_conn = NULL; + } + + /* Drop source replication slot if it was created */ + if (source_conn && state->source_slot_name && state->source_slot_name[0]) + { + printfPQExpBuffer(query, + "SELECT pg_drop_replication_slot(slot_name)" + " FROM pg_replication_slots" + " WHERE slot_name = '%s'", + state->source_slot_name); + res = PQexec(source_conn, query->data); + if (PQresultStatus(res) == PGRES_TUPLES_OK && PQntuples(res) > 0) + print_msg(VERBOSITY_NORMAL, + _(" dropped source slot %s\n"), + state->source_slot_name); + PQclear(res); + } + + /* Per-peer: drop slot and any reverse subscription */ + for (i = 0; i < state->num_peers; i++) + { + PeerNodeInfo *peer = &state->peers[i]; + PGconn *peer_conn; + char reverse_sub[NAMEDATALEN]; + + if (!peer->dsn || !peer->dsn[0]) + continue; + + peer_conn = PQconnectdb(peer->dsn); + if (PQstatus(peer_conn) != CONNECTION_OK) + { + print_msg(VERBOSITY_NORMAL, + _("warning: cannot connect to peer \"%s\"; skipping " + "peer-side cleanup: %s\n"), + peer->node_name, PQerrorMessage(peer_conn)); + PQfinish(peer_conn); + continue; + } + + if (peer->slot_name && peer->slot_name[0]) + { + printfPQExpBuffer(query, + "SELECT pg_drop_replication_slot(slot_name)" + " FROM pg_replication_slots" + " WHERE slot_name = '%s'", + peer->slot_name); + res = PQexec(peer_conn, query->data); + if (PQresultStatus(res) == PGRES_TUPLES_OK && PQntuples(res) > 0) + print_msg(VERBOSITY_NORMAL, + _(" dropped peer slot %s on %s\n"), + peer->slot_name, peer->node_name); + PQclear(res); + } + + /* + * Drop the reverse subscription (peer -> new subscriber) if it was + * created during a previous attempt. The sub_drop second argument + * is ifexists=true. + */ + snprintf(reverse_sub, sizeof(reverse_sub), "sub_%s_%s", + peer->node_name, subscriber_name); + printfPQExpBuffer(query, + "SELECT spock.sub_drop('%s', true)", + reverse_sub); + res = PQexec(peer_conn, query->data); + PQclear(res); + + PQfinish(peer_conn); + print_msg(VERBOSITY_NORMAL, + _(" cleaned up peer %s\n"), peer->node_name); + } + + if (source_conn) + PQfinish(source_conn); + + destroyPQExpBuffer(query); + + if (state->manifest_path && state->manifest_path[0]) + { + unlink(state->manifest_path); + print_msg(VERBOSITY_NORMAL, + _(" removed manifest %s\n"), state->manifest_path); + } + + print_msg(VERBOSITY_NORMAL, _("Cleanup complete.\n")); +} + int main(int argc, char **argv) @@ -194,6 +895,8 @@ main(int argc, char **argv) logfd; char *restore_point_name = NULL; char *extra_basebackup_args = NULL; + BidirectionalState bidir = {0}; + char bidir_manifest_path[MAXPGPATH] = {0}; static struct option long_options[] = { {"subscriber-name", required_argument, NULL, 'n'}, @@ -210,6 +913,10 @@ main(int argc, char **argv) {"databases", required_argument, NULL, 9}, {"extra-basebackup-args", required_argument, NULL, 10}, {"text-types", no_argument, NULL, 11}, + {"bidirectional", no_argument, NULL, 12}, + {"stall-timeout", required_argument, NULL, 13}, + {"max-wait", required_argument, NULL, 14}, + {"cleanup", no_argument, NULL, 15}, {NULL, 0, NULL, 0} }; @@ -299,6 +1006,22 @@ main(int argc, char **argv) case 11: force_text_transfer = true; break; + case 12: + bidir.enabled = true; + break; + case 13: + bidir.stall_timeout = atoi(optarg); + if (bidir.stall_timeout <= 0) + die(_("--stall-timeout must be a positive integer")); + break; + case 14: + bidir.max_wait = atoi(optarg); + if (bidir.max_wait < 0) + die(_("--max-wait must be a non-negative integer")); + break; + case 15: + bidir.cleanup_mode = true; + break; default: fprintf(stderr, _("Unknown option\n")); fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname); @@ -316,16 +1039,20 @@ main(int argc, char **argv) fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname); exit(1); } - else if (subscriber_name == NULL) + else if (subscriber_name == NULL && !bidir.cleanup_mode) { fprintf(stderr, _("No subscriber name specified\n")); fprintf(stderr, _("Try \"%s --help\" for more information.\n"), progname); exit(1); } - if (!base_prov_connstr || !strlen(base_prov_connstr)) + if (bidir.cleanup_mode && !bidir.enabled) + die(_("--cleanup requires --bidirectional.\n")); + + if (!bidir.cleanup_mode && (!base_prov_connstr || !strlen(base_prov_connstr))) die(_("Provider connection string must be specified.\n")); - if (!base_sub_connstr || !strlen(base_sub_connstr)) + if (!bidir.enabled && !bidir.cleanup_mode && + (!base_sub_connstr || !strlen(base_sub_connstr))) die(_("Subscriber connection string must be specified.\n")); if (apply_delay < 0) @@ -337,6 +1064,33 @@ main(int argc, char **argv) if (!replication_sets || !strlen(replication_sets)) replication_sets = "default,default_insert_only,ddl_sql"; + /* Build the manifest path from --pgdata */ + if (bidir.enabled || bidir.cleanup_mode) + { + snprintf(bidir_manifest_path, MAXPGPATH, + "%s/spock_bidirectional_manifest.json", data_dir); + bidir.manifest_path = bidir_manifest_path; + if (bidir.stall_timeout == 0) + bidir.stall_timeout = 600; + } + + /* --cleanup: read manifest, remove partial state, exit */ + if (bidir.cleanup_mode) + { + char *sub_name = NULL; + char *db = NULL; + char *src_dsn = NULL; + + if (!read_manifest(bidir.manifest_path, &bidir, &sub_name, &db, &src_dsn)) + { + fprintf(stderr, _("No manifest found at %s; nothing to clean up.\n"), + bidir.manifest_path); + exit(0); + } + cleanup_partial_state(&bidir, sub_name, db, src_dsn, false); + exit(0); + } + /* Init random numbers used for slot suffixes, etc */ srand(time(NULL)); @@ -372,9 +1126,12 @@ main(int argc, char **argv) if (!prov_connstr || !strlen(prov_connstr)) die(_("Provider connection string is not valid.\n")); - sub_connstr = get_connstr(base_sub_connstr, db); - if (!sub_connstr || !strlen(sub_connstr)) - die(_("Subscriber connection string is not valid.\n")); + if (!bidir.enabled) + { + sub_connstr = get_connstr(base_sub_connstr, db); + if (!sub_connstr || !strlen(sub_connstr)) + die(_("Subscriber connection string is not valid.\n")); + } } /* @@ -408,6 +1165,28 @@ main(int argc, char **argv) provider_conn = connectdb(prov_connstr); remote_info = get_remote_info(provider_conn); + /* + * --bidirectional: discover peers, verify preconditions, write the + * manifest, then exit. PR3+ continues from here after the physical + * backup has been taken and the subscriber is running. + */ + if (bidir.enabled) + { + bidir.num_peers = discover_peer_nodes(provider_conn, + remote_info->node_name, + subscriber_name, db, + &bidir.peers); + check_preconditions(provider_conn, bidir.peers, bidir.num_peers); + write_manifest(&bidir, subscriber_name, db, base_prov_connstr); + print_msg(VERBOSITY_NORMAL, + _("Bidirectional plumbing complete: %d peer(s) discovered, " + "preconditions OK, manifest written to %s.\n"), + bidir.num_peers, bidir.manifest_path); + PQfinish(provider_conn); + provider_conn = NULL; + exit(0); + } + /* only need to do this piece once */ if (dbnum == 0) From 401476f33fcbaddd32a33ebe0ae22ff201938b95 Mon Sep 17 00:00:00 2001 From: Asif Rehman Date: Mon, 13 Jul 2026 19:00:47 +0500 Subject: [PATCH 7/7] spock_create_subscriber: add TAP test for --bidirectional plumbing --- tests/tap/schedule | 2 + tests/tap/t/047_bidir_plumbing.pl | 165 ++++++++++++++++++++++++++++++ 2 files changed, 167 insertions(+) create mode 100644 tests/tap/t/047_bidir_plumbing.pl diff --git a/tests/tap/schedule b/tests/tap/schedule index 42e9bea6d..5d82eba9f 100644 --- a/tests/tap/schedule +++ b/tests/tap/schedule @@ -51,6 +51,7 @@ test: 025_tiebreaker_equal_warning test: 030_autoddl_repset_stickiness test: 032_lolor_largeobject_repset test: 044_apply_change_logging +test: 047_bidir_plumbing # Upgrade schema match test (builds from source, slow): #test: 018_upgrade_schema_match # @@ -58,3 +59,4 @@ test: 044_apply_change_logging # Regression tests test: 103_manager_worker_dboid_race test: 105_sub_disable_retransmit_after_disconnect + diff --git a/tests/tap/t/047_bidir_plumbing.pl b/tests/tap/t/047_bidir_plumbing.pl new file mode 100644 index 000000000..f693e4433 --- /dev/null +++ b/tests/tap/t/047_bidir_plumbing.pl @@ -0,0 +1,165 @@ +#!/usr/bin/perl +# ============================================================================= +# Test: 047_bidir_plumbing.pl — spock_create_subscriber --bidirectional (PR2) +# ============================================================================= +# Validates the PR2 plumbing phase of the SPOC-601 bidirectional node-join +# procedure. The test does NOT start a third PostgreSQL instance; it only +# exercises the utility's plumbing phase against an existing 2-node cluster: +# +# --bidirectional discover peers, check preconditions, write manifest +# --cleanup idempotently remove partial state / manifest +# +# Topology: +# n1 <-> n2 (full bidirectional Spock subscriptions, track_commit_timestamp=on) +# +# The utility is run with --pgdata pointing at a plain temp directory (no PG +# cluster) that exists solely to hold the manifest file. +# +# Test count breakdown: +# 1 binary found +# 1 temp pgdata created +# 5 create_cluster(2) [2 pg_isready + 2 spock checks + 1 pass] +# 1 cross_wire n1<->n2 +# 1 --bidirectional exits 0 +# 1 manifest file written +# 1 manifest: version 1 +# 1 manifest: subscriber_name n3 +# 1 manifest: dbname regression +# 1 manifest: source_dsn present +# 1 manifest: peer n2 listed +# 1 manifest: peer_slot_name present +# 1 --cleanup exits 0 +# 1 manifest removed +# 1 --cleanup with no manifest exits 0 (idempotent) +# 1 destroy_cluster +# --- +# 20 total +# ============================================================================= + +use strict; +use warnings; +use Test::More tests => 20; +use File::Path qw(remove_tree make_path); +use lib '.'; +use SpockTest qw(create_cluster cross_wire destroy_cluster system_or_bail + command_ok system_maybe get_test_config scalar_query psql_or_bail); + +# ============================================================================= +# Locate spock_create_subscriber binary +# ============================================================================= +my $SCS_BIN; +for my $dir (split(':', $ENV{PATH} // '')) { + my $c = "$dir/spock_create_subscriber"; + if (-x $c) { $SCS_BIN = $c; last; } +} +unless (defined $SCS_BIN) { + # Fall back to the build tree (CWD is tests/tap/ during make check_prove) + my $bt = '../../utils/spock_create_subscriber/spock_create_subscriber'; + $SCS_BIN = $bt if -x $bt; +} +BAIL_OUT("spock_create_subscriber binary not found; run 'make install' first") + unless defined $SCS_BIN; +pass("spock_create_subscriber binary found"); + +# ============================================================================= +# Scratch directory that stands in for n3's future PGDATA. +# It just needs to exist so the manifest can be written there. +# ============================================================================= +my $N3_PGDATA = '/tmp/spock_bidir_test_n3_pgdata'; +my $MANIFEST = "$N3_PGDATA/spock_bidirectional_manifest.json"; + +remove_tree($N3_PGDATA) if -d $N3_PGDATA; +make_path($N3_PGDATA) + or BAIL_OUT("could not create temp pgdata dir: $N3_PGDATA"); +pass("temp pgdata dir for n3 created"); + +# ============================================================================= +# SETUP: 2-node cluster, cross-wired bidirectionally +# create_cluster counts as 5 tests (pg_isready + spock check per node + pass) +# ============================================================================= +create_cluster(2, 'Create bidirectional 2-node cluster'); + +my $config = get_test_config(); +my $node_ports = $config->{node_ports}; +my $dbname = $config->{db_name}; +my $host = $config->{host}; +my $db_user = $config->{db_user}; +my $db_password = $config->{db_password}; + +my $n1_dsn = "host=$host port=$node_ports->[0] dbname=$dbname" + . " user=$db_user password=$db_password"; + +# Create bidirectional subscriptions n1->n2 and n2->n1 (1 test) +cross_wire(2, ['n1', 'n2'], 'Cross-wire n1 <-> n2 bidirectionally'); + +# ============================================================================= +# TEST: --bidirectional mode +# Discovers peer n2 from n1, checks preconditions, writes manifest. +# ============================================================================= + +command_ok( + [ $SCS_BIN, + '--bidirectional', + '--pgdata', $N3_PGDATA, + '--subscriber-name', 'n3', + '--provider-dsn', $n1_dsn, + ], + '--bidirectional plumbing exits 0' +); + +ok(-f $MANIFEST, + 'manifest written to /spock_bidirectional_manifest.json'); + +# Read and inspect manifest content +my $manifest_content = ''; +if (-f $MANIFEST) { + open my $fh, '<', $MANIFEST or die "Cannot read manifest: $!"; + local $/; + $manifest_content = <$fh>; + close $fh; +} + +like($manifest_content, qr/"version":\s*1/, + 'manifest: version is 1'); +like($manifest_content, qr/"subscriber_name":\s*"n3"/, + 'manifest: subscriber_name is n3'); +like($manifest_content, qr/"dbname":\s*"$dbname"/, + 'manifest: dbname matches provider dbname'); +ok(index($manifest_content, '"source_dsn":') >= 0, + 'manifest: source_dsn field present'); +like($manifest_content, qr/"node_name":\s*"n2"/, + 'manifest: peer n2 is listed in peers array'); +ok(index($manifest_content, '"peer_slot_name":') >= 0, + 'manifest: peer_slot_name field present'); + +# ============================================================================= +# TEST: --cleanup mode — removes manifest, exits 0 +# ============================================================================= + +command_ok( + [ $SCS_BIN, + '--bidirectional', + '--cleanup', + '--pgdata', $N3_PGDATA, + ], + '--cleanup with manifest exits 0' +); + +ok(!-f $MANIFEST, + 'manifest file removed by --cleanup'); + +# Second cleanup with no manifest must also exit 0 (idempotent) +command_ok( + [ $SCS_BIN, + '--bidirectional', + '--cleanup', + '--pgdata', $N3_PGDATA, + ], + '--cleanup with no manifest exits 0 (idempotent)' +); + +# ============================================================================= +# CLEANUP +# ============================================================================= +remove_tree($N3_PGDATA); +destroy_cluster('Cleanup');