|           Line data    Source code 
       1             : /*-------------------------------------------------------------------------
       2             :  *
       3             :  * pg_createsubscriber.c
       4             :  *    Create a new logical replica from a standby server
       5             :  *
       6             :  * Copyright (c) 2024-2025, PostgreSQL Global Development Group
       7             :  *
       8             :  * IDENTIFICATION
       9             :  *    src/bin/pg_basebackup/pg_createsubscriber.c
      10             :  *
      11             :  *-------------------------------------------------------------------------
      12             :  */
      13             : 
      14             : #include "postgres_fe.h"
      15             : 
      16             : #include <sys/stat.h>
      17             : #include <sys/time.h>
      18             : #include <sys/wait.h>
      19             : #include <time.h>
      20             : 
      21             : #include "common/connect.h"
      22             : #include "common/controldata_utils.h"
      23             : #include "common/logging.h"
      24             : #include "common/pg_prng.h"
      25             : #include "common/restricted_token.h"
      26             : #include "fe_utils/recovery_gen.h"
      27             : #include "fe_utils/simple_list.h"
      28             : #include "fe_utils/string_utils.h"
      29             : #include "fe_utils/version.h"
      30             : #include "getopt_long.h"
      31             : 
      32             : #define DEFAULT_SUB_PORT    "50432"
      33             : #define OBJECTTYPE_PUBLICATIONS  0x0001
      34             : 
      35             : /* Command-line options */
      36             : struct CreateSubscriberOptions
      37             : {
      38             :     char       *config_file;    /* configuration file */
      39             :     char       *pub_conninfo_str;   /* publisher connection string */
      40             :     char       *socket_dir;     /* directory for Unix-domain socket, if any */
      41             :     char       *sub_port;       /* subscriber port number */
      42             :     const char *sub_username;   /* subscriber username */
      43             :     bool        two_phase;      /* enable-two-phase option */
      44             :     SimpleStringList database_names;    /* list of database names */
      45             :     SimpleStringList pub_names; /* list of publication names */
      46             :     SimpleStringList sub_names; /* list of subscription names */
      47             :     SimpleStringList replslot_names;    /* list of replication slot names */
      48             :     int         recovery_timeout;   /* stop recovery after this time */
      49             :     bool        all_dbs;        /* all option */
      50             :     SimpleStringList objecttypes_to_clean;  /* list of object types to cleanup */
      51             : };
      52             : 
      53             : /* per-database publication/subscription info */
      54             : struct LogicalRepInfo
      55             : {
      56             :     char       *dbname;         /* database name */
      57             :     char       *pubconninfo;    /* publisher connection string */
      58             :     char       *subconninfo;    /* subscriber connection string */
      59             :     char       *pubname;        /* publication name */
      60             :     char       *subname;        /* subscription name */
      61             :     char       *replslotname;   /* replication slot name */
      62             : 
      63             :     bool        made_replslot;  /* replication slot was created */
      64             :     bool        made_publication;   /* publication was created */
      65             : };
      66             : 
      67             : /*
      68             :  * Information shared across all the databases (or publications and
      69             :  * subscriptions).
      70             :  */
      71             : struct LogicalRepInfos
      72             : {
      73             :     struct LogicalRepInfo *dbinfo;
      74             :     bool        two_phase;      /* enable-two-phase option */
      75             :     bits32      objecttypes_to_clean;   /* flags indicating which object types
      76             :                                          * to clean up on subscriber */
      77             : };
      78             : 
      79             : static void cleanup_objects_atexit(void);
      80             : static void usage();
      81             : static char *get_base_conninfo(const char *conninfo, char **dbname);
      82             : static char *get_sub_conninfo(const struct CreateSubscriberOptions *opt);
      83             : static char *get_exec_path(const char *argv0, const char *progname);
      84             : static void check_data_directory(const char *datadir);
      85             : static char *concat_conninfo_dbname(const char *conninfo, const char *dbname);
      86             : static struct LogicalRepInfo *store_pub_sub_info(const struct CreateSubscriberOptions *opt,
      87             :                                                  const char *pub_base_conninfo,
      88             :                                                  const char *sub_base_conninfo);
      89             : static PGconn *connect_database(const char *conninfo, bool exit_on_error);
      90             : static void disconnect_database(PGconn *conn, bool exit_on_error);
      91             : static uint64 get_primary_sysid(const char *conninfo);
      92             : static uint64 get_standby_sysid(const char *datadir);
      93             : static void modify_subscriber_sysid(const struct CreateSubscriberOptions *opt);
      94             : static bool server_is_in_recovery(PGconn *conn);
      95             : static char *generate_object_name(PGconn *conn);
      96             : static void check_publisher(const struct LogicalRepInfo *dbinfo);
      97             : static char *setup_publisher(struct LogicalRepInfo *dbinfo);
      98             : static void check_subscriber(const struct LogicalRepInfo *dbinfo);
      99             : static void setup_subscriber(struct LogicalRepInfo *dbinfo,
     100             :                              const char *consistent_lsn);
     101             : static void setup_recovery(const struct LogicalRepInfo *dbinfo, const char *datadir,
     102             :                            const char *lsn);
     103             : static void drop_primary_replication_slot(struct LogicalRepInfo *dbinfo,
     104             :                                           const char *slotname);
     105             : static void drop_failover_replication_slots(struct LogicalRepInfo *dbinfo);
     106             : static char *create_logical_replication_slot(PGconn *conn,
     107             :                                              struct LogicalRepInfo *dbinfo);
     108             : static void drop_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo,
     109             :                                   const char *slot_name);
     110             : static void pg_ctl_status(const char *pg_ctl_cmd, int rc);
     111             : static void start_standby_server(const struct CreateSubscriberOptions *opt,
     112             :                                  bool restricted_access,
     113             :                                  bool restrict_logical_worker);
     114             : static void stop_standby_server(const char *datadir);
     115             : static void wait_for_end_recovery(const char *conninfo,
     116             :                                   const struct CreateSubscriberOptions *opt);
     117             : static void create_publication(PGconn *conn, struct LogicalRepInfo *dbinfo);
     118             : static void drop_publication(PGconn *conn, const char *pubname,
     119             :                              const char *dbname, bool *made_publication);
     120             : static void check_and_drop_publications(PGconn *conn, struct LogicalRepInfo *dbinfo);
     121             : static void create_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo);
     122             : static void set_replication_progress(PGconn *conn, const struct LogicalRepInfo *dbinfo,
     123             :                                      const char *lsn);
     124             : static void enable_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo);
     125             : static void check_and_drop_existing_subscriptions(PGconn *conn,
     126             :                                                   const struct LogicalRepInfo *dbinfo);
     127             : static void drop_existing_subscriptions(PGconn *conn, const char *subname,
     128             :                                         const char *dbname);
     129             : static void get_publisher_databases(struct CreateSubscriberOptions *opt,
     130             :                                     bool dbnamespecified);
     131             : 
     132             : #define USEC_PER_SEC    1000000
     133             : #define WAIT_INTERVAL   1       /* 1 second */
     134             : 
     135             : static const char *progname;
     136             : 
     137             : static char *primary_slot_name = NULL;
     138             : static bool dry_run = false;
     139             : 
     140             : static bool success = false;
     141             : 
     142             : static struct LogicalRepInfos dbinfos;
     143             : static int  num_dbs = 0;        /* number of specified databases */
     144             : static int  num_pubs = 0;       /* number of specified publications */
     145             : static int  num_subs = 0;       /* number of specified subscriptions */
     146             : static int  num_replslots = 0;  /* number of specified replication slots */
     147             : 
     148             : static pg_prng_state prng_state;
     149             : 
     150             : static char *pg_ctl_path = NULL;
     151             : static char *pg_resetwal_path = NULL;
     152             : 
     153             : /* standby / subscriber data directory */
     154             : static char *subscriber_dir = NULL;
     155             : 
     156             : static bool recovery_ended = false;
     157             : static bool standby_running = false;
     158             : 
     159             : enum WaitPMResult
     160             : {
     161             :     POSTMASTER_READY,
     162             :     POSTMASTER_STILL_STARTING
     163             : };
     164             : 
     165             : 
     166             : /*
     167             :  * Cleanup objects that were created by pg_createsubscriber if there is an
     168             :  * error.
     169             :  *
     170             :  * Publications and replication slots are created on primary. Depending on the
     171             :  * step it failed, it should remove the already created objects if it is
     172             :  * possible (sometimes it won't work due to a connection issue).
     173             :  * There is no cleanup on the target server. The steps on the target server are
     174             :  * executed *after* promotion, hence, at this point, a failure means recreate
     175             :  * the physical replica and start again.
     176             :  */
     177             : static void
     178          20 : cleanup_objects_atexit(void)
     179             : {
     180          20 :     if (success)
     181           8 :         return;
     182             : 
     183             :     /*
     184             :      * If the server is promoted, there is no way to use the current setup
     185             :      * again. Warn the user that a new replication setup should be done before
     186             :      * trying again.
     187             :      */
     188          12 :     if (recovery_ended)
     189             :     {
     190           0 :         pg_log_warning("failed after the end of recovery");
     191           0 :         pg_log_warning_hint("The target server cannot be used as a physical replica anymore.  "
     192             :                             "You must recreate the physical replica before continuing.");
     193             :     }
     194             : 
     195          36 :     for (int i = 0; i < num_dbs; i++)
     196             :     {
     197          24 :         struct LogicalRepInfo *dbinfo = &dbinfos.dbinfo[i];
     198             : 
     199          24 :         if (dbinfo->made_publication || dbinfo->made_replslot)
     200             :         {
     201             :             PGconn     *conn;
     202             : 
     203           0 :             conn = connect_database(dbinfo->pubconninfo, false);
     204           0 :             if (conn != NULL)
     205             :             {
     206           0 :                 if (dbinfo->made_publication)
     207           0 :                     drop_publication(conn, dbinfo->pubname, dbinfo->dbname,
     208             :                                      &dbinfo->made_publication);
     209           0 :                 if (dbinfo->made_replslot)
     210           0 :                     drop_replication_slot(conn, dbinfo, dbinfo->replslotname);
     211           0 :                 disconnect_database(conn, false);
     212             :             }
     213             :             else
     214             :             {
     215             :                 /*
     216             :                  * If a connection could not be established, inform the user
     217             :                  * that some objects were left on primary and should be
     218             :                  * removed before trying again.
     219             :                  */
     220           0 :                 if (dbinfo->made_publication)
     221             :                 {
     222           0 :                     pg_log_warning("publication \"%s\" created in database \"%s\" on primary was left behind",
     223             :                                    dbinfo->pubname,
     224             :                                    dbinfo->dbname);
     225           0 :                     pg_log_warning_hint("Drop this publication before trying again.");
     226             :                 }
     227           0 :                 if (dbinfo->made_replslot)
     228             :                 {
     229           0 :                     pg_log_warning("replication slot \"%s\" created in database \"%s\" on primary was left behind",
     230             :                                    dbinfo->replslotname,
     231             :                                    dbinfo->dbname);
     232           0 :                     pg_log_warning_hint("Drop this replication slot soon to avoid retention of WAL files.");
     233             :                 }
     234             :             }
     235             :         }
     236             :     }
     237             : 
     238          12 :     if (standby_running)
     239           8 :         stop_standby_server(subscriber_dir);
     240             : }
     241             : 
     242             : static void
     243           2 : usage(void)
     244             : {
     245           2 :     printf(_("%s creates a new logical replica from a standby server.\n\n"),
     246             :            progname);
     247           2 :     printf(_("Usage:\n"));
     248           2 :     printf(_("  %s [OPTION]...\n"), progname);
     249           2 :     printf(_("\nOptions:\n"));
     250           2 :     printf(_("  -a, --all                       create subscriptions for all databases except template\n"
     251             :              "                                  databases and databases that don't allow connections\n"));
     252           2 :     printf(_("  -d, --database=DBNAME           database in which to create a subscription\n"));
     253           2 :     printf(_("  -D, --pgdata=DATADIR            location for the subscriber data directory\n"));
     254           2 :     printf(_("  -n, --dry-run                   dry run, just show what would be done\n"));
     255           2 :     printf(_("  -p, --subscriber-port=PORT      subscriber port number (default %s)\n"), DEFAULT_SUB_PORT);
     256           2 :     printf(_("  -P, --publisher-server=CONNSTR  publisher connection string\n"));
     257           2 :     printf(_("  -s, --socketdir=DIR             socket directory to use (default current dir.)\n"));
     258           2 :     printf(_("  -t, --recovery-timeout=SECS     seconds to wait for recovery to end\n"));
     259           2 :     printf(_("  -T, --enable-two-phase          enable two-phase commit for all subscriptions\n"));
     260           2 :     printf(_("  -U, --subscriber-username=NAME  user name for subscriber connection\n"));
     261           2 :     printf(_("  -v, --verbose                   output verbose messages\n"));
     262           2 :     printf(_("      --clean=OBJECTTYPE          drop all objects of the specified type from specified\n"
     263             :              "                                  databases on the subscriber; accepts: \"%s\"\n"), "publications");
     264           2 :     printf(_("      --config-file=FILENAME      use specified main server configuration\n"
     265             :              "                                  file when running target cluster\n"));
     266           2 :     printf(_("      --publication=NAME          publication name\n"));
     267           2 :     printf(_("      --replication-slot=NAME     replication slot name\n"));
     268           2 :     printf(_("      --subscription=NAME         subscription name\n"));
     269           2 :     printf(_("  -V, --version                   output version information, then exit\n"));
     270           2 :     printf(_("  -?, --help                      show this help, then exit\n"));
     271           2 :     printf(_("\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT);
     272           2 :     printf(_("%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
     273           2 : }
     274             : 
     275             : /*
     276             :  * Subroutine to append "keyword=value" to a connection string,
     277             :  * with proper quoting of the value.  (We assume keywords don't need that.)
     278             :  */
     279             : static void
     280         214 : appendConnStrItem(PQExpBuffer buf, const char *keyword, const char *val)
     281             : {
     282         214 :     if (buf->len > 0)
     283         158 :         appendPQExpBufferChar(buf, ' ');
     284         214 :     appendPQExpBufferStr(buf, keyword);
     285         214 :     appendPQExpBufferChar(buf, '=');
     286         214 :     appendConnStrVal(buf, val);
     287         214 : }
     288             : 
     289             : /*
     290             :  * Validate a connection string. Returns a base connection string that is a
     291             :  * connection string without a database name.
     292             :  *
     293             :  * Since we might process multiple databases, each database name will be
     294             :  * appended to this base connection string to provide a final connection
     295             :  * string. If the second argument (dbname) is not null, returns dbname if the
     296             :  * provided connection string contains it.
     297             :  *
     298             :  * It is the caller's responsibility to free the returned connection string and
     299             :  * dbname.
     300             :  */
     301             : static char *
     302          28 : get_base_conninfo(const char *conninfo, char **dbname)
     303             : {
     304             :     PQExpBuffer buf;
     305             :     PQconninfoOption *conn_opts;
     306             :     PQconninfoOption *conn_opt;
     307          28 :     char       *errmsg = NULL;
     308             :     char       *ret;
     309             : 
     310          28 :     conn_opts = PQconninfoParse(conninfo, &errmsg);
     311          28 :     if (conn_opts == NULL)
     312             :     {
     313           0 :         pg_log_error("could not parse connection string: %s", errmsg);
     314           0 :         PQfreemem(errmsg);
     315           0 :         return NULL;
     316             :     }
     317             : 
     318          28 :     buf = createPQExpBuffer();
     319        1456 :     for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
     320             :     {
     321        1428 :         if (conn_opt->val != NULL && conn_opt->val[0] != '\0')
     322             :         {
     323          66 :             if (strcmp(conn_opt->keyword, "dbname") == 0)
     324             :             {
     325          18 :                 if (dbname)
     326          18 :                     *dbname = pg_strdup(conn_opt->val);
     327          18 :                 continue;
     328             :             }
     329          48 :             appendConnStrItem(buf, conn_opt->keyword, conn_opt->val);
     330             :         }
     331             :     }
     332             : 
     333          28 :     ret = pg_strdup(buf->data);
     334             : 
     335          28 :     destroyPQExpBuffer(buf);
     336          28 :     PQconninfoFree(conn_opts);
     337             : 
     338          28 :     return ret;
     339             : }
     340             : 
     341             : /*
     342             :  * Build a subscriber connection string. Only a few parameters are supported
     343             :  * since it starts a server with restricted access.
     344             :  */
     345             : static char *
     346          28 : get_sub_conninfo(const struct CreateSubscriberOptions *opt)
     347             : {
     348          28 :     PQExpBuffer buf = createPQExpBuffer();
     349             :     char       *ret;
     350             : 
     351          28 :     appendConnStrItem(buf, "port", opt->sub_port);
     352             : #if !defined(WIN32)
     353          28 :     appendConnStrItem(buf, "host", opt->socket_dir);
     354             : #endif
     355          28 :     if (opt->sub_username != NULL)
     356           0 :         appendConnStrItem(buf, "user", opt->sub_username);
     357          28 :     appendConnStrItem(buf, "fallback_application_name", progname);
     358             : 
     359          28 :     ret = pg_strdup(buf->data);
     360             : 
     361          28 :     destroyPQExpBuffer(buf);
     362             : 
     363          28 :     return ret;
     364             : }
     365             : 
     366             : /*
     367             :  * Verify if a PostgreSQL binary (progname) is available in the same directory as
     368             :  * pg_createsubscriber and it has the same version.  It returns the absolute
     369             :  * path of the progname.
     370             :  */
     371             : static char *
     372          40 : get_exec_path(const char *argv0, const char *progname)
     373             : {
     374             :     char       *versionstr;
     375             :     char       *exec_path;
     376             :     int         ret;
     377             : 
     378          40 :     versionstr = psprintf("%s (PostgreSQL) %s\n", progname, PG_VERSION);
     379          40 :     exec_path = pg_malloc(MAXPGPATH);
     380          40 :     ret = find_other_exec(argv0, progname, versionstr, exec_path);
     381             : 
     382          40 :     if (ret < 0)
     383             :     {
     384             :         char        full_path[MAXPGPATH];
     385             : 
     386           0 :         if (find_my_exec(argv0, full_path) < 0)
     387           0 :             strlcpy(full_path, progname, sizeof(full_path));
     388             : 
     389           0 :         if (ret == -1)
     390           0 :             pg_fatal("program \"%s\" is needed by %s but was not found in the same directory as \"%s\"",
     391             :                      progname, "pg_createsubscriber", full_path);
     392             :         else
     393           0 :             pg_fatal("program \"%s\" was found by \"%s\" but was not the same version as %s",
     394             :                      progname, full_path, "pg_createsubscriber");
     395             :     }
     396             : 
     397          40 :     pg_log_debug("%s path is:  %s", progname, exec_path);
     398             : 
     399          40 :     return exec_path;
     400             : }
     401             : 
     402             : /*
     403             :  * Is it a cluster directory? These are preliminary checks. It is far from
     404             :  * making an accurate check. If it is not a clone from the publisher, it will
     405             :  * eventually fail in a future step.
     406             :  */
     407             : static void
     408          20 : check_data_directory(const char *datadir)
     409             : {
     410             :     struct stat statbuf;
     411             :     uint32      major_version;
     412             :     char       *version_str;
     413             : 
     414          20 :     pg_log_info("checking if directory \"%s\" is a cluster data directory",
     415             :                 datadir);
     416             : 
     417          20 :     if (stat(datadir, &statbuf) != 0)
     418             :     {
     419           0 :         if (errno == ENOENT)
     420           0 :             pg_fatal("data directory \"%s\" does not exist", datadir);
     421             :         else
     422           0 :             pg_fatal("could not access directory \"%s\": %m", datadir);
     423             :     }
     424             : 
     425             :     /*
     426             :      * Retrieve the contents of this cluster's PG_VERSION.  We require
     427             :      * compatibility with the same major version as the one this tool is
     428             :      * compiled with.
     429             :      */
     430          20 :     major_version = GET_PG_MAJORVERSION_NUM(get_pg_version(datadir, &version_str));
     431          20 :     if (major_version != PG_MAJORVERSION_NUM)
     432             :     {
     433           0 :         pg_log_error("data directory is of wrong version");
     434           0 :         pg_log_error_detail("File \"%s\" contains \"%s\", which is not compatible with this program's version \"%s\".",
     435             :                             "PG_VERSION", version_str, PG_MAJORVERSION);
     436           0 :         exit(1);
     437             :     }
     438          20 : }
     439             : 
     440             : /*
     441             :  * Append database name into a base connection string.
     442             :  *
     443             :  * dbname is the only parameter that changes so it is not included in the base
     444             :  * connection string. This function concatenates dbname to build a "real"
     445             :  * connection string.
     446             :  */
     447             : static char *
     448          82 : concat_conninfo_dbname(const char *conninfo, const char *dbname)
     449             : {
     450          82 :     PQExpBuffer buf = createPQExpBuffer();
     451             :     char       *ret;
     452             : 
     453             :     Assert(conninfo != NULL);
     454             : 
     455          82 :     appendPQExpBufferStr(buf, conninfo);
     456          82 :     appendConnStrItem(buf, "dbname", dbname);
     457             : 
     458          82 :     ret = pg_strdup(buf->data);
     459          82 :     destroyPQExpBuffer(buf);
     460             : 
     461          82 :     return ret;
     462             : }
     463             : 
     464             : /*
     465             :  * Store publication and subscription information.
     466             :  *
     467             :  * If publication, replication slot and subscription names were specified,
     468             :  * store it here. Otherwise, a generated name will be assigned to the object in
     469             :  * setup_publisher().
     470             :  */
     471             : static struct LogicalRepInfo *
     472          20 : store_pub_sub_info(const struct CreateSubscriberOptions *opt,
     473             :                    const char *pub_base_conninfo,
     474             :                    const char *sub_base_conninfo)
     475             : {
     476             :     struct LogicalRepInfo *dbinfo;
     477          20 :     SimpleStringListCell *pubcell = NULL;
     478          20 :     SimpleStringListCell *subcell = NULL;
     479          20 :     SimpleStringListCell *replslotcell = NULL;
     480          20 :     int         i = 0;
     481             : 
     482          20 :     dbinfo = pg_malloc_array(struct LogicalRepInfo, num_dbs);
     483             : 
     484          20 :     if (num_pubs > 0)
     485           4 :         pubcell = opt->pub_names.head;
     486          20 :     if (num_subs > 0)
     487           2 :         subcell = opt->sub_names.head;
     488          20 :     if (num_replslots > 0)
     489           4 :         replslotcell = opt->replslot_names.head;
     490             : 
     491          60 :     for (SimpleStringListCell *cell = opt->database_names.head; cell; cell = cell->next)
     492             :     {
     493             :         char       *conninfo;
     494             : 
     495             :         /* Fill publisher attributes */
     496          40 :         conninfo = concat_conninfo_dbname(pub_base_conninfo, cell->val);
     497          40 :         dbinfo[i].pubconninfo = conninfo;
     498          40 :         dbinfo[i].dbname = cell->val;
     499          40 :         if (num_pubs > 0)
     500           8 :             dbinfo[i].pubname = pubcell->val;
     501             :         else
     502          32 :             dbinfo[i].pubname = NULL;
     503          40 :         if (num_replslots > 0)
     504           6 :             dbinfo[i].replslotname = replslotcell->val;
     505             :         else
     506          34 :             dbinfo[i].replslotname = NULL;
     507          40 :         dbinfo[i].made_replslot = false;
     508          40 :         dbinfo[i].made_publication = false;
     509             :         /* Fill subscriber attributes */
     510          40 :         conninfo = concat_conninfo_dbname(sub_base_conninfo, cell->val);
     511          40 :         dbinfo[i].subconninfo = conninfo;
     512          40 :         if (num_subs > 0)
     513           4 :             dbinfo[i].subname = subcell->val;
     514             :         else
     515          36 :             dbinfo[i].subname = NULL;
     516             :         /* Other fields will be filled later */
     517             : 
     518          40 :         pg_log_debug("publisher(%d): publication: %s ; replication slot: %s ; connection string: %s", i,
     519             :                      dbinfo[i].pubname ? dbinfo[i].pubname : "(auto)",
     520             :                      dbinfo[i].replslotname ? dbinfo[i].replslotname : "(auto)",
     521             :                      dbinfo[i].pubconninfo);
     522          40 :         pg_log_debug("subscriber(%d): subscription: %s ; connection string: %s, two_phase: %s", i,
     523             :                      dbinfo[i].subname ? dbinfo[i].subname : "(auto)",
     524             :                      dbinfo[i].subconninfo,
     525             :                      dbinfos.two_phase ? "true" : "false");
     526             : 
     527          40 :         if (num_pubs > 0)
     528           8 :             pubcell = pubcell->next;
     529          40 :         if (num_subs > 0)
     530           4 :             subcell = subcell->next;
     531          40 :         if (num_replslots > 0)
     532           6 :             replslotcell = replslotcell->next;
     533             : 
     534          40 :         i++;
     535             :     }
     536             : 
     537          20 :     return dbinfo;
     538             : }
     539             : 
     540             : /*
     541             :  * Open a new connection. If exit_on_error is true, it has an undesired
     542             :  * condition and it should exit immediately.
     543             :  */
     544             : static PGconn *
     545         114 : connect_database(const char *conninfo, bool exit_on_error)
     546             : {
     547             :     PGconn     *conn;
     548             :     PGresult   *res;
     549             : 
     550         114 :     conn = PQconnectdb(conninfo);
     551         114 :     if (PQstatus(conn) != CONNECTION_OK)
     552             :     {
     553           0 :         pg_log_error("connection to database failed: %s",
     554             :                      PQerrorMessage(conn));
     555           0 :         PQfinish(conn);
     556             : 
     557           0 :         if (exit_on_error)
     558           0 :             exit(1);
     559           0 :         return NULL;
     560             :     }
     561             : 
     562             :     /* Secure search_path */
     563         114 :     res = PQexec(conn, ALWAYS_SECURE_SEARCH_PATH_SQL);
     564         114 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
     565             :     {
     566           0 :         pg_log_error("could not clear \"search_path\": %s",
     567             :                      PQresultErrorMessage(res));
     568           0 :         PQclear(res);
     569           0 :         PQfinish(conn);
     570             : 
     571           0 :         if (exit_on_error)
     572           0 :             exit(1);
     573           0 :         return NULL;
     574             :     }
     575         114 :     PQclear(res);
     576             : 
     577         114 :     return conn;
     578             : }
     579             : 
     580             : /*
     581             :  * Close the connection. If exit_on_error is true, it has an undesired
     582             :  * condition and it should exit immediately.
     583             :  */
     584             : static void
     585         114 : disconnect_database(PGconn *conn, bool exit_on_error)
     586             : {
     587             :     Assert(conn != NULL);
     588             : 
     589         114 :     PQfinish(conn);
     590             : 
     591         114 :     if (exit_on_error)
     592           4 :         exit(1);
     593         110 : }
     594             : 
     595             : /*
     596             :  * Obtain the system identifier using the provided connection. It will be used
     597             :  * to compare if a data directory is a clone of another one.
     598             :  */
     599             : static uint64
     600          20 : get_primary_sysid(const char *conninfo)
     601             : {
     602             :     PGconn     *conn;
     603             :     PGresult   *res;
     604             :     uint64      sysid;
     605             : 
     606          20 :     pg_log_info("getting system identifier from publisher");
     607             : 
     608          20 :     conn = connect_database(conninfo, true);
     609             : 
     610          20 :     res = PQexec(conn, "SELECT system_identifier FROM pg_catalog.pg_control_system()");
     611          20 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
     612             :     {
     613           0 :         pg_log_error("could not get system identifier: %s",
     614             :                      PQresultErrorMessage(res));
     615           0 :         disconnect_database(conn, true);
     616             :     }
     617          20 :     if (PQntuples(res) != 1)
     618             :     {
     619           0 :         pg_log_error("could not get system identifier: got %d rows, expected %d row",
     620             :                      PQntuples(res), 1);
     621           0 :         disconnect_database(conn, true);
     622             :     }
     623             : 
     624          20 :     sysid = strtou64(PQgetvalue(res, 0, 0), NULL, 10);
     625             : 
     626          20 :     pg_log_info("system identifier is %" PRIu64 " on publisher", sysid);
     627             : 
     628          20 :     PQclear(res);
     629          20 :     disconnect_database(conn, false);
     630             : 
     631          20 :     return sysid;
     632             : }
     633             : 
     634             : /*
     635             :  * Obtain the system identifier from control file. It will be used to compare
     636             :  * if a data directory is a clone of another one. This routine is used locally
     637             :  * and avoids a connection.
     638             :  */
     639             : static uint64
     640          20 : get_standby_sysid(const char *datadir)
     641             : {
     642             :     ControlFileData *cf;
     643             :     bool        crc_ok;
     644             :     uint64      sysid;
     645             : 
     646          20 :     pg_log_info("getting system identifier from subscriber");
     647             : 
     648          20 :     cf = get_controlfile(datadir, &crc_ok);
     649          20 :     if (!crc_ok)
     650           0 :         pg_fatal("control file appears to be corrupt");
     651             : 
     652          20 :     sysid = cf->system_identifier;
     653             : 
     654          20 :     pg_log_info("system identifier is %" PRIu64 " on subscriber", sysid);
     655             : 
     656          20 :     pg_free(cf);
     657             : 
     658          20 :     return sysid;
     659             : }
     660             : 
     661             : /*
     662             :  * Modify the system identifier. Since a standby server preserves the system
     663             :  * identifier, it makes sense to change it to avoid situations in which WAL
     664             :  * files from one of the systems might be used in the other one.
     665             :  */
     666             : static void
     667           8 : modify_subscriber_sysid(const struct CreateSubscriberOptions *opt)
     668             : {
     669             :     ControlFileData *cf;
     670             :     bool        crc_ok;
     671             :     struct timeval tv;
     672             : 
     673             :     char       *cmd_str;
     674             : 
     675           8 :     pg_log_info("modifying system identifier of subscriber");
     676             : 
     677           8 :     cf = get_controlfile(subscriber_dir, &crc_ok);
     678           8 :     if (!crc_ok)
     679           0 :         pg_fatal("control file appears to be corrupt");
     680             : 
     681             :     /*
     682             :      * Select a new system identifier.
     683             :      *
     684             :      * XXX this code was extracted from BootStrapXLOG().
     685             :      */
     686           8 :     gettimeofday(&tv, NULL);
     687           8 :     cf->system_identifier = ((uint64) tv.tv_sec) << 32;
     688           8 :     cf->system_identifier |= ((uint64) tv.tv_usec) << 12;
     689           8 :     cf->system_identifier |= getpid() & 0xFFF;
     690             : 
     691           8 :     if (!dry_run)
     692           2 :         update_controlfile(subscriber_dir, cf, true);
     693             : 
     694           8 :     pg_log_info("system identifier is %" PRIu64 " on subscriber",
     695             :                 cf->system_identifier);
     696             : 
     697           8 :     pg_log_info("running pg_resetwal on the subscriber");
     698             : 
     699           8 :     cmd_str = psprintf("\"%s\" -D \"%s\" > \"%s\"", pg_resetwal_path,
     700             :                        subscriber_dir, DEVNULL);
     701             : 
     702           8 :     pg_log_debug("pg_resetwal command is: %s", cmd_str);
     703             : 
     704           8 :     if (!dry_run)
     705             :     {
     706           2 :         int         rc = system(cmd_str);
     707             : 
     708           2 :         if (rc == 0)
     709           2 :             pg_log_info("subscriber successfully changed the system identifier");
     710             :         else
     711           0 :             pg_fatal("could not change system identifier of subscriber: %s", wait_result_to_str(rc));
     712             :     }
     713             : 
     714           8 :     pg_free(cf);
     715           8 : }
     716             : 
     717             : /*
     718             :  * Generate an object name using a prefix, database oid and a random integer.
     719             :  * It is used in case the user does not specify an object name (publication,
     720             :  * subscription, replication slot).
     721             :  */
     722             : static char *
     723          16 : generate_object_name(PGconn *conn)
     724             : {
     725             :     PGresult   *res;
     726             :     Oid         oid;
     727             :     uint32      rand;
     728             :     char       *objname;
     729             : 
     730          16 :     res = PQexec(conn,
     731             :                  "SELECT oid FROM pg_catalog.pg_database "
     732             :                  "WHERE datname = pg_catalog.current_database()");
     733          16 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
     734             :     {
     735           0 :         pg_log_error("could not obtain database OID: %s",
     736             :                      PQresultErrorMessage(res));
     737           0 :         disconnect_database(conn, true);
     738             :     }
     739             : 
     740          16 :     if (PQntuples(res) != 1)
     741             :     {
     742           0 :         pg_log_error("could not obtain database OID: got %d rows, expected %d row",
     743             :                      PQntuples(res), 1);
     744           0 :         disconnect_database(conn, true);
     745             :     }
     746             : 
     747             :     /* Database OID */
     748          16 :     oid = strtoul(PQgetvalue(res, 0, 0), NULL, 10);
     749             : 
     750          16 :     PQclear(res);
     751             : 
     752             :     /* Random unsigned integer */
     753          16 :     rand = pg_prng_uint32(&prng_state);
     754             : 
     755             :     /*
     756             :      * Build the object name. The name must not exceed NAMEDATALEN - 1. This
     757             :      * current schema uses a maximum of 40 characters (20 + 10 + 1 + 8 +
     758             :      * '\0').
     759             :      */
     760          16 :     objname = psprintf("pg_createsubscriber_%u_%x", oid, rand);
     761             : 
     762          16 :     return objname;
     763             : }
     764             : 
     765             : /*
     766             :  * Create the publications and replication slots in preparation for logical
     767             :  * replication. Returns the LSN from latest replication slot. It will be the
     768             :  * replication start point that is used to adjust the subscriptions (see
     769             :  * set_replication_progress).
     770             :  */
     771             : static char *
     772           8 : setup_publisher(struct LogicalRepInfo *dbinfo)
     773             : {
     774           8 :     char       *lsn = NULL;
     775             : 
     776           8 :     pg_prng_seed(&prng_state, (uint64) (getpid() ^ time(NULL)));
     777             : 
     778          24 :     for (int i = 0; i < num_dbs; i++)
     779             :     {
     780             :         PGconn     *conn;
     781          16 :         char       *genname = NULL;
     782             : 
     783          16 :         conn = connect_database(dbinfo[i].pubconninfo, true);
     784             : 
     785             :         /*
     786             :          * If an object name was not specified as command-line options, assign
     787             :          * a generated object name. The replication slot has a different rule.
     788             :          * The subscription name is assigned to the replication slot name if
     789             :          * no replication slot is specified. It follows the same rule as
     790             :          * CREATE SUBSCRIPTION.
     791             :          */
     792          16 :         if (num_pubs == 0 || num_subs == 0 || num_replslots == 0)
     793          16 :             genname = generate_object_name(conn);
     794          16 :         if (num_pubs == 0)
     795           8 :             dbinfo[i].pubname = pg_strdup(genname);
     796          16 :         if (num_subs == 0)
     797          12 :             dbinfo[i].subname = pg_strdup(genname);
     798          16 :         if (num_replslots == 0)
     799          10 :             dbinfo[i].replslotname = pg_strdup(dbinfo[i].subname);
     800             : 
     801             :         /*
     802             :          * Create publication on publisher. This step should be executed
     803             :          * *before* promoting the subscriber to avoid any transactions between
     804             :          * consistent LSN and the new publication rows (such transactions
     805             :          * wouldn't see the new publication rows resulting in an error).
     806             :          */
     807          16 :         create_publication(conn, &dbinfo[i]);
     808             : 
     809             :         /* Create replication slot on publisher */
     810          16 :         if (lsn)
     811           2 :             pg_free(lsn);
     812          16 :         lsn = create_logical_replication_slot(conn, &dbinfo[i]);
     813          16 :         if (lsn == NULL && !dry_run)
     814           0 :             exit(1);
     815             : 
     816             :         /*
     817             :          * Since we are using the LSN returned by the last replication slot as
     818             :          * recovery_target_lsn, this LSN is ahead of the current WAL position
     819             :          * and the recovery waits until the publisher writes a WAL record to
     820             :          * reach the target and ends the recovery. On idle systems, this wait
     821             :          * time is unpredictable and could lead to failure in promoting the
     822             :          * subscriber. To avoid that, insert a harmless WAL record.
     823             :          */
     824          16 :         if (i == num_dbs - 1 && !dry_run)
     825             :         {
     826             :             PGresult   *res;
     827             : 
     828           2 :             res = PQexec(conn, "SELECT pg_log_standby_snapshot()");
     829           2 :             if (PQresultStatus(res) != PGRES_TUPLES_OK)
     830             :             {
     831           0 :                 pg_log_error("could not write an additional WAL record: %s",
     832             :                              PQresultErrorMessage(res));
     833           0 :                 disconnect_database(conn, true);
     834             :             }
     835           2 :             PQclear(res);
     836             :         }
     837             : 
     838          16 :         disconnect_database(conn, false);
     839             :     }
     840             : 
     841           8 :     return lsn;
     842             : }
     843             : 
     844             : /*
     845             :  * Is recovery still in progress?
     846             :  */
     847             : static bool
     848          36 : server_is_in_recovery(PGconn *conn)
     849             : {
     850             :     PGresult   *res;
     851             :     int         ret;
     852             : 
     853          36 :     res = PQexec(conn, "SELECT pg_catalog.pg_is_in_recovery()");
     854             : 
     855          36 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
     856             :     {
     857           0 :         pg_log_error("could not obtain recovery progress: %s",
     858             :                      PQresultErrorMessage(res));
     859           0 :         disconnect_database(conn, true);
     860             :     }
     861             : 
     862             : 
     863          36 :     ret = strcmp("t", PQgetvalue(res, 0, 0));
     864             : 
     865          36 :     PQclear(res);
     866             : 
     867          36 :     return ret == 0;
     868             : }
     869             : 
     870             : /*
     871             :  * Is the primary server ready for logical replication?
     872             :  *
     873             :  * XXX Does it not allow a synchronous replica?
     874             :  */
     875             : static void
     876          12 : check_publisher(const struct LogicalRepInfo *dbinfo)
     877             : {
     878             :     PGconn     *conn;
     879             :     PGresult   *res;
     880          12 :     bool        failed = false;
     881             : 
     882             :     char       *wal_level;
     883             :     int         max_repslots;
     884             :     int         cur_repslots;
     885             :     int         max_walsenders;
     886             :     int         cur_walsenders;
     887             :     int         max_prepared_transactions;
     888             :     char       *max_slot_wal_keep_size;
     889             : 
     890          12 :     pg_log_info("checking settings on publisher");
     891             : 
     892          12 :     conn = connect_database(dbinfo[0].pubconninfo, true);
     893             : 
     894             :     /*
     895             :      * If the primary server is in recovery (i.e. cascading replication),
     896             :      * objects (publication) cannot be created because it is read only.
     897             :      */
     898          12 :     if (server_is_in_recovery(conn))
     899             :     {
     900           2 :         pg_log_error("primary server cannot be in recovery");
     901           2 :         disconnect_database(conn, true);
     902             :     }
     903             : 
     904             :     /*------------------------------------------------------------------------
     905             :      * Logical replication requires a few parameters to be set on publisher.
     906             :      * Since these parameters are not a requirement for physical replication,
     907             :      * we should check it to make sure it won't fail.
     908             :      *
     909             :      * - wal_level = logical
     910             :      * - max_replication_slots >= current + number of dbs to be converted
     911             :      * - max_wal_senders >= current + number of dbs to be converted
     912             :      * - max_slot_wal_keep_size = -1 (to prevent deletion of required WAL files)
     913             :      * -----------------------------------------------------------------------
     914             :      */
     915          10 :     res = PQexec(conn,
     916             :                  "SELECT pg_catalog.current_setting('wal_level'),"
     917             :                  " pg_catalog.current_setting('max_replication_slots'),"
     918             :                  " (SELECT count(*) FROM pg_catalog.pg_replication_slots),"
     919             :                  " pg_catalog.current_setting('max_wal_senders'),"
     920             :                  " (SELECT count(*) FROM pg_catalog.pg_stat_activity WHERE backend_type = 'walsender'),"
     921             :                  " pg_catalog.current_setting('max_prepared_transactions'),"
     922             :                  " pg_catalog.current_setting('max_slot_wal_keep_size')");
     923             : 
     924          10 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
     925             :     {
     926           0 :         pg_log_error("could not obtain publisher settings: %s",
     927             :                      PQresultErrorMessage(res));
     928           0 :         disconnect_database(conn, true);
     929             :     }
     930             : 
     931          10 :     wal_level = pg_strdup(PQgetvalue(res, 0, 0));
     932          10 :     max_repslots = atoi(PQgetvalue(res, 0, 1));
     933          10 :     cur_repslots = atoi(PQgetvalue(res, 0, 2));
     934          10 :     max_walsenders = atoi(PQgetvalue(res, 0, 3));
     935          10 :     cur_walsenders = atoi(PQgetvalue(res, 0, 4));
     936          10 :     max_prepared_transactions = atoi(PQgetvalue(res, 0, 5));
     937          10 :     max_slot_wal_keep_size = pg_strdup(PQgetvalue(res, 0, 6));
     938             : 
     939          10 :     PQclear(res);
     940             : 
     941          10 :     pg_log_debug("publisher: wal_level: %s", wal_level);
     942          10 :     pg_log_debug("publisher: max_replication_slots: %d", max_repslots);
     943          10 :     pg_log_debug("publisher: current replication slots: %d", cur_repslots);
     944          10 :     pg_log_debug("publisher: max_wal_senders: %d", max_walsenders);
     945          10 :     pg_log_debug("publisher: current wal senders: %d", cur_walsenders);
     946          10 :     pg_log_debug("publisher: max_prepared_transactions: %d",
     947             :                  max_prepared_transactions);
     948          10 :     pg_log_debug("publisher: max_slot_wal_keep_size: %s",
     949             :                  max_slot_wal_keep_size);
     950             : 
     951          10 :     disconnect_database(conn, false);
     952             : 
     953          10 :     if (strcmp(wal_level, "logical") != 0)
     954             :     {
     955           2 :         pg_log_error("publisher requires \"wal_level\" >= \"logical\"");
     956           2 :         failed = true;
     957             :     }
     958             : 
     959          10 :     if (max_repslots - cur_repslots < num_dbs)
     960             :     {
     961           2 :         pg_log_error("publisher requires %d replication slots, but only %d remain",
     962             :                      num_dbs, max_repslots - cur_repslots);
     963           2 :         pg_log_error_hint("Increase the configuration parameter \"%s\" to at least %d.",
     964             :                           "max_replication_slots", cur_repslots + num_dbs);
     965           2 :         failed = true;
     966             :     }
     967             : 
     968          10 :     if (max_walsenders - cur_walsenders < num_dbs)
     969             :     {
     970           2 :         pg_log_error("publisher requires %d WAL sender processes, but only %d remain",
     971             :                      num_dbs, max_walsenders - cur_walsenders);
     972           2 :         pg_log_error_hint("Increase the configuration parameter \"%s\" to at least %d.",
     973             :                           "max_wal_senders", cur_walsenders + num_dbs);
     974           2 :         failed = true;
     975             :     }
     976             : 
     977          10 :     if (max_prepared_transactions != 0 && !dbinfos.two_phase)
     978             :     {
     979           0 :         pg_log_warning("two_phase option will not be enabled for replication slots");
     980           0 :         pg_log_warning_detail("Subscriptions will be created with the two_phase option disabled.  "
     981             :                               "Prepared transactions will be replicated at COMMIT PREPARED.");
     982           0 :         pg_log_warning_hint("You can use the command-line option --enable-two-phase to enable two_phase.");
     983             :     }
     984             : 
     985             :     /*
     986             :      * Validate 'max_slot_wal_keep_size'. If this parameter is set to a
     987             :      * non-default value, it may cause replication failures due to required
     988             :      * WAL files being prematurely removed.
     989             :      */
     990          10 :     if (dry_run && (strcmp(max_slot_wal_keep_size, "-1") != 0))
     991             :     {
     992           0 :         pg_log_warning("required WAL could be removed from the publisher");
     993           0 :         pg_log_warning_hint("Set the configuration parameter \"%s\" to -1 to ensure that required WAL files are not prematurely removed.",
     994             :                             "max_slot_wal_keep_size");
     995             :     }
     996             : 
     997          10 :     pg_free(wal_level);
     998             : 
     999          10 :     if (failed)
    1000           2 :         exit(1);
    1001           8 : }
    1002             : 
    1003             : /*
    1004             :  * Is the standby server ready for logical replication?
    1005             :  *
    1006             :  * XXX Does it not allow a time-delayed replica?
    1007             :  *
    1008             :  * XXX In a cascaded replication scenario (P -> S -> C), if the target server
    1009             :  * is S, it cannot detect there is a replica (server C) because server S starts
    1010             :  * accepting only local connections and server C cannot connect to it. Hence,
    1011             :  * there is not a reliable way to provide a suitable error saying the server C
    1012             :  * will be broken at the end of this process (due to pg_resetwal).
    1013             :  */
    1014             : static void
    1015          16 : check_subscriber(const struct LogicalRepInfo *dbinfo)
    1016             : {
    1017             :     PGconn     *conn;
    1018             :     PGresult   *res;
    1019          16 :     bool        failed = false;
    1020             : 
    1021             :     int         max_lrworkers;
    1022             :     int         max_reporigins;
    1023             :     int         max_wprocs;
    1024             : 
    1025          16 :     pg_log_info("checking settings on subscriber");
    1026             : 
    1027          16 :     conn = connect_database(dbinfo[0].subconninfo, true);
    1028             : 
    1029             :     /* The target server must be a standby */
    1030          16 :     if (!server_is_in_recovery(conn))
    1031             :     {
    1032           2 :         pg_log_error("target server must be a standby");
    1033           2 :         disconnect_database(conn, true);
    1034             :     }
    1035             : 
    1036             :     /*------------------------------------------------------------------------
    1037             :      * Logical replication requires a few parameters to be set on subscriber.
    1038             :      * Since these parameters are not a requirement for physical replication,
    1039             :      * we should check it to make sure it won't fail.
    1040             :      *
    1041             :      * - max_active_replication_origins >= number of dbs to be converted
    1042             :      * - max_logical_replication_workers >= number of dbs to be converted
    1043             :      * - max_worker_processes >= 1 + number of dbs to be converted
    1044             :      *------------------------------------------------------------------------
    1045             :      */
    1046          14 :     res = PQexec(conn,
    1047             :                  "SELECT setting FROM pg_catalog.pg_settings WHERE name IN ("
    1048             :                  "'max_logical_replication_workers', "
    1049             :                  "'max_active_replication_origins', "
    1050             :                  "'max_worker_processes', "
    1051             :                  "'primary_slot_name') "
    1052             :                  "ORDER BY name");
    1053             : 
    1054          14 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
    1055             :     {
    1056           0 :         pg_log_error("could not obtain subscriber settings: %s",
    1057             :                      PQresultErrorMessage(res));
    1058           0 :         disconnect_database(conn, true);
    1059             :     }
    1060             : 
    1061          14 :     max_reporigins = atoi(PQgetvalue(res, 0, 0));
    1062          14 :     max_lrworkers = atoi(PQgetvalue(res, 1, 0));
    1063          14 :     max_wprocs = atoi(PQgetvalue(res, 2, 0));
    1064          14 :     if (strcmp(PQgetvalue(res, 3, 0), "") != 0)
    1065          12 :         primary_slot_name = pg_strdup(PQgetvalue(res, 3, 0));
    1066             : 
    1067          14 :     pg_log_debug("subscriber: max_logical_replication_workers: %d",
    1068             :                  max_lrworkers);
    1069          14 :     pg_log_debug("subscriber: max_active_replication_origins: %d", max_reporigins);
    1070          14 :     pg_log_debug("subscriber: max_worker_processes: %d", max_wprocs);
    1071          14 :     if (primary_slot_name)
    1072          12 :         pg_log_debug("subscriber: primary_slot_name: %s", primary_slot_name);
    1073             : 
    1074          14 :     PQclear(res);
    1075             : 
    1076          14 :     disconnect_database(conn, false);
    1077             : 
    1078          14 :     if (max_reporigins < num_dbs)
    1079             :     {
    1080           2 :         pg_log_error("subscriber requires %d active replication origins, but only %d remain",
    1081             :                      num_dbs, max_reporigins);
    1082           2 :         pg_log_error_hint("Increase the configuration parameter \"%s\" to at least %d.",
    1083             :                           "max_active_replication_origins", num_dbs);
    1084           2 :         failed = true;
    1085             :     }
    1086             : 
    1087          14 :     if (max_lrworkers < num_dbs)
    1088             :     {
    1089           2 :         pg_log_error("subscriber requires %d logical replication workers, but only %d remain",
    1090             :                      num_dbs, max_lrworkers);
    1091           2 :         pg_log_error_hint("Increase the configuration parameter \"%s\" to at least %d.",
    1092             :                           "max_logical_replication_workers", num_dbs);
    1093           2 :         failed = true;
    1094             :     }
    1095             : 
    1096          14 :     if (max_wprocs < num_dbs + 1)
    1097             :     {
    1098           2 :         pg_log_error("subscriber requires %d worker processes, but only %d remain",
    1099             :                      num_dbs + 1, max_wprocs);
    1100           2 :         pg_log_error_hint("Increase the configuration parameter \"%s\" to at least %d.",
    1101             :                           "max_worker_processes", num_dbs + 1);
    1102           2 :         failed = true;
    1103             :     }
    1104             : 
    1105          14 :     if (failed)
    1106           2 :         exit(1);
    1107          12 : }
    1108             : 
    1109             : /*
    1110             :  * Drop a specified subscription. This is to avoid duplicate subscriptions on
    1111             :  * the primary (publisher node) and the newly created subscriber. We
    1112             :  * shouldn't drop the associated slot as that would be used by the publisher
    1113             :  * node.
    1114             :  */
    1115             : static void
    1116           8 : drop_existing_subscriptions(PGconn *conn, const char *subname, const char *dbname)
    1117             : {
    1118           8 :     PQExpBuffer query = createPQExpBuffer();
    1119             :     PGresult   *res;
    1120             : 
    1121             :     Assert(conn != NULL);
    1122             : 
    1123             :     /*
    1124             :      * Construct a query string. These commands are allowed to be executed
    1125             :      * within a transaction.
    1126             :      */
    1127           8 :     appendPQExpBuffer(query, "ALTER SUBSCRIPTION %s DISABLE;",
    1128             :                       subname);
    1129           8 :     appendPQExpBuffer(query, " ALTER SUBSCRIPTION %s SET (slot_name = NONE);",
    1130             :                       subname);
    1131           8 :     appendPQExpBuffer(query, " DROP SUBSCRIPTION %s;", subname);
    1132             : 
    1133           8 :     pg_log_info("dropping subscription \"%s\" in database \"%s\"",
    1134             :                 subname, dbname);
    1135             : 
    1136           8 :     if (!dry_run)
    1137             :     {
    1138           2 :         res = PQexec(conn, query->data);
    1139             : 
    1140           2 :         if (PQresultStatus(res) != PGRES_COMMAND_OK)
    1141             :         {
    1142           0 :             pg_log_error("could not drop subscription \"%s\": %s",
    1143             :                          subname, PQresultErrorMessage(res));
    1144           0 :             disconnect_database(conn, true);
    1145             :         }
    1146             : 
    1147           2 :         PQclear(res);
    1148             :     }
    1149             : 
    1150           8 :     destroyPQExpBuffer(query);
    1151           8 : }
    1152             : 
    1153             : /*
    1154             :  * Retrieve and drop the pre-existing subscriptions.
    1155             :  */
    1156             : static void
    1157          16 : check_and_drop_existing_subscriptions(PGconn *conn,
    1158             :                                       const struct LogicalRepInfo *dbinfo)
    1159             : {
    1160          16 :     PQExpBuffer query = createPQExpBuffer();
    1161             :     char       *dbname;
    1162             :     PGresult   *res;
    1163             : 
    1164             :     Assert(conn != NULL);
    1165             : 
    1166          16 :     dbname = PQescapeLiteral(conn, dbinfo->dbname, strlen(dbinfo->dbname));
    1167             : 
    1168          16 :     appendPQExpBuffer(query,
    1169             :                       "SELECT s.subname FROM pg_catalog.pg_subscription s "
    1170             :                       "INNER JOIN pg_catalog.pg_database d ON (s.subdbid = d.oid) "
    1171             :                       "WHERE d.datname = %s",
    1172             :                       dbname);
    1173          16 :     res = PQexec(conn, query->data);
    1174             : 
    1175          16 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
    1176             :     {
    1177           0 :         pg_log_error("could not obtain pre-existing subscriptions: %s",
    1178             :                      PQresultErrorMessage(res));
    1179           0 :         disconnect_database(conn, true);
    1180             :     }
    1181             : 
    1182          24 :     for (int i = 0; i < PQntuples(res); i++)
    1183           8 :         drop_existing_subscriptions(conn, PQgetvalue(res, i, 0),
    1184           8 :                                     dbinfo->dbname);
    1185             : 
    1186          16 :     PQclear(res);
    1187          16 :     destroyPQExpBuffer(query);
    1188          16 :     PQfreemem(dbname);
    1189          16 : }
    1190             : 
    1191             : /*
    1192             :  * Create the subscriptions, adjust the initial location for logical
    1193             :  * replication and enable the subscriptions. That's the last step for logical
    1194             :  * replication setup.
    1195             :  */
    1196             : static void
    1197           8 : setup_subscriber(struct LogicalRepInfo *dbinfo, const char *consistent_lsn)
    1198             : {
    1199          24 :     for (int i = 0; i < num_dbs; i++)
    1200             :     {
    1201             :         PGconn     *conn;
    1202             : 
    1203             :         /* Connect to subscriber. */
    1204          16 :         conn = connect_database(dbinfo[i].subconninfo, true);
    1205             : 
    1206             :         /*
    1207             :          * We don't need the pre-existing subscriptions on the newly formed
    1208             :          * subscriber. They can connect to other publisher nodes and either
    1209             :          * get some unwarranted data or can lead to ERRORs in connecting to
    1210             :          * such nodes.
    1211             :          */
    1212          16 :         check_and_drop_existing_subscriptions(conn, &dbinfo[i]);
    1213             : 
    1214             :         /* Check and drop the required publications in the given database. */
    1215          16 :         check_and_drop_publications(conn, &dbinfo[i]);
    1216             : 
    1217          16 :         create_subscription(conn, &dbinfo[i]);
    1218             : 
    1219             :         /* Set the replication progress to the correct LSN */
    1220          16 :         set_replication_progress(conn, &dbinfo[i], consistent_lsn);
    1221             : 
    1222             :         /* Enable subscription */
    1223          16 :         enable_subscription(conn, &dbinfo[i]);
    1224             : 
    1225          16 :         disconnect_database(conn, false);
    1226             :     }
    1227           8 : }
    1228             : 
    1229             : /*
    1230             :  * Write the required recovery parameters.
    1231             :  */
    1232             : static void
    1233           8 : setup_recovery(const struct LogicalRepInfo *dbinfo, const char *datadir, const char *lsn)
    1234             : {
    1235             :     PGconn     *conn;
    1236             :     PQExpBuffer recoveryconfcontents;
    1237             : 
    1238             :     /*
    1239             :      * Despite of the recovery parameters will be written to the subscriber,
    1240             :      * use a publisher connection. The primary_conninfo is generated using the
    1241             :      * connection settings.
    1242             :      */
    1243           8 :     conn = connect_database(dbinfo[0].pubconninfo, true);
    1244             : 
    1245             :     /*
    1246             :      * Write recovery parameters.
    1247             :      *
    1248             :      * The subscriber is not running yet. In dry run mode, the recovery
    1249             :      * parameters *won't* be written. An invalid LSN is used for printing
    1250             :      * purposes. Additional recovery parameters are added here. It avoids
    1251             :      * unexpected behavior such as end of recovery as soon as a consistent
    1252             :      * state is reached (recovery_target) and failure due to multiple recovery
    1253             :      * targets (name, time, xid, LSN).
    1254             :      */
    1255           8 :     recoveryconfcontents = GenerateRecoveryConfig(conn, NULL, NULL);
    1256           8 :     appendPQExpBufferStr(recoveryconfcontents, "recovery_target = ''\n");
    1257           8 :     appendPQExpBufferStr(recoveryconfcontents,
    1258             :                          "recovery_target_timeline = 'latest'\n");
    1259             : 
    1260             :     /*
    1261             :      * Set recovery_target_inclusive = false to avoid reapplying the
    1262             :      * transaction committed at 'lsn' after subscription is enabled. This is
    1263             :      * because the provided 'lsn' is also used as the replication start point
    1264             :      * for the subscription. So, the server can send the transaction committed
    1265             :      * at that 'lsn' after replication is started which can lead to applying
    1266             :      * the same transaction twice if we keep recovery_target_inclusive = true.
    1267             :      */
    1268           8 :     appendPQExpBufferStr(recoveryconfcontents,
    1269             :                          "recovery_target_inclusive = false\n");
    1270           8 :     appendPQExpBufferStr(recoveryconfcontents,
    1271             :                          "recovery_target_action = promote\n");
    1272           8 :     appendPQExpBufferStr(recoveryconfcontents, "recovery_target_name = ''\n");
    1273           8 :     appendPQExpBufferStr(recoveryconfcontents, "recovery_target_time = ''\n");
    1274           8 :     appendPQExpBufferStr(recoveryconfcontents, "recovery_target_xid = ''\n");
    1275             : 
    1276           8 :     if (dry_run)
    1277             :     {
    1278           6 :         appendPQExpBufferStr(recoveryconfcontents, "# dry run mode");
    1279           6 :         appendPQExpBuffer(recoveryconfcontents,
    1280             :                           "recovery_target_lsn = '%X/%08X'\n",
    1281           6 :                           LSN_FORMAT_ARGS((XLogRecPtr) InvalidXLogRecPtr));
    1282             :     }
    1283             :     else
    1284             :     {
    1285           2 :         appendPQExpBuffer(recoveryconfcontents, "recovery_target_lsn = '%s'\n",
    1286             :                           lsn);
    1287           2 :         WriteRecoveryConfig(conn, datadir, recoveryconfcontents);
    1288             :     }
    1289           8 :     disconnect_database(conn, false);
    1290             : 
    1291           8 :     pg_log_debug("recovery parameters:\n%s", recoveryconfcontents->data);
    1292           8 : }
    1293             : 
    1294             : /*
    1295             :  * Drop physical replication slot on primary if the standby was using it. After
    1296             :  * the transformation, it has no use.
    1297             :  *
    1298             :  * XXX we might not fail here. Instead, we provide a warning so the user
    1299             :  * eventually drops this replication slot later.
    1300             :  */
    1301             : static void
    1302           8 : drop_primary_replication_slot(struct LogicalRepInfo *dbinfo, const char *slotname)
    1303             : {
    1304             :     PGconn     *conn;
    1305             : 
    1306             :     /* Replication slot does not exist, do nothing */
    1307           8 :     if (!primary_slot_name)
    1308           0 :         return;
    1309             : 
    1310           8 :     conn = connect_database(dbinfo[0].pubconninfo, false);
    1311           8 :     if (conn != NULL)
    1312             :     {
    1313           8 :         drop_replication_slot(conn, &dbinfo[0], slotname);
    1314           8 :         disconnect_database(conn, false);
    1315             :     }
    1316             :     else
    1317             :     {
    1318           0 :         pg_log_warning("could not drop replication slot \"%s\" on primary",
    1319             :                        slotname);
    1320           0 :         pg_log_warning_hint("Drop this replication slot soon to avoid retention of WAL files.");
    1321             :     }
    1322             : }
    1323             : 
    1324             : /*
    1325             :  * Drop failover replication slots on subscriber. After the transformation,
    1326             :  * they have no use.
    1327             :  *
    1328             :  * XXX We do not fail here. Instead, we provide a warning so the user can drop
    1329             :  * them later.
    1330             :  */
    1331             : static void
    1332           8 : drop_failover_replication_slots(struct LogicalRepInfo *dbinfo)
    1333             : {
    1334             :     PGconn     *conn;
    1335             :     PGresult   *res;
    1336             : 
    1337           8 :     conn = connect_database(dbinfo[0].subconninfo, false);
    1338           8 :     if (conn != NULL)
    1339             :     {
    1340             :         /* Get failover replication slot names */
    1341           8 :         res = PQexec(conn,
    1342             :                      "SELECT slot_name FROM pg_catalog.pg_replication_slots WHERE failover");
    1343             : 
    1344           8 :         if (PQresultStatus(res) == PGRES_TUPLES_OK)
    1345             :         {
    1346             :             /* Remove failover replication slots from subscriber */
    1347          16 :             for (int i = 0; i < PQntuples(res); i++)
    1348           8 :                 drop_replication_slot(conn, &dbinfo[0], PQgetvalue(res, i, 0));
    1349             :         }
    1350             :         else
    1351             :         {
    1352           0 :             pg_log_warning("could not obtain failover replication slot information: %s",
    1353             :                            PQresultErrorMessage(res));
    1354           0 :             pg_log_warning_hint("Drop the failover replication slots on subscriber soon to avoid retention of WAL files.");
    1355             :         }
    1356             : 
    1357           8 :         PQclear(res);
    1358           8 :         disconnect_database(conn, false);
    1359             :     }
    1360             :     else
    1361             :     {
    1362           0 :         pg_log_warning("could not drop failover replication slot");
    1363           0 :         pg_log_warning_hint("Drop the failover replication slots on subscriber soon to avoid retention of WAL files.");
    1364             :     }
    1365           8 : }
    1366             : 
    1367             : /*
    1368             :  * Create a logical replication slot and returns a LSN.
    1369             :  *
    1370             :  * CreateReplicationSlot() is not used because it does not provide the one-row
    1371             :  * result set that contains the LSN.
    1372             :  */
    1373             : static char *
    1374          16 : create_logical_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo)
    1375             : {
    1376          16 :     PQExpBuffer str = createPQExpBuffer();
    1377          16 :     PGresult   *res = NULL;
    1378          16 :     const char *slot_name = dbinfo->replslotname;
    1379             :     char       *slot_name_esc;
    1380          16 :     char       *lsn = NULL;
    1381             : 
    1382             :     Assert(conn != NULL);
    1383             : 
    1384          16 :     pg_log_info("creating the replication slot \"%s\" in database \"%s\" on publisher",
    1385             :                 slot_name, dbinfo->dbname);
    1386             : 
    1387          16 :     slot_name_esc = PQescapeLiteral(conn, slot_name, strlen(slot_name));
    1388             : 
    1389          16 :     appendPQExpBuffer(str,
    1390             :                       "SELECT lsn FROM pg_catalog.pg_create_logical_replication_slot(%s, 'pgoutput', false, %s, false)",
    1391             :                       slot_name_esc,
    1392          16 :                       dbinfos.two_phase ? "true" : "false");
    1393             : 
    1394          16 :     PQfreemem(slot_name_esc);
    1395             : 
    1396          16 :     pg_log_debug("command is: %s", str->data);
    1397             : 
    1398          16 :     if (!dry_run)
    1399             :     {
    1400           4 :         res = PQexec(conn, str->data);
    1401           4 :         if (PQresultStatus(res) != PGRES_TUPLES_OK)
    1402             :         {
    1403           0 :             pg_log_error("could not create replication slot \"%s\" in database \"%s\": %s",
    1404             :                          slot_name, dbinfo->dbname,
    1405             :                          PQresultErrorMessage(res));
    1406           0 :             PQclear(res);
    1407           0 :             destroyPQExpBuffer(str);
    1408           0 :             return NULL;
    1409             :         }
    1410             : 
    1411           4 :         lsn = pg_strdup(PQgetvalue(res, 0, 0));
    1412           4 :         PQclear(res);
    1413             :     }
    1414             : 
    1415             :     /* For cleanup purposes */
    1416          16 :     dbinfo->made_replslot = true;
    1417             : 
    1418          16 :     destroyPQExpBuffer(str);
    1419             : 
    1420          16 :     return lsn;
    1421             : }
    1422             : 
    1423             : static void
    1424          16 : drop_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo,
    1425             :                       const char *slot_name)
    1426             : {
    1427          16 :     PQExpBuffer str = createPQExpBuffer();
    1428             :     char       *slot_name_esc;
    1429             :     PGresult   *res;
    1430             : 
    1431             :     Assert(conn != NULL);
    1432             : 
    1433          16 :     pg_log_info("dropping the replication slot \"%s\" in database \"%s\"",
    1434             :                 slot_name, dbinfo->dbname);
    1435             : 
    1436          16 :     slot_name_esc = PQescapeLiteral(conn, slot_name, strlen(slot_name));
    1437             : 
    1438          16 :     appendPQExpBuffer(str, "SELECT pg_catalog.pg_drop_replication_slot(%s)", slot_name_esc);
    1439             : 
    1440          16 :     PQfreemem(slot_name_esc);
    1441             : 
    1442          16 :     pg_log_debug("command is: %s", str->data);
    1443             : 
    1444          16 :     if (!dry_run)
    1445             :     {
    1446           4 :         res = PQexec(conn, str->data);
    1447           4 :         if (PQresultStatus(res) != PGRES_TUPLES_OK)
    1448             :         {
    1449           0 :             pg_log_error("could not drop replication slot \"%s\" in database \"%s\": %s",
    1450             :                          slot_name, dbinfo->dbname, PQresultErrorMessage(res));
    1451           0 :             dbinfo->made_replslot = false;   /* don't try again. */
    1452             :         }
    1453             : 
    1454           4 :         PQclear(res);
    1455             :     }
    1456             : 
    1457          16 :     destroyPQExpBuffer(str);
    1458          16 : }
    1459             : 
    1460             : /*
    1461             :  * Reports a suitable message if pg_ctl fails.
    1462             :  */
    1463             : static void
    1464          48 : pg_ctl_status(const char *pg_ctl_cmd, int rc)
    1465             : {
    1466          48 :     if (rc != 0)
    1467             :     {
    1468           0 :         if (WIFEXITED(rc))
    1469             :         {
    1470           0 :             pg_log_error("pg_ctl failed with exit code %d", WEXITSTATUS(rc));
    1471             :         }
    1472           0 :         else if (WIFSIGNALED(rc))
    1473             :         {
    1474             : #if defined(WIN32)
    1475             :             pg_log_error("pg_ctl was terminated by exception 0x%X",
    1476             :                          WTERMSIG(rc));
    1477             :             pg_log_error_detail("See C include file \"ntstatus.h\" for a description of the hexadecimal value.");
    1478             : #else
    1479           0 :             pg_log_error("pg_ctl was terminated by signal %d: %s",
    1480             :                          WTERMSIG(rc), pg_strsignal(WTERMSIG(rc)));
    1481             : #endif
    1482             :         }
    1483             :         else
    1484             :         {
    1485           0 :             pg_log_error("pg_ctl exited with unrecognized status %d", rc);
    1486             :         }
    1487             : 
    1488           0 :         pg_log_error_detail("The failed command was: %s", pg_ctl_cmd);
    1489           0 :         exit(1);
    1490             :     }
    1491          48 : }
    1492             : 
    1493             : static void
    1494          24 : start_standby_server(const struct CreateSubscriberOptions *opt, bool restricted_access,
    1495             :                      bool restrict_logical_worker)
    1496             : {
    1497          24 :     PQExpBuffer pg_ctl_cmd = createPQExpBuffer();
    1498             :     int         rc;
    1499             : 
    1500          24 :     appendPQExpBuffer(pg_ctl_cmd, "\"%s\" start -D ", pg_ctl_path);
    1501          24 :     appendShellString(pg_ctl_cmd, subscriber_dir);
    1502          24 :     appendPQExpBufferStr(pg_ctl_cmd, " -s -o \"-c sync_replication_slots=off\"");
    1503             : 
    1504             :     /* Prevent unintended slot invalidation */
    1505          24 :     appendPQExpBufferStr(pg_ctl_cmd, " -o \"-c idle_replication_slot_timeout=0\"");
    1506             : 
    1507          24 :     if (restricted_access)
    1508             :     {
    1509          24 :         appendPQExpBuffer(pg_ctl_cmd, " -o \"-p %s\"", opt->sub_port);
    1510             : #if !defined(WIN32)
    1511             : 
    1512             :         /*
    1513             :          * An empty listen_addresses list means the server does not listen on
    1514             :          * any IP interfaces; only Unix-domain sockets can be used to connect
    1515             :          * to the server. Prevent external connections to minimize the chance
    1516             :          * of failure.
    1517             :          */
    1518          24 :         appendPQExpBufferStr(pg_ctl_cmd, " -o \"-c listen_addresses='' -c unix_socket_permissions=0700");
    1519          24 :         if (opt->socket_dir)
    1520          24 :             appendPQExpBuffer(pg_ctl_cmd, " -c unix_socket_directories='%s'",
    1521          24 :                               opt->socket_dir);
    1522          24 :         appendPQExpBufferChar(pg_ctl_cmd, '"');
    1523             : #endif
    1524             :     }
    1525          24 :     if (opt->config_file != NULL)
    1526           0 :         appendPQExpBuffer(pg_ctl_cmd, " -o \"-c config_file=%s\"",
    1527           0 :                           opt->config_file);
    1528             : 
    1529             :     /* Suppress to start logical replication if requested */
    1530          24 :     if (restrict_logical_worker)
    1531           8 :         appendPQExpBufferStr(pg_ctl_cmd, " -o \"-c max_logical_replication_workers=0\"");
    1532             : 
    1533          24 :     pg_log_debug("pg_ctl command is: %s", pg_ctl_cmd->data);
    1534          24 :     rc = system(pg_ctl_cmd->data);
    1535          24 :     pg_ctl_status(pg_ctl_cmd->data, rc);
    1536          24 :     standby_running = true;
    1537          24 :     destroyPQExpBuffer(pg_ctl_cmd);
    1538          24 :     pg_log_info("server was started");
    1539          24 : }
    1540             : 
    1541             : static void
    1542          24 : stop_standby_server(const char *datadir)
    1543             : {
    1544             :     char       *pg_ctl_cmd;
    1545             :     int         rc;
    1546             : 
    1547          24 :     pg_ctl_cmd = psprintf("\"%s\" stop -D \"%s\" -s", pg_ctl_path,
    1548             :                           datadir);
    1549          24 :     pg_log_debug("pg_ctl command is: %s", pg_ctl_cmd);
    1550          24 :     rc = system(pg_ctl_cmd);
    1551          24 :     pg_ctl_status(pg_ctl_cmd, rc);
    1552          24 :     standby_running = false;
    1553          24 :     pg_log_info("server was stopped");
    1554          24 : }
    1555             : 
    1556             : /*
    1557             :  * Returns after the server finishes the recovery process.
    1558             :  *
    1559             :  * If recovery_timeout option is set, terminate abnormally without finishing
    1560             :  * the recovery process. By default, it waits forever.
    1561             :  *
    1562             :  * XXX Is the recovery process still in progress? When recovery process has a
    1563             :  * better progress reporting mechanism, it should be added here.
    1564             :  */
    1565             : static void
    1566           8 : wait_for_end_recovery(const char *conninfo, const struct CreateSubscriberOptions *opt)
    1567             : {
    1568             :     PGconn     *conn;
    1569           8 :     int         status = POSTMASTER_STILL_STARTING;
    1570           8 :     int         timer = 0;
    1571             : 
    1572           8 :     pg_log_info("waiting for the target server to reach the consistent state");
    1573             : 
    1574           8 :     conn = connect_database(conninfo, true);
    1575             : 
    1576             :     for (;;)
    1577           0 :     {
    1578           8 :         bool        in_recovery = server_is_in_recovery(conn);
    1579             : 
    1580             :         /*
    1581             :          * Does the recovery process finish? In dry run mode, there is no
    1582             :          * recovery mode. Bail out as the recovery process has ended.
    1583             :          */
    1584           8 :         if (!in_recovery || dry_run)
    1585             :         {
    1586           8 :             status = POSTMASTER_READY;
    1587           8 :             recovery_ended = true;
    1588           8 :             break;
    1589             :         }
    1590             : 
    1591             :         /* Bail out after recovery_timeout seconds if this option is set */
    1592           0 :         if (opt->recovery_timeout > 0 && timer >= opt->recovery_timeout)
    1593             :         {
    1594           0 :             stop_standby_server(subscriber_dir);
    1595           0 :             pg_log_error("recovery timed out");
    1596           0 :             disconnect_database(conn, true);
    1597             :         }
    1598             : 
    1599             :         /* Keep waiting */
    1600           0 :         pg_usleep(WAIT_INTERVAL * USEC_PER_SEC);
    1601             : 
    1602           0 :         timer += WAIT_INTERVAL;
    1603             :     }
    1604             : 
    1605           8 :     disconnect_database(conn, false);
    1606             : 
    1607           8 :     if (status == POSTMASTER_STILL_STARTING)
    1608           0 :         pg_fatal("server did not end recovery");
    1609             : 
    1610           8 :     pg_log_info("target server reached the consistent state");
    1611           8 :     pg_log_info_hint("If pg_createsubscriber fails after this point, you must recreate the physical replica before continuing.");
    1612           8 : }
    1613             : 
    1614             : /*
    1615             :  * Create a publication that includes all tables in the database.
    1616             :  */
    1617             : static void
    1618          16 : create_publication(PGconn *conn, struct LogicalRepInfo *dbinfo)
    1619             : {
    1620          16 :     PQExpBuffer str = createPQExpBuffer();
    1621             :     PGresult   *res;
    1622             :     char       *ipubname_esc;
    1623             :     char       *spubname_esc;
    1624             : 
    1625             :     Assert(conn != NULL);
    1626             : 
    1627          16 :     ipubname_esc = PQescapeIdentifier(conn, dbinfo->pubname, strlen(dbinfo->pubname));
    1628          16 :     spubname_esc = PQescapeLiteral(conn, dbinfo->pubname, strlen(dbinfo->pubname));
    1629             : 
    1630             :     /* Check if the publication already exists */
    1631          16 :     appendPQExpBuffer(str,
    1632             :                       "SELECT 1 FROM pg_catalog.pg_publication "
    1633             :                       "WHERE pubname = %s",
    1634             :                       spubname_esc);
    1635          16 :     res = PQexec(conn, str->data);
    1636          16 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
    1637             :     {
    1638           0 :         pg_log_error("could not obtain publication information: %s",
    1639             :                      PQresultErrorMessage(res));
    1640           0 :         disconnect_database(conn, true);
    1641             :     }
    1642             : 
    1643          16 :     if (PQntuples(res) == 1)
    1644             :     {
    1645             :         /*
    1646             :          * Unfortunately, if it reaches this code path, it will always fail
    1647             :          * (unless you decide to change the existing publication name). That's
    1648             :          * bad but it is very unlikely that the user will choose a name with
    1649             :          * pg_createsubscriber_ prefix followed by the exact database oid and
    1650             :          * a random number.
    1651             :          */
    1652           0 :         pg_log_error("publication \"%s\" already exists", dbinfo->pubname);
    1653           0 :         pg_log_error_hint("Consider renaming this publication before continuing.");
    1654           0 :         disconnect_database(conn, true);
    1655             :     }
    1656             : 
    1657          16 :     PQclear(res);
    1658          16 :     resetPQExpBuffer(str);
    1659             : 
    1660          16 :     pg_log_info("creating publication \"%s\" in database \"%s\"",
    1661             :                 dbinfo->pubname, dbinfo->dbname);
    1662             : 
    1663          16 :     appendPQExpBuffer(str, "CREATE PUBLICATION %s FOR ALL TABLES",
    1664             :                       ipubname_esc);
    1665             : 
    1666          16 :     pg_log_debug("command is: %s", str->data);
    1667             : 
    1668          16 :     if (!dry_run)
    1669             :     {
    1670           4 :         res = PQexec(conn, str->data);
    1671           4 :         if (PQresultStatus(res) != PGRES_COMMAND_OK)
    1672             :         {
    1673           0 :             pg_log_error("could not create publication \"%s\" in database \"%s\": %s",
    1674             :                          dbinfo->pubname, dbinfo->dbname, PQresultErrorMessage(res));
    1675           0 :             disconnect_database(conn, true);
    1676             :         }
    1677           4 :         PQclear(res);
    1678             :     }
    1679             : 
    1680             :     /* For cleanup purposes */
    1681          16 :     dbinfo->made_publication = true;
    1682             : 
    1683          16 :     PQfreemem(ipubname_esc);
    1684          16 :     PQfreemem(spubname_esc);
    1685          16 :     destroyPQExpBuffer(str);
    1686          16 : }
    1687             : 
    1688             : /*
    1689             :  * Drop the specified publication in the given database.
    1690             :  */
    1691             : static void
    1692          20 : drop_publication(PGconn *conn, const char *pubname, const char *dbname,
    1693             :                  bool *made_publication)
    1694             : {
    1695          20 :     PQExpBuffer str = createPQExpBuffer();
    1696             :     PGresult   *res;
    1697             :     char       *pubname_esc;
    1698             : 
    1699             :     Assert(conn != NULL);
    1700             : 
    1701          20 :     pubname_esc = PQescapeIdentifier(conn, pubname, strlen(pubname));
    1702             : 
    1703          20 :     pg_log_info("dropping publication \"%s\" in database \"%s\"",
    1704             :                 pubname, dbname);
    1705             : 
    1706          20 :     appendPQExpBuffer(str, "DROP PUBLICATION %s", pubname_esc);
    1707             : 
    1708          20 :     PQfreemem(pubname_esc);
    1709             : 
    1710          20 :     pg_log_debug("command is: %s", str->data);
    1711             : 
    1712          20 :     if (!dry_run)
    1713             :     {
    1714           8 :         res = PQexec(conn, str->data);
    1715           8 :         if (PQresultStatus(res) != PGRES_COMMAND_OK)
    1716             :         {
    1717           0 :             pg_log_error("could not drop publication \"%s\" in database \"%s\": %s",
    1718             :                          pubname, dbname, PQresultErrorMessage(res));
    1719           0 :             *made_publication = false;  /* don't try again. */
    1720             : 
    1721             :             /*
    1722             :              * Don't disconnect and exit here. This routine is used by primary
    1723             :              * (cleanup publication / replication slot due to an error) and
    1724             :              * subscriber (remove the replicated publications). In both cases,
    1725             :              * it can continue and provide instructions for the user to remove
    1726             :              * it later if cleanup fails.
    1727             :              */
    1728             :         }
    1729           8 :         PQclear(res);
    1730             :     }
    1731             : 
    1732          20 :     destroyPQExpBuffer(str);
    1733          20 : }
    1734             : 
    1735             : /*
    1736             :  * Retrieve and drop the publications.
    1737             :  *
    1738             :  * Since the publications were created before the consistent LSN, they
    1739             :  * remain on the subscriber even after the physical replica is
    1740             :  * promoted. Remove these publications from the subscriber because
    1741             :  * they have no use. Additionally, if requested, drop all pre-existing
    1742             :  * publications.
    1743             :  */
    1744             : static void
    1745          16 : check_and_drop_publications(PGconn *conn, struct LogicalRepInfo *dbinfo)
    1746             : {
    1747             :     PGresult   *res;
    1748          16 :     bool        drop_all_pubs = dbinfos.objecttypes_to_clean & OBJECTTYPE_PUBLICATIONS;
    1749             : 
    1750             :     Assert(conn != NULL);
    1751             : 
    1752          16 :     if (drop_all_pubs)
    1753             :     {
    1754           4 :         pg_log_info("dropping all existing publications in database \"%s\"",
    1755             :                     dbinfo->dbname);
    1756             : 
    1757             :         /* Fetch all publication names */
    1758           4 :         res = PQexec(conn, "SELECT pubname FROM pg_catalog.pg_publication;");
    1759           4 :         if (PQresultStatus(res) != PGRES_TUPLES_OK)
    1760             :         {
    1761           0 :             pg_log_error("could not obtain publication information: %s",
    1762             :                          PQresultErrorMessage(res));
    1763           0 :             PQclear(res);
    1764           0 :             disconnect_database(conn, true);
    1765             :         }
    1766             : 
    1767             :         /* Drop each publication */
    1768          12 :         for (int i = 0; i < PQntuples(res); i++)
    1769           8 :             drop_publication(conn, PQgetvalue(res, i, 0), dbinfo->dbname,
    1770             :                              &dbinfo->made_publication);
    1771             : 
    1772           4 :         PQclear(res);
    1773             :     }
    1774             : 
    1775             :     /*
    1776             :      * In dry-run mode, we don't create publications, but we still try to drop
    1777             :      * those to provide necessary information to the user.
    1778             :      */
    1779          16 :     if (!drop_all_pubs || dry_run)
    1780          12 :         drop_publication(conn, dbinfo->pubname, dbinfo->dbname,
    1781             :                          &dbinfo->made_publication);
    1782          16 : }
    1783             : 
    1784             : /*
    1785             :  * Create a subscription with some predefined options.
    1786             :  *
    1787             :  * A replication slot was already created in a previous step. Let's use it.  It
    1788             :  * is not required to copy data. The subscription will be created but it will
    1789             :  * not be enabled now. That's because the replication progress must be set and
    1790             :  * the replication origin name (one of the function arguments) contains the
    1791             :  * subscription OID in its name. Once the subscription is created,
    1792             :  * set_replication_progress() can obtain the chosen origin name and set up its
    1793             :  * initial location.
    1794             :  */
    1795             : static void
    1796          16 : create_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo)
    1797             : {
    1798          16 :     PQExpBuffer str = createPQExpBuffer();
    1799             :     PGresult   *res;
    1800             :     char       *pubname_esc;
    1801             :     char       *subname_esc;
    1802             :     char       *pubconninfo_esc;
    1803             :     char       *replslotname_esc;
    1804             : 
    1805             :     Assert(conn != NULL);
    1806             : 
    1807          16 :     pubname_esc = PQescapeIdentifier(conn, dbinfo->pubname, strlen(dbinfo->pubname));
    1808          16 :     subname_esc = PQescapeIdentifier(conn, dbinfo->subname, strlen(dbinfo->subname));
    1809          16 :     pubconninfo_esc = PQescapeLiteral(conn, dbinfo->pubconninfo, strlen(dbinfo->pubconninfo));
    1810          16 :     replslotname_esc = PQescapeLiteral(conn, dbinfo->replslotname, strlen(dbinfo->replslotname));
    1811             : 
    1812          16 :     pg_log_info("creating subscription \"%s\" in database \"%s\"",
    1813             :                 dbinfo->subname, dbinfo->dbname);
    1814             : 
    1815          16 :     appendPQExpBuffer(str,
    1816             :                       "CREATE SUBSCRIPTION %s CONNECTION %s PUBLICATION %s "
    1817             :                       "WITH (create_slot = false, enabled = false, "
    1818             :                       "slot_name = %s, copy_data = false, two_phase = %s)",
    1819             :                       subname_esc, pubconninfo_esc, pubname_esc, replslotname_esc,
    1820          16 :                       dbinfos.two_phase ? "true" : "false");
    1821             : 
    1822          16 :     PQfreemem(pubname_esc);
    1823          16 :     PQfreemem(subname_esc);
    1824          16 :     PQfreemem(pubconninfo_esc);
    1825          16 :     PQfreemem(replslotname_esc);
    1826             : 
    1827          16 :     pg_log_debug("command is: %s", str->data);
    1828             : 
    1829          16 :     if (!dry_run)
    1830             :     {
    1831           4 :         res = PQexec(conn, str->data);
    1832           4 :         if (PQresultStatus(res) != PGRES_COMMAND_OK)
    1833             :         {
    1834           0 :             pg_log_error("could not create subscription \"%s\" in database \"%s\": %s",
    1835             :                          dbinfo->subname, dbinfo->dbname, PQresultErrorMessage(res));
    1836           0 :             disconnect_database(conn, true);
    1837             :         }
    1838           4 :         PQclear(res);
    1839             :     }
    1840             : 
    1841          16 :     destroyPQExpBuffer(str);
    1842          16 : }
    1843             : 
    1844             : /*
    1845             :  * Sets the replication progress to the consistent LSN.
    1846             :  *
    1847             :  * The subscriber caught up to the consistent LSN provided by the last
    1848             :  * replication slot that was created. The goal is to set up the initial
    1849             :  * location for the logical replication that is the exact LSN that the
    1850             :  * subscriber was promoted. Once the subscription is enabled it will start
    1851             :  * streaming from that location onwards.  In dry run mode, the subscription OID
    1852             :  * and LSN are set to invalid values for printing purposes.
    1853             :  */
    1854             : static void
    1855          16 : set_replication_progress(PGconn *conn, const struct LogicalRepInfo *dbinfo, const char *lsn)
    1856             : {
    1857          16 :     PQExpBuffer str = createPQExpBuffer();
    1858             :     PGresult   *res;
    1859             :     Oid         suboid;
    1860             :     char       *subname;
    1861             :     char       *dbname;
    1862             :     char       *originname;
    1863             :     char       *lsnstr;
    1864             : 
    1865             :     Assert(conn != NULL);
    1866             : 
    1867          16 :     subname = PQescapeLiteral(conn, dbinfo->subname, strlen(dbinfo->subname));
    1868          16 :     dbname = PQescapeLiteral(conn, dbinfo->dbname, strlen(dbinfo->dbname));
    1869             : 
    1870          16 :     appendPQExpBuffer(str,
    1871             :                       "SELECT s.oid FROM pg_catalog.pg_subscription s "
    1872             :                       "INNER JOIN pg_catalog.pg_database d ON (s.subdbid = d.oid) "
    1873             :                       "WHERE s.subname = %s AND d.datname = %s",
    1874             :                       subname, dbname);
    1875             : 
    1876          16 :     res = PQexec(conn, str->data);
    1877          16 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
    1878             :     {
    1879           0 :         pg_log_error("could not obtain subscription OID: %s",
    1880             :                      PQresultErrorMessage(res));
    1881           0 :         disconnect_database(conn, true);
    1882             :     }
    1883             : 
    1884          16 :     if (PQntuples(res) != 1 && !dry_run)
    1885             :     {
    1886           0 :         pg_log_error("could not obtain subscription OID: got %d rows, expected %d row",
    1887             :                      PQntuples(res), 1);
    1888           0 :         disconnect_database(conn, true);
    1889             :     }
    1890             : 
    1891          16 :     if (dry_run)
    1892             :     {
    1893          12 :         suboid = InvalidOid;
    1894          12 :         lsnstr = psprintf("%X/%08X", LSN_FORMAT_ARGS((XLogRecPtr) InvalidXLogRecPtr));
    1895             :     }
    1896             :     else
    1897             :     {
    1898           4 :         suboid = strtoul(PQgetvalue(res, 0, 0), NULL, 10);
    1899           4 :         lsnstr = psprintf("%s", lsn);
    1900             :     }
    1901             : 
    1902          16 :     PQclear(res);
    1903             : 
    1904             :     /*
    1905             :      * The origin name is defined as pg_%u. %u is the subscription OID. See
    1906             :      * ApplyWorkerMain().
    1907             :      */
    1908          16 :     originname = psprintf("pg_%u", suboid);
    1909             : 
    1910          16 :     pg_log_info("setting the replication progress (node name \"%s\", LSN %s) in database \"%s\"",
    1911             :                 originname, lsnstr, dbinfo->dbname);
    1912             : 
    1913          16 :     resetPQExpBuffer(str);
    1914          16 :     appendPQExpBuffer(str,
    1915             :                       "SELECT pg_catalog.pg_replication_origin_advance('%s', '%s')",
    1916             :                       originname, lsnstr);
    1917             : 
    1918          16 :     pg_log_debug("command is: %s", str->data);
    1919             : 
    1920          16 :     if (!dry_run)
    1921             :     {
    1922           4 :         res = PQexec(conn, str->data);
    1923           4 :         if (PQresultStatus(res) != PGRES_TUPLES_OK)
    1924             :         {
    1925           0 :             pg_log_error("could not set replication progress for subscription \"%s\": %s",
    1926             :                          dbinfo->subname, PQresultErrorMessage(res));
    1927           0 :             disconnect_database(conn, true);
    1928             :         }
    1929           4 :         PQclear(res);
    1930             :     }
    1931             : 
    1932          16 :     PQfreemem(subname);
    1933          16 :     PQfreemem(dbname);
    1934          16 :     pg_free(originname);
    1935          16 :     pg_free(lsnstr);
    1936          16 :     destroyPQExpBuffer(str);
    1937          16 : }
    1938             : 
    1939             : /*
    1940             :  * Enables the subscription.
    1941             :  *
    1942             :  * The subscription was created in a previous step but it was disabled. After
    1943             :  * adjusting the initial logical replication location, enable the subscription.
    1944             :  */
    1945             : static void
    1946          16 : enable_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo)
    1947             : {
    1948          16 :     PQExpBuffer str = createPQExpBuffer();
    1949             :     PGresult   *res;
    1950             :     char       *subname;
    1951             : 
    1952             :     Assert(conn != NULL);
    1953             : 
    1954          16 :     subname = PQescapeIdentifier(conn, dbinfo->subname, strlen(dbinfo->subname));
    1955             : 
    1956          16 :     pg_log_info("enabling subscription \"%s\" in database \"%s\"",
    1957             :                 dbinfo->subname, dbinfo->dbname);
    1958             : 
    1959          16 :     appendPQExpBuffer(str, "ALTER SUBSCRIPTION %s ENABLE", subname);
    1960             : 
    1961          16 :     pg_log_debug("command is: %s", str->data);
    1962             : 
    1963          16 :     if (!dry_run)
    1964             :     {
    1965           4 :         res = PQexec(conn, str->data);
    1966           4 :         if (PQresultStatus(res) != PGRES_COMMAND_OK)
    1967             :         {
    1968           0 :             pg_log_error("could not enable subscription \"%s\": %s",
    1969             :                          dbinfo->subname, PQresultErrorMessage(res));
    1970           0 :             disconnect_database(conn, true);
    1971             :         }
    1972             : 
    1973           4 :         PQclear(res);
    1974             :     }
    1975             : 
    1976          16 :     PQfreemem(subname);
    1977          16 :     destroyPQExpBuffer(str);
    1978          16 : }
    1979             : 
    1980             : /*
    1981             :  * Fetch a list of all connectable non-template databases from the source server
    1982             :  * and form a list such that they appear as if the user has specified multiple
    1983             :  * --database options, one for each source database.
    1984             :  */
    1985             : static void
    1986           2 : get_publisher_databases(struct CreateSubscriberOptions *opt,
    1987             :                         bool dbnamespecified)
    1988             : {
    1989             :     PGconn     *conn;
    1990             :     PGresult   *res;
    1991             : 
    1992             :     /* If a database name was specified, just connect to it. */
    1993           2 :     if (dbnamespecified)
    1994           0 :         conn = connect_database(opt->pub_conninfo_str, true);
    1995             :     else
    1996             :     {
    1997             :         /* Otherwise, try postgres first and then template1. */
    1998             :         char       *conninfo;
    1999             : 
    2000           2 :         conninfo = concat_conninfo_dbname(opt->pub_conninfo_str, "postgres");
    2001           2 :         conn = connect_database(conninfo, false);
    2002           2 :         pg_free(conninfo);
    2003           2 :         if (!conn)
    2004             :         {
    2005           0 :             conninfo = concat_conninfo_dbname(opt->pub_conninfo_str, "template1");
    2006           0 :             conn = connect_database(conninfo, true);
    2007           0 :             pg_free(conninfo);
    2008             :         }
    2009             :     }
    2010             : 
    2011           2 :     res = PQexec(conn, "SELECT datname FROM pg_database WHERE datistemplate = false AND datallowconn AND datconnlimit <> -2 ORDER BY 1");
    2012           2 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
    2013             :     {
    2014           0 :         pg_log_error("could not obtain a list of databases: %s", PQresultErrorMessage(res));
    2015           0 :         PQclear(res);
    2016           0 :         disconnect_database(conn, true);
    2017             :     }
    2018             : 
    2019           8 :     for (int i = 0; i < PQntuples(res); i++)
    2020             :     {
    2021           6 :         const char *dbname = PQgetvalue(res, i, 0);
    2022             : 
    2023           6 :         simple_string_list_append(&opt->database_names, dbname);
    2024             : 
    2025             :         /* Increment num_dbs to reflect multiple --database options */
    2026           6 :         num_dbs++;
    2027             :     }
    2028             : 
    2029           2 :     PQclear(res);
    2030           2 :     disconnect_database(conn, false);
    2031           2 : }
    2032             : 
    2033             : int
    2034          46 : main(int argc, char **argv)
    2035             : {
    2036             :     static struct option long_options[] =
    2037             :     {
    2038             :         {"all", no_argument, NULL, 'a'},
    2039             :         {"database", required_argument, NULL, 'd'},
    2040             :         {"pgdata", required_argument, NULL, 'D'},
    2041             :         {"dry-run", no_argument, NULL, 'n'},
    2042             :         {"subscriber-port", required_argument, NULL, 'p'},
    2043             :         {"publisher-server", required_argument, NULL, 'P'},
    2044             :         {"socketdir", required_argument, NULL, 's'},
    2045             :         {"recovery-timeout", required_argument, NULL, 't'},
    2046             :         {"enable-two-phase", no_argument, NULL, 'T'},
    2047             :         {"subscriber-username", required_argument, NULL, 'U'},
    2048             :         {"verbose", no_argument, NULL, 'v'},
    2049             :         {"version", no_argument, NULL, 'V'},
    2050             :         {"help", no_argument, NULL, '?'},
    2051             :         {"config-file", required_argument, NULL, 1},
    2052             :         {"publication", required_argument, NULL, 2},
    2053             :         {"replication-slot", required_argument, NULL, 3},
    2054             :         {"subscription", required_argument, NULL, 4},
    2055             :         {"clean", required_argument, NULL, 5},
    2056             :         {NULL, 0, NULL, 0}
    2057             :     };
    2058             : 
    2059          46 :     struct CreateSubscriberOptions opt = {0};
    2060             : 
    2061             :     int         c;
    2062             :     int         option_index;
    2063             : 
    2064             :     char       *pub_base_conninfo;
    2065             :     char       *sub_base_conninfo;
    2066          46 :     char       *dbname_conninfo = NULL;
    2067             : 
    2068             :     uint64      pub_sysid;
    2069             :     uint64      sub_sysid;
    2070             :     struct stat statbuf;
    2071             : 
    2072             :     char       *consistent_lsn;
    2073             : 
    2074             :     char        pidfile[MAXPGPATH];
    2075             : 
    2076          46 :     pg_logging_init(argv[0]);
    2077          46 :     pg_logging_set_level(PG_LOG_WARNING);
    2078          46 :     progname = get_progname(argv[0]);
    2079          46 :     set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_basebackup"));
    2080             : 
    2081          46 :     if (argc > 1)
    2082             :     {
    2083          44 :         if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
    2084             :         {
    2085           2 :             usage();
    2086           2 :             exit(0);
    2087             :         }
    2088          42 :         else if (strcmp(argv[1], "-V") == 0
    2089          42 :                  || strcmp(argv[1], "--version") == 0)
    2090             :         {
    2091           2 :             puts("pg_createsubscriber (PostgreSQL) " PG_VERSION);
    2092           2 :             exit(0);
    2093             :         }
    2094             :     }
    2095             : 
    2096             :     /* Default settings */
    2097          42 :     subscriber_dir = NULL;
    2098          42 :     opt.config_file = NULL;
    2099          42 :     opt.pub_conninfo_str = NULL;
    2100          42 :     opt.socket_dir = NULL;
    2101          42 :     opt.sub_port = DEFAULT_SUB_PORT;
    2102          42 :     opt.sub_username = NULL;
    2103          42 :     opt.two_phase = false;
    2104          42 :     opt.database_names = (SimpleStringList)
    2105             :     {
    2106             :         0
    2107             :     };
    2108          42 :     opt.recovery_timeout = 0;
    2109          42 :     opt.all_dbs = false;
    2110             : 
    2111             :     /*
    2112             :      * Don't allow it to be run as root. It uses pg_ctl which does not allow
    2113             :      * it either.
    2114             :      */
    2115             : #ifndef WIN32
    2116          42 :     if (geteuid() == 0)
    2117             :     {
    2118           0 :         pg_log_error("cannot be executed by \"root\"");
    2119           0 :         pg_log_error_hint("You must run %s as the PostgreSQL superuser.",
    2120             :                           progname);
    2121           0 :         exit(1);
    2122             :     }
    2123             : #endif
    2124             : 
    2125          42 :     get_restricted_token();
    2126             : 
    2127         324 :     while ((c = getopt_long(argc, argv, "ad:D:np:P:s:t:TU:v",
    2128         324 :                             long_options, &option_index)) != -1)
    2129             :     {
    2130         288 :         switch (c)
    2131             :         {
    2132           6 :             case 'a':
    2133           6 :                 opt.all_dbs = true;
    2134           6 :                 break;
    2135          50 :             case 'd':
    2136          50 :                 if (!simple_string_list_member(&opt.database_names, optarg))
    2137             :                 {
    2138          48 :                     simple_string_list_append(&opt.database_names, optarg);
    2139          48 :                     num_dbs++;
    2140             :                 }
    2141             :                 else
    2142           2 :                     pg_fatal("database \"%s\" specified more than once for -d/--database", optarg);
    2143          48 :                 break;
    2144          38 :             case 'D':
    2145          38 :                 subscriber_dir = pg_strdup(optarg);
    2146          38 :                 canonicalize_path(subscriber_dir);
    2147          38 :                 break;
    2148          18 :             case 'n':
    2149          18 :                 dry_run = true;
    2150          18 :                 break;
    2151          24 :             case 'p':
    2152          24 :                 opt.sub_port = pg_strdup(optarg);
    2153          24 :                 break;
    2154          36 :             case 'P':
    2155          36 :                 opt.pub_conninfo_str = pg_strdup(optarg);
    2156          36 :                 break;
    2157          24 :             case 's':
    2158          24 :                 opt.socket_dir = pg_strdup(optarg);
    2159          24 :                 canonicalize_path(opt.socket_dir);
    2160          24 :                 break;
    2161           6 :             case 't':
    2162           6 :                 opt.recovery_timeout = atoi(optarg);
    2163           6 :                 break;
    2164           2 :             case 'T':
    2165           2 :                 opt.two_phase = true;
    2166           2 :                 break;
    2167           0 :             case 'U':
    2168           0 :                 opt.sub_username = pg_strdup(optarg);
    2169           0 :                 break;
    2170          38 :             case 'v':
    2171          38 :                 pg_logging_increase_verbosity();
    2172          38 :                 break;
    2173           0 :             case 1:
    2174           0 :                 opt.config_file = pg_strdup(optarg);
    2175           0 :                 break;
    2176          24 :             case 2:
    2177          24 :                 if (!simple_string_list_member(&opt.pub_names, optarg))
    2178             :                 {
    2179          22 :                     simple_string_list_append(&opt.pub_names, optarg);
    2180          22 :                     num_pubs++;
    2181             :                 }
    2182             :                 else
    2183           2 :                     pg_fatal("publication \"%s\" specified more than once for --publication", optarg);
    2184          22 :                 break;
    2185           8 :             case 3:
    2186           8 :                 if (!simple_string_list_member(&opt.replslot_names, optarg))
    2187             :                 {
    2188           8 :                     simple_string_list_append(&opt.replslot_names, optarg);
    2189           8 :                     num_replslots++;
    2190             :                 }
    2191             :                 else
    2192           0 :                     pg_fatal("replication slot \"%s\" specified more than once for --replication-slot", optarg);
    2193           8 :                 break;
    2194          10 :             case 4:
    2195          10 :                 if (!simple_string_list_member(&opt.sub_names, optarg))
    2196             :                 {
    2197          10 :                     simple_string_list_append(&opt.sub_names, optarg);
    2198          10 :                     num_subs++;
    2199             :                 }
    2200             :                 else
    2201           0 :                     pg_fatal("subscription \"%s\" specified more than once for --subscription", optarg);
    2202          10 :                 break;
    2203           2 :             case 5:
    2204           2 :                 if (!simple_string_list_member(&opt.objecttypes_to_clean, optarg))
    2205           2 :                     simple_string_list_append(&opt.objecttypes_to_clean, optarg);
    2206             :                 else
    2207           0 :                     pg_fatal("object type \"%s\" specified more than once for --clean", optarg);
    2208           2 :                 break;
    2209           2 :             default:
    2210             :                 /* getopt_long already emitted a complaint */
    2211           2 :                 pg_log_error_hint("Try \"%s --help\" for more information.", progname);
    2212           2 :                 exit(1);
    2213             :         }
    2214             :     }
    2215             : 
    2216             :     /* Validate that --all is not used with incompatible options */
    2217          36 :     if (opt.all_dbs)
    2218             :     {
    2219           6 :         char       *bad_switch = NULL;
    2220             : 
    2221           6 :         if (num_dbs > 0)
    2222           2 :             bad_switch = "--database";
    2223           4 :         else if (num_pubs > 0)
    2224           2 :             bad_switch = "--publication";
    2225           2 :         else if (num_replslots > 0)
    2226           0 :             bad_switch = "--replication-slot";
    2227           2 :         else if (num_subs > 0)
    2228           0 :             bad_switch = "--subscription";
    2229             : 
    2230           6 :         if (bad_switch)
    2231             :         {
    2232           4 :             pg_log_error("options %s and -a/--all cannot be used together", bad_switch);
    2233           4 :             pg_log_error_hint("Try \"%s --help\" for more information.", progname);
    2234           4 :             exit(1);
    2235             :         }
    2236             :     }
    2237             : 
    2238             :     /* Any non-option arguments? */
    2239          32 :     if (optind < argc)
    2240             :     {
    2241           0 :         pg_log_error("too many command-line arguments (first is \"%s\")",
    2242             :                      argv[optind]);
    2243           0 :         pg_log_error_hint("Try \"%s --help\" for more information.", progname);
    2244           0 :         exit(1);
    2245             :     }
    2246             : 
    2247             :     /* Required arguments */
    2248          32 :     if (subscriber_dir == NULL)
    2249             :     {
    2250           2 :         pg_log_error("no subscriber data directory specified");
    2251           2 :         pg_log_error_hint("Try \"%s --help\" for more information.", progname);
    2252           2 :         exit(1);
    2253             :     }
    2254             : 
    2255             :     /* If socket directory is not provided, use the current directory */
    2256          30 :     if (opt.socket_dir == NULL)
    2257             :     {
    2258             :         char        cwd[MAXPGPATH];
    2259             : 
    2260          10 :         if (!getcwd(cwd, MAXPGPATH))
    2261           0 :             pg_fatal("could not determine current directory");
    2262          10 :         opt.socket_dir = pg_strdup(cwd);
    2263          10 :         canonicalize_path(opt.socket_dir);
    2264             :     }
    2265             : 
    2266             :     /*
    2267             :      * Parse connection string. Build a base connection string that might be
    2268             :      * reused by multiple databases.
    2269             :      */
    2270          30 :     if (opt.pub_conninfo_str == NULL)
    2271             :     {
    2272             :         /*
    2273             :          * TODO use primary_conninfo (if available) from subscriber and
    2274             :          * extract publisher connection string. Assume that there are
    2275             :          * identical entries for physical and logical replication. If there is
    2276             :          * not, we would fail anyway.
    2277             :          */
    2278           2 :         pg_log_error("no publisher connection string specified");
    2279           2 :         pg_log_error_hint("Try \"%s --help\" for more information.", progname);
    2280           2 :         exit(1);
    2281             :     }
    2282          28 :     pg_log_info("validating publisher connection string");
    2283          28 :     pub_base_conninfo = get_base_conninfo(opt.pub_conninfo_str,
    2284             :                                           &dbname_conninfo);
    2285          28 :     if (pub_base_conninfo == NULL)
    2286           0 :         exit(1);
    2287             : 
    2288          28 :     pg_log_info("validating subscriber connection string");
    2289          28 :     sub_base_conninfo = get_sub_conninfo(&opt);
    2290             : 
    2291             :     /*
    2292             :      * Fetch all databases from the source (publisher) and treat them as if
    2293             :      * the user specified has multiple --database options, one for each source
    2294             :      * database.
    2295             :      */
    2296          28 :     if (opt.all_dbs)
    2297             :     {
    2298           2 :         bool        dbnamespecified = (dbname_conninfo != NULL);
    2299             : 
    2300           2 :         get_publisher_databases(&opt, dbnamespecified);
    2301             :     }
    2302             : 
    2303          28 :     if (opt.database_names.head == NULL)
    2304             :     {
    2305           4 :         pg_log_info("no database was specified");
    2306             : 
    2307             :         /*
    2308             :          * Try to obtain the dbname from the publisher conninfo. If dbname
    2309             :          * parameter is not available, error out.
    2310             :          */
    2311           4 :         if (dbname_conninfo)
    2312             :         {
    2313           2 :             simple_string_list_append(&opt.database_names, dbname_conninfo);
    2314           2 :             num_dbs++;
    2315             : 
    2316           2 :             pg_log_info("database name \"%s\" was extracted from the publisher connection string",
    2317             :                         dbname_conninfo);
    2318             :         }
    2319             :         else
    2320             :         {
    2321           2 :             pg_log_error("no database name specified");
    2322           2 :             pg_log_error_hint("Try \"%s --help\" for more information.",
    2323             :                               progname);
    2324           2 :             exit(1);
    2325             :         }
    2326             :     }
    2327             : 
    2328             :     /* Number of object names must match number of databases */
    2329          26 :     if (num_pubs > 0 && num_pubs != num_dbs)
    2330             :     {
    2331           2 :         pg_log_error("wrong number of publication names specified");
    2332           2 :         pg_log_error_detail("The number of specified publication names (%d) must match the number of specified database names (%d).",
    2333             :                             num_pubs, num_dbs);
    2334           2 :         exit(1);
    2335             :     }
    2336          24 :     if (num_subs > 0 && num_subs != num_dbs)
    2337             :     {
    2338           2 :         pg_log_error("wrong number of subscription names specified");
    2339           2 :         pg_log_error_detail("The number of specified subscription names (%d) must match the number of specified database names (%d).",
    2340             :                             num_subs, num_dbs);
    2341           2 :         exit(1);
    2342             :     }
    2343          22 :     if (num_replslots > 0 && num_replslots != num_dbs)
    2344             :     {
    2345           2 :         pg_log_error("wrong number of replication slot names specified");
    2346           2 :         pg_log_error_detail("The number of specified replication slot names (%d) must match the number of specified database names (%d).",
    2347             :                             num_replslots, num_dbs);
    2348           2 :         exit(1);
    2349             :     }
    2350             : 
    2351             :     /* Verify the object types specified for removal from the subscriber */
    2352          22 :     for (SimpleStringListCell *cell = opt.objecttypes_to_clean.head; cell; cell = cell->next)
    2353             :     {
    2354           2 :         if (pg_strcasecmp(cell->val, "publications") == 0)
    2355           2 :             dbinfos.objecttypes_to_clean |= OBJECTTYPE_PUBLICATIONS;
    2356             :         else
    2357             :         {
    2358           0 :             pg_log_error("invalid object type \"%s\" specified for --clean", cell->val);
    2359           0 :             pg_log_error_hint("The valid value is: \"%s\"", "publications");
    2360           0 :             exit(1);
    2361             :         }
    2362             :     }
    2363             : 
    2364             :     /* Get the absolute path of pg_ctl and pg_resetwal on the subscriber */
    2365          20 :     pg_ctl_path = get_exec_path(argv[0], "pg_ctl");
    2366          20 :     pg_resetwal_path = get_exec_path(argv[0], "pg_resetwal");
    2367             : 
    2368             :     /* Rudimentary check for a data directory */
    2369          20 :     check_data_directory(subscriber_dir);
    2370             : 
    2371          20 :     dbinfos.two_phase = opt.two_phase;
    2372             : 
    2373             :     /*
    2374             :      * Store database information for publisher and subscriber. It should be
    2375             :      * called before atexit() because its return is used in the
    2376             :      * cleanup_objects_atexit().
    2377             :      */
    2378          20 :     dbinfos.dbinfo = store_pub_sub_info(&opt, pub_base_conninfo, sub_base_conninfo);
    2379             : 
    2380             :     /* Register a function to clean up objects in case of failure */
    2381          20 :     atexit(cleanup_objects_atexit);
    2382             : 
    2383             :     /*
    2384             :      * Check if the subscriber data directory has the same system identifier
    2385             :      * than the publisher data directory.
    2386             :      */
    2387          20 :     pub_sysid = get_primary_sysid(dbinfos.dbinfo[0].pubconninfo);
    2388          20 :     sub_sysid = get_standby_sysid(subscriber_dir);
    2389          20 :     if (pub_sysid != sub_sysid)
    2390           2 :         pg_fatal("subscriber data directory is not a copy of the source database cluster");
    2391             : 
    2392             :     /* Subscriber PID file */
    2393          18 :     snprintf(pidfile, MAXPGPATH, "%s/postmaster.pid", subscriber_dir);
    2394             : 
    2395             :     /*
    2396             :      * The standby server must not be running. If the server is started under
    2397             :      * service manager and pg_createsubscriber stops it, the service manager
    2398             :      * might react to this action and start the server again. Therefore,
    2399             :      * refuse to proceed if the server is running to avoid possible failures.
    2400             :      */
    2401          18 :     if (stat(pidfile, &statbuf) == 0)
    2402             :     {
    2403           2 :         pg_log_error("standby server is running");
    2404           2 :         pg_log_error_hint("Stop the standby server and try again.");
    2405           2 :         exit(1);
    2406             :     }
    2407             : 
    2408             :     /*
    2409             :      * Start a short-lived standby server with temporary parameters (provided
    2410             :      * by command-line options). The goal is to avoid connections during the
    2411             :      * transformation steps.
    2412             :      */
    2413          16 :     pg_log_info("starting the standby server with command-line options");
    2414          16 :     start_standby_server(&opt, true, false);
    2415             : 
    2416             :     /* Check if the standby server is ready for logical replication */
    2417          16 :     check_subscriber(dbinfos.dbinfo);
    2418             : 
    2419             :     /* Check if the primary server is ready for logical replication */
    2420          12 :     check_publisher(dbinfos.dbinfo);
    2421             : 
    2422             :     /*
    2423             :      * Stop the target server. The recovery process requires that the server
    2424             :      * reaches a consistent state before targeting the recovery stop point.
    2425             :      * Make sure a consistent state is reached (stop the target server
    2426             :      * guarantees it) *before* creating the replication slots in
    2427             :      * setup_publisher().
    2428             :      */
    2429           8 :     pg_log_info("stopping the subscriber");
    2430           8 :     stop_standby_server(subscriber_dir);
    2431             : 
    2432             :     /* Create the required objects for each database on publisher */
    2433           8 :     consistent_lsn = setup_publisher(dbinfos.dbinfo);
    2434             : 
    2435             :     /* Write the required recovery parameters */
    2436           8 :     setup_recovery(dbinfos.dbinfo, subscriber_dir, consistent_lsn);
    2437             : 
    2438             :     /*
    2439             :      * Start subscriber so the recovery parameters will take effect. Wait
    2440             :      * until accepting connections. We don't want to start logical replication
    2441             :      * during setup.
    2442             :      */
    2443           8 :     pg_log_info("starting the subscriber");
    2444           8 :     start_standby_server(&opt, true, true);
    2445             : 
    2446             :     /* Waiting the subscriber to be promoted */
    2447           8 :     wait_for_end_recovery(dbinfos.dbinfo[0].subconninfo, &opt);
    2448             : 
    2449             :     /*
    2450             :      * Create the subscription for each database on subscriber. It does not
    2451             :      * enable it immediately because it needs to adjust the replication start
    2452             :      * point to the LSN reported by setup_publisher().  It also cleans up
    2453             :      * publications created by this tool and replication to the standby.
    2454             :      */
    2455           8 :     setup_subscriber(dbinfos.dbinfo, consistent_lsn);
    2456             : 
    2457             :     /* Remove primary_slot_name if it exists on primary */
    2458           8 :     drop_primary_replication_slot(dbinfos.dbinfo, primary_slot_name);
    2459             : 
    2460             :     /* Remove failover replication slots if they exist on subscriber */
    2461           8 :     drop_failover_replication_slots(dbinfos.dbinfo);
    2462             : 
    2463             :     /* Stop the subscriber */
    2464           8 :     pg_log_info("stopping the subscriber");
    2465           8 :     stop_standby_server(subscriber_dir);
    2466             : 
    2467             :     /* Change system identifier from subscriber */
    2468           8 :     modify_subscriber_sysid(&opt);
    2469             : 
    2470           8 :     success = true;
    2471             : 
    2472           8 :     pg_log_info("Done!");
    2473             : 
    2474           8 :     return 0;
    2475             : }
 |