LCOV - code coverage report
Current view: top level - src/bin/pg_basebackup - pg_createsubscriber.c (source / functions) Hit Total Coverage
Test: PostgreSQL 19devel Lines: 751 900 83.4 %
Date: 2025-08-09 08:18:06 Functions: 39 39 100.0 %
Legend: Lines: hit not hit

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

Generated by: LCOV version 1.16