LCOV - code coverage report
Current view: top level - src/bin/pg_basebackup - pg_createsubscriber.c (source / functions) Hit Total Coverage
Test: PostgreSQL 18devel Lines: 755 901 83.8 %
Date: 2025-04-01 16:15:31 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_remove; /* list of object types to remove */
      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_remove;  /* flags indicating which object types
      75             :                                          * to remove 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 or 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(_("  -R, --remove=OBJECTTYPE         remove all objects of the specified type from specified\n"
     257             :              "                                  databases on the subscriber; accepts: publications\n"));
     258           2 :     printf(_("  -s, --socketdir=DIR             socket directory to use (default current dir.)\n"));
     259           2 :     printf(_("  -t, --recovery-timeout=SECS     seconds to wait for recovery to end\n"));
     260           2 :     printf(_("  -T, --enable-two-phase          enable two-phase commit for all subscriptions\n"));
     261           2 :     printf(_("  -U, --subscriber-username=NAME  user name for subscriber connection\n"));
     262           2 :     printf(_("  -v, --verbose                   output verbose messages\n"));
     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        1344 :     for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
     319             :     {
     320        1316 :         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          38 : server_is_in_recovery(PGconn *conn)
     843             : {
     844             :     PGresult   *res;
     845             :     int         ret;
     846             : 
     847          38 :     res = PQexec(conn, "SELECT pg_catalog.pg_is_in_recovery()");
     848             : 
     849          38 :     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          38 :     ret = strcmp("t", PQgetvalue(res, 0, 0));
     858             : 
     859          38 :     PQclear(res);
     860             : 
     861          38 :     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 --enable-two-phase switch 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 :     appendPQExpBuffer(recoveryconfcontents, "recovery_target = ''\n");
    1251           8 :     appendPQExpBuffer(recoveryconfcontents,
    1252             :                       "recovery_target_timeline = 'latest'\n");
    1253           8 :     appendPQExpBuffer(recoveryconfcontents,
    1254             :                       "recovery_target_inclusive = true\n");
    1255           8 :     appendPQExpBuffer(recoveryconfcontents,
    1256             :                       "recovery_target_action = promote\n");
    1257           8 :     appendPQExpBuffer(recoveryconfcontents, "recovery_target_name = ''\n");
    1258           8 :     appendPQExpBuffer(recoveryconfcontents, "recovery_target_time = ''\n");
    1259           8 :     appendPQExpBuffer(recoveryconfcontents, "recovery_target_xid = ''\n");
    1260             : 
    1261           8 :     if (dry_run)
    1262             :     {
    1263           6 :         appendPQExpBuffer(recoveryconfcontents, "# dry run mode");
    1264           6 :         appendPQExpBuffer(recoveryconfcontents,
    1265             :                           "recovery_target_lsn = '%X/%X'\n",
    1266           6 :                           LSN_FORMAT_ARGS((XLogRecPtr) InvalidXLogRecPtr));
    1267             :     }
    1268             :     else
    1269             :     {
    1270           2 :         appendPQExpBuffer(recoveryconfcontents, "recovery_target_lsn = '%s'\n",
    1271             :                           lsn);
    1272           2 :         WriteRecoveryConfig(conn, datadir, recoveryconfcontents);
    1273             :     }
    1274           8 :     disconnect_database(conn, false);
    1275             : 
    1276           8 :     pg_log_debug("recovery parameters:\n%s", recoveryconfcontents->data);
    1277           8 : }
    1278             : 
    1279             : /*
    1280             :  * Drop physical replication slot on primary if the standby was using it. After
    1281             :  * the transformation, it has no use.
    1282             :  *
    1283             :  * XXX we might not fail here. Instead, we provide a warning so the user
    1284             :  * eventually drops this replication slot later.
    1285             :  */
    1286             : static void
    1287           8 : drop_primary_replication_slot(struct LogicalRepInfo *dbinfo, const char *slotname)
    1288             : {
    1289             :     PGconn     *conn;
    1290             : 
    1291             :     /* Replication slot does not exist, do nothing */
    1292           8 :     if (!primary_slot_name)
    1293           0 :         return;
    1294             : 
    1295           8 :     conn = connect_database(dbinfo[0].pubconninfo, false);
    1296           8 :     if (conn != NULL)
    1297             :     {
    1298           8 :         drop_replication_slot(conn, &dbinfo[0], slotname);
    1299           8 :         disconnect_database(conn, false);
    1300             :     }
    1301             :     else
    1302             :     {
    1303           0 :         pg_log_warning("could not drop replication slot \"%s\" on primary",
    1304             :                        slotname);
    1305           0 :         pg_log_warning_hint("Drop this replication slot soon to avoid retention of WAL files.");
    1306             :     }
    1307             : }
    1308             : 
    1309             : /*
    1310             :  * Drop failover replication slots on subscriber. After the transformation,
    1311             :  * they have no use.
    1312             :  *
    1313             :  * XXX We do not fail here. Instead, we provide a warning so the user can drop
    1314             :  * them later.
    1315             :  */
    1316             : static void
    1317           8 : drop_failover_replication_slots(struct LogicalRepInfo *dbinfo)
    1318             : {
    1319             :     PGconn     *conn;
    1320             :     PGresult   *res;
    1321             : 
    1322           8 :     conn = connect_database(dbinfo[0].subconninfo, false);
    1323           8 :     if (conn != NULL)
    1324             :     {
    1325             :         /* Get failover replication slot names */
    1326           8 :         res = PQexec(conn,
    1327             :                      "SELECT slot_name FROM pg_catalog.pg_replication_slots WHERE failover");
    1328             : 
    1329           8 :         if (PQresultStatus(res) == PGRES_TUPLES_OK)
    1330             :         {
    1331             :             /* Remove failover replication slots from subscriber */
    1332          16 :             for (int i = 0; i < PQntuples(res); i++)
    1333           8 :                 drop_replication_slot(conn, &dbinfo[0], PQgetvalue(res, i, 0));
    1334             :         }
    1335             :         else
    1336             :         {
    1337           0 :             pg_log_warning("could not obtain failover replication slot information: %s",
    1338             :                            PQresultErrorMessage(res));
    1339           0 :             pg_log_warning_hint("Drop the failover replication slots on subscriber soon to avoid retention of WAL files.");
    1340             :         }
    1341             : 
    1342           8 :         PQclear(res);
    1343           8 :         disconnect_database(conn, false);
    1344             :     }
    1345             :     else
    1346             :     {
    1347           0 :         pg_log_warning("could not drop failover replication slot");
    1348           0 :         pg_log_warning_hint("Drop the failover replication slots on subscriber soon to avoid retention of WAL files.");
    1349             :     }
    1350           8 : }
    1351             : 
    1352             : /*
    1353             :  * Create a logical replication slot and returns a LSN.
    1354             :  *
    1355             :  * CreateReplicationSlot() is not used because it does not provide the one-row
    1356             :  * result set that contains the LSN.
    1357             :  */
    1358             : static char *
    1359          16 : create_logical_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo)
    1360             : {
    1361          16 :     PQExpBuffer str = createPQExpBuffer();
    1362          16 :     PGresult   *res = NULL;
    1363          16 :     const char *slot_name = dbinfo->replslotname;
    1364             :     char       *slot_name_esc;
    1365          16 :     char       *lsn = NULL;
    1366             : 
    1367             :     Assert(conn != NULL);
    1368             : 
    1369          16 :     pg_log_info("creating the replication slot \"%s\" in database \"%s\"",
    1370             :                 slot_name, dbinfo->dbname);
    1371             : 
    1372          16 :     slot_name_esc = PQescapeLiteral(conn, slot_name, strlen(slot_name));
    1373             : 
    1374          16 :     appendPQExpBuffer(str,
    1375             :                       "SELECT lsn FROM pg_catalog.pg_create_logical_replication_slot(%s, 'pgoutput', false, %s, false)",
    1376             :                       slot_name_esc,
    1377          16 :                       dbinfos.two_phase ? "true" : "false");
    1378             : 
    1379          16 :     PQfreemem(slot_name_esc);
    1380             : 
    1381          16 :     pg_log_debug("command is: %s", str->data);
    1382             : 
    1383          16 :     if (!dry_run)
    1384             :     {
    1385           4 :         res = PQexec(conn, str->data);
    1386           4 :         if (PQresultStatus(res) != PGRES_TUPLES_OK)
    1387             :         {
    1388           0 :             pg_log_error("could not create replication slot \"%s\" in database \"%s\": %s",
    1389             :                          slot_name, dbinfo->dbname,
    1390             :                          PQresultErrorMessage(res));
    1391           0 :             PQclear(res);
    1392           0 :             destroyPQExpBuffer(str);
    1393           0 :             return NULL;
    1394             :         }
    1395             : 
    1396           4 :         lsn = pg_strdup(PQgetvalue(res, 0, 0));
    1397           4 :         PQclear(res);
    1398             :     }
    1399             : 
    1400             :     /* For cleanup purposes */
    1401          16 :     dbinfo->made_replslot = true;
    1402             : 
    1403          16 :     destroyPQExpBuffer(str);
    1404             : 
    1405          16 :     return lsn;
    1406             : }
    1407             : 
    1408             : static void
    1409          16 : drop_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo,
    1410             :                       const char *slot_name)
    1411             : {
    1412          16 :     PQExpBuffer str = createPQExpBuffer();
    1413             :     char       *slot_name_esc;
    1414             :     PGresult   *res;
    1415             : 
    1416             :     Assert(conn != NULL);
    1417             : 
    1418          16 :     pg_log_info("dropping the replication slot \"%s\" in database \"%s\"",
    1419             :                 slot_name, dbinfo->dbname);
    1420             : 
    1421          16 :     slot_name_esc = PQescapeLiteral(conn, slot_name, strlen(slot_name));
    1422             : 
    1423          16 :     appendPQExpBuffer(str, "SELECT pg_catalog.pg_drop_replication_slot(%s)", slot_name_esc);
    1424             : 
    1425          16 :     PQfreemem(slot_name_esc);
    1426             : 
    1427          16 :     pg_log_debug("command is: %s", str->data);
    1428             : 
    1429          16 :     if (!dry_run)
    1430             :     {
    1431           4 :         res = PQexec(conn, str->data);
    1432           4 :         if (PQresultStatus(res) != PGRES_TUPLES_OK)
    1433             :         {
    1434           0 :             pg_log_error("could not drop replication slot \"%s\" in database \"%s\": %s",
    1435             :                          slot_name, dbinfo->dbname, PQresultErrorMessage(res));
    1436           0 :             dbinfo->made_replslot = false;   /* don't try again. */
    1437             :         }
    1438             : 
    1439           4 :         PQclear(res);
    1440             :     }
    1441             : 
    1442          16 :     destroyPQExpBuffer(str);
    1443          16 : }
    1444             : 
    1445             : /*
    1446             :  * Reports a suitable message if pg_ctl fails.
    1447             :  */
    1448             : static void
    1449          48 : pg_ctl_status(const char *pg_ctl_cmd, int rc)
    1450             : {
    1451          48 :     if (rc != 0)
    1452             :     {
    1453           0 :         if (WIFEXITED(rc))
    1454             :         {
    1455           0 :             pg_log_error("pg_ctl failed with exit code %d", WEXITSTATUS(rc));
    1456             :         }
    1457           0 :         else if (WIFSIGNALED(rc))
    1458             :         {
    1459             : #if defined(WIN32)
    1460             :             pg_log_error("pg_ctl was terminated by exception 0x%X",
    1461             :                          WTERMSIG(rc));
    1462             :             pg_log_error_detail("See C include file \"ntstatus.h\" for a description of the hexadecimal value.");
    1463             : #else
    1464           0 :             pg_log_error("pg_ctl was terminated by signal %d: %s",
    1465             :                          WTERMSIG(rc), pg_strsignal(WTERMSIG(rc)));
    1466             : #endif
    1467             :         }
    1468             :         else
    1469             :         {
    1470           0 :             pg_log_error("pg_ctl exited with unrecognized status %d", rc);
    1471             :         }
    1472             : 
    1473           0 :         pg_log_error_detail("The failed command was: %s", pg_ctl_cmd);
    1474           0 :         exit(1);
    1475             :     }
    1476          48 : }
    1477             : 
    1478             : static void
    1479          24 : start_standby_server(const struct CreateSubscriberOptions *opt, bool restricted_access,
    1480             :                      bool restrict_logical_worker)
    1481             : {
    1482          24 :     PQExpBuffer pg_ctl_cmd = createPQExpBuffer();
    1483             :     int         rc;
    1484             : 
    1485          24 :     appendPQExpBuffer(pg_ctl_cmd, "\"%s\" start -D ", pg_ctl_path);
    1486          24 :     appendShellString(pg_ctl_cmd, subscriber_dir);
    1487          24 :     appendPQExpBuffer(pg_ctl_cmd, " -s -o \"-c sync_replication_slots=off\"");
    1488             : 
    1489             :     /* Prevent unintended slot invalidation */
    1490          24 :     appendPQExpBuffer(pg_ctl_cmd, " -o \"-c idle_replication_slot_timeout=0\"");
    1491             : 
    1492          24 :     if (restricted_access)
    1493             :     {
    1494          24 :         appendPQExpBuffer(pg_ctl_cmd, " -o \"-p %s\"", opt->sub_port);
    1495             : #if !defined(WIN32)
    1496             : 
    1497             :         /*
    1498             :          * An empty listen_addresses list means the server does not listen on
    1499             :          * any IP interfaces; only Unix-domain sockets can be used to connect
    1500             :          * to the server. Prevent external connections to minimize the chance
    1501             :          * of failure.
    1502             :          */
    1503          24 :         appendPQExpBufferStr(pg_ctl_cmd, " -o \"-c listen_addresses='' -c unix_socket_permissions=0700");
    1504          24 :         if (opt->socket_dir)
    1505          24 :             appendPQExpBuffer(pg_ctl_cmd, " -c unix_socket_directories='%s'",
    1506             :                               opt->socket_dir);
    1507          24 :         appendPQExpBufferChar(pg_ctl_cmd, '"');
    1508             : #endif
    1509             :     }
    1510          24 :     if (opt->config_file != NULL)
    1511           0 :         appendPQExpBuffer(pg_ctl_cmd, " -o \"-c config_file=%s\"",
    1512             :                           opt->config_file);
    1513             : 
    1514             :     /* Suppress to start logical replication if requested */
    1515          24 :     if (restrict_logical_worker)
    1516           8 :         appendPQExpBuffer(pg_ctl_cmd, " -o \"-c max_logical_replication_workers=0\"");
    1517             : 
    1518          24 :     pg_log_debug("pg_ctl command is: %s", pg_ctl_cmd->data);
    1519          24 :     rc = system(pg_ctl_cmd->data);
    1520          24 :     pg_ctl_status(pg_ctl_cmd->data, rc);
    1521          24 :     standby_running = true;
    1522          24 :     destroyPQExpBuffer(pg_ctl_cmd);
    1523          24 :     pg_log_info("server was started");
    1524          24 : }
    1525             : 
    1526             : static void
    1527          24 : stop_standby_server(const char *datadir)
    1528             : {
    1529             :     char       *pg_ctl_cmd;
    1530             :     int         rc;
    1531             : 
    1532          24 :     pg_ctl_cmd = psprintf("\"%s\" stop -D \"%s\" -s", pg_ctl_path,
    1533             :                           datadir);
    1534          24 :     pg_log_debug("pg_ctl command is: %s", pg_ctl_cmd);
    1535          24 :     rc = system(pg_ctl_cmd);
    1536          24 :     pg_ctl_status(pg_ctl_cmd, rc);
    1537          24 :     standby_running = false;
    1538          24 :     pg_log_info("server was stopped");
    1539          24 : }
    1540             : 
    1541             : /*
    1542             :  * Returns after the server finishes the recovery process.
    1543             :  *
    1544             :  * If recovery_timeout option is set, terminate abnormally without finishing
    1545             :  * the recovery process. By default, it waits forever.
    1546             :  *
    1547             :  * XXX Is the recovery process still in progress? When recovery process has a
    1548             :  * better progress reporting mechanism, it should be added here.
    1549             :  */
    1550             : static void
    1551           8 : wait_for_end_recovery(const char *conninfo, const struct CreateSubscriberOptions *opt)
    1552             : {
    1553             :     PGconn     *conn;
    1554           8 :     int         status = POSTMASTER_STILL_STARTING;
    1555           8 :     int         timer = 0;
    1556             : 
    1557           8 :     pg_log_info("waiting for the target server to reach the consistent state");
    1558             : 
    1559           8 :     conn = connect_database(conninfo, true);
    1560             : 
    1561             :     for (;;)
    1562           2 :     {
    1563          10 :         bool        in_recovery = server_is_in_recovery(conn);
    1564             : 
    1565             :         /*
    1566             :          * Does the recovery process finish? In dry run mode, there is no
    1567             :          * recovery mode. Bail out as the recovery process has ended.
    1568             :          */
    1569          10 :         if (!in_recovery || dry_run)
    1570             :         {
    1571           8 :             status = POSTMASTER_READY;
    1572           8 :             recovery_ended = true;
    1573           8 :             break;
    1574             :         }
    1575             : 
    1576             :         /* Bail out after recovery_timeout seconds if this option is set */
    1577           2 :         if (opt->recovery_timeout > 0 && timer >= opt->recovery_timeout)
    1578             :         {
    1579           0 :             stop_standby_server(subscriber_dir);
    1580           0 :             pg_log_error("recovery timed out");
    1581           0 :             disconnect_database(conn, true);
    1582             :         }
    1583             : 
    1584             :         /* Keep waiting */
    1585           2 :         pg_usleep(WAIT_INTERVAL * USEC_PER_SEC);
    1586             : 
    1587           2 :         timer += WAIT_INTERVAL;
    1588             :     }
    1589             : 
    1590           8 :     disconnect_database(conn, false);
    1591             : 
    1592           8 :     if (status == POSTMASTER_STILL_STARTING)
    1593           0 :         pg_fatal("server did not end recovery");
    1594             : 
    1595           8 :     pg_log_info("target server reached the consistent state");
    1596           8 :     pg_log_info_hint("If pg_createsubscriber fails after this point, you must recreate the physical replica before continuing.");
    1597           8 : }
    1598             : 
    1599             : /*
    1600             :  * Create a publication that includes all tables in the database.
    1601             :  */
    1602             : static void
    1603          16 : create_publication(PGconn *conn, struct LogicalRepInfo *dbinfo)
    1604             : {
    1605          16 :     PQExpBuffer str = createPQExpBuffer();
    1606             :     PGresult   *res;
    1607             :     char       *ipubname_esc;
    1608             :     char       *spubname_esc;
    1609             : 
    1610             :     Assert(conn != NULL);
    1611             : 
    1612          16 :     ipubname_esc = PQescapeIdentifier(conn, dbinfo->pubname, strlen(dbinfo->pubname));
    1613          16 :     spubname_esc = PQescapeLiteral(conn, dbinfo->pubname, strlen(dbinfo->pubname));
    1614             : 
    1615             :     /* Check if the publication already exists */
    1616          16 :     appendPQExpBuffer(str,
    1617             :                       "SELECT 1 FROM pg_catalog.pg_publication "
    1618             :                       "WHERE pubname = %s",
    1619             :                       spubname_esc);
    1620          16 :     res = PQexec(conn, str->data);
    1621          16 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
    1622             :     {
    1623           0 :         pg_log_error("could not obtain publication information: %s",
    1624             :                      PQresultErrorMessage(res));
    1625           0 :         disconnect_database(conn, true);
    1626             :     }
    1627             : 
    1628          16 :     if (PQntuples(res) == 1)
    1629             :     {
    1630             :         /*
    1631             :          * Unfortunately, if it reaches this code path, it will always fail
    1632             :          * (unless you decide to change the existing publication name). That's
    1633             :          * bad but it is very unlikely that the user will choose a name with
    1634             :          * pg_createsubscriber_ prefix followed by the exact database oid and
    1635             :          * a random number.
    1636             :          */
    1637           0 :         pg_log_error("publication \"%s\" already exists", dbinfo->pubname);
    1638           0 :         pg_log_error_hint("Consider renaming this publication before continuing.");
    1639           0 :         disconnect_database(conn, true);
    1640             :     }
    1641             : 
    1642          16 :     PQclear(res);
    1643          16 :     resetPQExpBuffer(str);
    1644             : 
    1645          16 :     pg_log_info("creating publication \"%s\" in database \"%s\"",
    1646             :                 dbinfo->pubname, dbinfo->dbname);
    1647             : 
    1648          16 :     appendPQExpBuffer(str, "CREATE PUBLICATION %s FOR ALL TABLES",
    1649             :                       ipubname_esc);
    1650             : 
    1651          16 :     pg_log_debug("command is: %s", str->data);
    1652             : 
    1653          16 :     if (!dry_run)
    1654             :     {
    1655           4 :         res = PQexec(conn, str->data);
    1656           4 :         if (PQresultStatus(res) != PGRES_COMMAND_OK)
    1657             :         {
    1658           0 :             pg_log_error("could not create publication \"%s\" in database \"%s\": %s",
    1659             :                          dbinfo->pubname, dbinfo->dbname, PQresultErrorMessage(res));
    1660           0 :             disconnect_database(conn, true);
    1661             :         }
    1662           4 :         PQclear(res);
    1663             :     }
    1664             : 
    1665             :     /* For cleanup purposes */
    1666          16 :     dbinfo->made_publication = true;
    1667             : 
    1668          16 :     PQfreemem(ipubname_esc);
    1669          16 :     PQfreemem(spubname_esc);
    1670          16 :     destroyPQExpBuffer(str);
    1671          16 : }
    1672             : 
    1673             : /*
    1674             :  * Drop the specified publication in the given database.
    1675             :  */
    1676             : static void
    1677          20 : drop_publication(PGconn *conn, const char *pubname, const char *dbname,
    1678             :                  bool *made_publication)
    1679             : {
    1680          20 :     PQExpBuffer str = createPQExpBuffer();
    1681             :     PGresult   *res;
    1682             :     char       *pubname_esc;
    1683             : 
    1684             :     Assert(conn != NULL);
    1685             : 
    1686          20 :     pubname_esc = PQescapeIdentifier(conn, pubname, strlen(pubname));
    1687             : 
    1688          20 :     pg_log_info("dropping publication \"%s\" in database \"%s\"",
    1689             :                 pubname, dbname);
    1690             : 
    1691          20 :     appendPQExpBuffer(str, "DROP PUBLICATION %s", pubname_esc);
    1692             : 
    1693          20 :     PQfreemem(pubname_esc);
    1694             : 
    1695          20 :     pg_log_debug("command is: %s", str->data);
    1696             : 
    1697          20 :     if (!dry_run)
    1698             :     {
    1699           8 :         res = PQexec(conn, str->data);
    1700           8 :         if (PQresultStatus(res) != PGRES_COMMAND_OK)
    1701             :         {
    1702           0 :             pg_log_error("could not drop publication \"%s\" in database \"%s\": %s",
    1703             :                          pubname, dbname, PQresultErrorMessage(res));
    1704           0 :             *made_publication = false;  /* don't try again. */
    1705             : 
    1706             :             /*
    1707             :              * Don't disconnect and exit here. This routine is used by primary
    1708             :              * (cleanup publication / replication slot due to an error) and
    1709             :              * subscriber (remove the replicated publications). In both cases,
    1710             :              * it can continue and provide instructions for the user to remove
    1711             :              * it later if cleanup fails.
    1712             :              */
    1713             :         }
    1714           8 :         PQclear(res);
    1715             :     }
    1716             : 
    1717          20 :     destroyPQExpBuffer(str);
    1718          20 : }
    1719             : 
    1720             : /*
    1721             :  * Retrieve and drop the publications.
    1722             :  *
    1723             :  * Since the publications were created before the consistent LSN, they
    1724             :  * remain on the subscriber even after the physical replica is
    1725             :  * promoted. Remove these publications from the subscriber because
    1726             :  * they have no use. Additionally, if requested, drop all pre-existing
    1727             :  * publications.
    1728             :  */
    1729             : static void
    1730          16 : check_and_drop_publications(PGconn *conn, struct LogicalRepInfo *dbinfo)
    1731             : {
    1732             :     PGresult   *res;
    1733          16 :     bool        drop_all_pubs = dbinfos.objecttypes_to_remove & OBJECTTYPE_PUBLICATIONS;
    1734             : 
    1735             :     Assert(conn != NULL);
    1736             : 
    1737          16 :     if (drop_all_pubs)
    1738             :     {
    1739           4 :         pg_log_info("dropping all existing publications in database \"%s\"",
    1740             :                     dbinfo->dbname);
    1741             : 
    1742             :         /* Fetch all publication names */
    1743           4 :         res = PQexec(conn, "SELECT pubname FROM pg_catalog.pg_publication;");
    1744           4 :         if (PQresultStatus(res) != PGRES_TUPLES_OK)
    1745             :         {
    1746           0 :             pg_log_error("could not obtain publication information: %s",
    1747             :                          PQresultErrorMessage(res));
    1748           0 :             PQclear(res);
    1749           0 :             disconnect_database(conn, true);
    1750             :         }
    1751             : 
    1752             :         /* Drop each publication */
    1753          12 :         for (int i = 0; i < PQntuples(res); i++)
    1754           8 :             drop_publication(conn, PQgetvalue(res, i, 0), dbinfo->dbname,
    1755             :                              &dbinfo->made_publication);
    1756             : 
    1757           4 :         PQclear(res);
    1758             :     }
    1759             : 
    1760             :     /*
    1761             :      * In dry-run mode, we don't create publications, but we still try to drop
    1762             :      * those to provide necessary information to the user.
    1763             :      */
    1764          16 :     if (!drop_all_pubs || dry_run)
    1765          12 :         drop_publication(conn, dbinfo->pubname, dbinfo->dbname,
    1766             :                          &dbinfo->made_publication);
    1767          16 : }
    1768             : 
    1769             : /*
    1770             :  * Create a subscription with some predefined options.
    1771             :  *
    1772             :  * A replication slot was already created in a previous step. Let's use it.  It
    1773             :  * is not required to copy data. The subscription will be created but it will
    1774             :  * not be enabled now. That's because the replication progress must be set and
    1775             :  * the replication origin name (one of the function arguments) contains the
    1776             :  * subscription OID in its name. Once the subscription is created,
    1777             :  * set_replication_progress() can obtain the chosen origin name and set up its
    1778             :  * initial location.
    1779             :  */
    1780             : static void
    1781          16 : create_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo)
    1782             : {
    1783          16 :     PQExpBuffer str = createPQExpBuffer();
    1784             :     PGresult   *res;
    1785             :     char       *pubname_esc;
    1786             :     char       *subname_esc;
    1787             :     char       *pubconninfo_esc;
    1788             :     char       *replslotname_esc;
    1789             : 
    1790             :     Assert(conn != NULL);
    1791             : 
    1792          16 :     pubname_esc = PQescapeIdentifier(conn, dbinfo->pubname, strlen(dbinfo->pubname));
    1793          16 :     subname_esc = PQescapeIdentifier(conn, dbinfo->subname, strlen(dbinfo->subname));
    1794          16 :     pubconninfo_esc = PQescapeLiteral(conn, dbinfo->pubconninfo, strlen(dbinfo->pubconninfo));
    1795          16 :     replslotname_esc = PQescapeLiteral(conn, dbinfo->replslotname, strlen(dbinfo->replslotname));
    1796             : 
    1797          16 :     pg_log_info("creating subscription \"%s\" in database \"%s\"",
    1798             :                 dbinfo->subname, dbinfo->dbname);
    1799             : 
    1800          16 :     appendPQExpBuffer(str,
    1801             :                       "CREATE SUBSCRIPTION %s CONNECTION %s PUBLICATION %s "
    1802             :                       "WITH (create_slot = false, enabled = false, "
    1803             :                       "slot_name = %s, copy_data = false, two_phase = %s)",
    1804             :                       subname_esc, pubconninfo_esc, pubname_esc, replslotname_esc,
    1805          16 :                       dbinfos.two_phase ? "true" : "false");
    1806             : 
    1807          16 :     PQfreemem(pubname_esc);
    1808          16 :     PQfreemem(subname_esc);
    1809          16 :     PQfreemem(pubconninfo_esc);
    1810          16 :     PQfreemem(replslotname_esc);
    1811             : 
    1812          16 :     pg_log_debug("command is: %s", str->data);
    1813             : 
    1814          16 :     if (!dry_run)
    1815             :     {
    1816           4 :         res = PQexec(conn, str->data);
    1817           4 :         if (PQresultStatus(res) != PGRES_COMMAND_OK)
    1818             :         {
    1819           0 :             pg_log_error("could not create subscription \"%s\" in database \"%s\": %s",
    1820             :                          dbinfo->subname, dbinfo->dbname, PQresultErrorMessage(res));
    1821           0 :             disconnect_database(conn, true);
    1822             :         }
    1823           4 :         PQclear(res);
    1824             :     }
    1825             : 
    1826          16 :     destroyPQExpBuffer(str);
    1827          16 : }
    1828             : 
    1829             : /*
    1830             :  * Sets the replication progress to the consistent LSN.
    1831             :  *
    1832             :  * The subscriber caught up to the consistent LSN provided by the last
    1833             :  * replication slot that was created. The goal is to set up the initial
    1834             :  * location for the logical replication that is the exact LSN that the
    1835             :  * subscriber was promoted. Once the subscription is enabled it will start
    1836             :  * streaming from that location onwards.  In dry run mode, the subscription OID
    1837             :  * and LSN are set to invalid values for printing purposes.
    1838             :  */
    1839             : static void
    1840          16 : set_replication_progress(PGconn *conn, const struct LogicalRepInfo *dbinfo, const char *lsn)
    1841             : {
    1842          16 :     PQExpBuffer str = createPQExpBuffer();
    1843             :     PGresult   *res;
    1844             :     Oid         suboid;
    1845             :     char       *subname;
    1846             :     char       *dbname;
    1847             :     char       *originname;
    1848             :     char       *lsnstr;
    1849             : 
    1850             :     Assert(conn != NULL);
    1851             : 
    1852          16 :     subname = PQescapeLiteral(conn, dbinfo->subname, strlen(dbinfo->subname));
    1853          16 :     dbname = PQescapeLiteral(conn, dbinfo->dbname, strlen(dbinfo->dbname));
    1854             : 
    1855          16 :     appendPQExpBuffer(str,
    1856             :                       "SELECT s.oid FROM pg_catalog.pg_subscription s "
    1857             :                       "INNER JOIN pg_catalog.pg_database d ON (s.subdbid = d.oid) "
    1858             :                       "WHERE s.subname = %s AND d.datname = %s",
    1859             :                       subname, dbname);
    1860             : 
    1861          16 :     res = PQexec(conn, str->data);
    1862          16 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
    1863             :     {
    1864           0 :         pg_log_error("could not obtain subscription OID: %s",
    1865             :                      PQresultErrorMessage(res));
    1866           0 :         disconnect_database(conn, true);
    1867             :     }
    1868             : 
    1869          16 :     if (PQntuples(res) != 1 && !dry_run)
    1870             :     {
    1871           0 :         pg_log_error("could not obtain subscription OID: got %d rows, expected %d row",
    1872             :                      PQntuples(res), 1);
    1873           0 :         disconnect_database(conn, true);
    1874             :     }
    1875             : 
    1876          16 :     if (dry_run)
    1877             :     {
    1878          12 :         suboid = InvalidOid;
    1879          12 :         lsnstr = psprintf("%X/%X", LSN_FORMAT_ARGS((XLogRecPtr) InvalidXLogRecPtr));
    1880             :     }
    1881             :     else
    1882             :     {
    1883           4 :         suboid = strtoul(PQgetvalue(res, 0, 0), NULL, 10);
    1884           4 :         lsnstr = psprintf("%s", lsn);
    1885             :     }
    1886             : 
    1887          16 :     PQclear(res);
    1888             : 
    1889             :     /*
    1890             :      * The origin name is defined as pg_%u. %u is the subscription OID. See
    1891             :      * ApplyWorkerMain().
    1892             :      */
    1893          16 :     originname = psprintf("pg_%u", suboid);
    1894             : 
    1895          16 :     pg_log_info("setting the replication progress (node name \"%s\", LSN %s) in database \"%s\"",
    1896             :                 originname, lsnstr, dbinfo->dbname);
    1897             : 
    1898          16 :     resetPQExpBuffer(str);
    1899          16 :     appendPQExpBuffer(str,
    1900             :                       "SELECT pg_catalog.pg_replication_origin_advance('%s', '%s')",
    1901             :                       originname, lsnstr);
    1902             : 
    1903          16 :     pg_log_debug("command is: %s", str->data);
    1904             : 
    1905          16 :     if (!dry_run)
    1906             :     {
    1907           4 :         res = PQexec(conn, str->data);
    1908           4 :         if (PQresultStatus(res) != PGRES_TUPLES_OK)
    1909             :         {
    1910           0 :             pg_log_error("could not set replication progress for subscription \"%s\": %s",
    1911             :                          dbinfo->subname, PQresultErrorMessage(res));
    1912           0 :             disconnect_database(conn, true);
    1913             :         }
    1914           4 :         PQclear(res);
    1915             :     }
    1916             : 
    1917          16 :     PQfreemem(subname);
    1918          16 :     PQfreemem(dbname);
    1919          16 :     pg_free(originname);
    1920          16 :     pg_free(lsnstr);
    1921          16 :     destroyPQExpBuffer(str);
    1922          16 : }
    1923             : 
    1924             : /*
    1925             :  * Enables the subscription.
    1926             :  *
    1927             :  * The subscription was created in a previous step but it was disabled. After
    1928             :  * adjusting the initial logical replication location, enable the subscription.
    1929             :  */
    1930             : static void
    1931          16 : enable_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo)
    1932             : {
    1933          16 :     PQExpBuffer str = createPQExpBuffer();
    1934             :     PGresult   *res;
    1935             :     char       *subname;
    1936             : 
    1937             :     Assert(conn != NULL);
    1938             : 
    1939          16 :     subname = PQescapeIdentifier(conn, dbinfo->subname, strlen(dbinfo->subname));
    1940             : 
    1941          16 :     pg_log_info("enabling subscription \"%s\" in database \"%s\"",
    1942             :                 dbinfo->subname, dbinfo->dbname);
    1943             : 
    1944          16 :     appendPQExpBuffer(str, "ALTER SUBSCRIPTION %s ENABLE", subname);
    1945             : 
    1946          16 :     pg_log_debug("command is: %s", str->data);
    1947             : 
    1948          16 :     if (!dry_run)
    1949             :     {
    1950           4 :         res = PQexec(conn, str->data);
    1951           4 :         if (PQresultStatus(res) != PGRES_COMMAND_OK)
    1952             :         {
    1953           0 :             pg_log_error("could not enable subscription \"%s\": %s",
    1954             :                          dbinfo->subname, PQresultErrorMessage(res));
    1955           0 :             disconnect_database(conn, true);
    1956             :         }
    1957             : 
    1958           4 :         PQclear(res);
    1959             :     }
    1960             : 
    1961          16 :     PQfreemem(subname);
    1962          16 :     destroyPQExpBuffer(str);
    1963          16 : }
    1964             : 
    1965             : /*
    1966             :  * Fetch a list of all not-template databases from the source server and form
    1967             :  * a list such that they appear as if the user has specified multiple
    1968             :  * --database options, one for each source database.
    1969             :  */
    1970             : static void
    1971           2 : get_publisher_databases(struct CreateSubscriberOptions *opt,
    1972             :                         bool dbnamespecified)
    1973             : {
    1974             :     PGconn     *conn;
    1975             :     PGresult   *res;
    1976             : 
    1977             :     /* If a database name was specified, just connect to it. */
    1978           2 :     if (dbnamespecified)
    1979           0 :         conn = connect_database(opt->pub_conninfo_str, true);
    1980             :     else
    1981             :     {
    1982             :         /* Otherwise, try postgres first and then template1. */
    1983             :         char       *conninfo;
    1984             : 
    1985           2 :         conninfo = concat_conninfo_dbname(opt->pub_conninfo_str, "postgres");
    1986           2 :         conn = connect_database(conninfo, false);
    1987           2 :         pg_free(conninfo);
    1988           2 :         if (!conn)
    1989             :         {
    1990           0 :             conninfo = concat_conninfo_dbname(opt->pub_conninfo_str, "template1");
    1991           0 :             conn = connect_database(conninfo, true);
    1992           0 :             pg_free(conninfo);
    1993             :         }
    1994             :     }
    1995             : 
    1996           2 :     res = PQexec(conn, "SELECT datname FROM pg_database WHERE datistemplate = false AND datallowconn AND datconnlimit <> -2 ORDER BY 1");
    1997           2 :     if (PQresultStatus(res) != PGRES_TUPLES_OK)
    1998             :     {
    1999           0 :         pg_log_error("could not obtain a list of databases: %s", PQresultErrorMessage(res));
    2000           0 :         PQclear(res);
    2001           0 :         disconnect_database(conn, true);
    2002             :     }
    2003             : 
    2004           8 :     for (int i = 0; i < PQntuples(res); i++)
    2005             :     {
    2006           6 :         const char *dbname = PQgetvalue(res, i, 0);
    2007             : 
    2008           6 :         simple_string_list_append(&opt->database_names, dbname);
    2009             : 
    2010             :         /* Increment num_dbs to reflect multiple --database options */
    2011           6 :         num_dbs++;
    2012             :     }
    2013             : 
    2014           2 :     PQclear(res);
    2015           2 :     disconnect_database(conn, false);
    2016           2 : }
    2017             : 
    2018             : int
    2019          46 : main(int argc, char **argv)
    2020             : {
    2021             :     static struct option long_options[] =
    2022             :     {
    2023             :         {"all", no_argument, NULL, 'a'},
    2024             :         {"database", required_argument, NULL, 'd'},
    2025             :         {"pgdata", required_argument, NULL, 'D'},
    2026             :         {"dry-run", no_argument, NULL, 'n'},
    2027             :         {"subscriber-port", required_argument, NULL, 'p'},
    2028             :         {"publisher-server", required_argument, NULL, 'P'},
    2029             :         {"remove", required_argument, NULL, 'R'},
    2030             :         {"socketdir", required_argument, NULL, 's'},
    2031             :         {"recovery-timeout", required_argument, NULL, 't'},
    2032             :         {"enable-two-phase", no_argument, NULL, 'T'},
    2033             :         {"subscriber-username", required_argument, NULL, 'U'},
    2034             :         {"verbose", no_argument, NULL, 'v'},
    2035             :         {"version", no_argument, NULL, 'V'},
    2036             :         {"help", no_argument, NULL, '?'},
    2037             :         {"config-file", required_argument, NULL, 1},
    2038             :         {"publication", required_argument, NULL, 2},
    2039             :         {"replication-slot", required_argument, NULL, 3},
    2040             :         {"subscription", required_argument, NULL, 4},
    2041             :         {NULL, 0, NULL, 0}
    2042             :     };
    2043             : 
    2044          46 :     struct CreateSubscriberOptions opt = {0};
    2045             : 
    2046             :     int         c;
    2047             :     int         option_index;
    2048             : 
    2049             :     char       *pub_base_conninfo;
    2050             :     char       *sub_base_conninfo;
    2051          46 :     char       *dbname_conninfo = NULL;
    2052             : 
    2053             :     uint64      pub_sysid;
    2054             :     uint64      sub_sysid;
    2055             :     struct stat statbuf;
    2056             : 
    2057             :     char       *consistent_lsn;
    2058             : 
    2059             :     char        pidfile[MAXPGPATH];
    2060             : 
    2061          46 :     pg_logging_init(argv[0]);
    2062          46 :     pg_logging_set_level(PG_LOG_WARNING);
    2063          46 :     progname = get_progname(argv[0]);
    2064          46 :     set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_basebackup"));
    2065             : 
    2066          46 :     if (argc > 1)
    2067             :     {
    2068          44 :         if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
    2069             :         {
    2070           2 :             usage();
    2071           2 :             exit(0);
    2072             :         }
    2073          42 :         else if (strcmp(argv[1], "-V") == 0
    2074          42 :                  || strcmp(argv[1], "--version") == 0)
    2075             :         {
    2076           2 :             puts("pg_createsubscriber (PostgreSQL) " PG_VERSION);
    2077           2 :             exit(0);
    2078             :         }
    2079             :     }
    2080             : 
    2081             :     /* Default settings */
    2082          42 :     subscriber_dir = NULL;
    2083          42 :     opt.config_file = NULL;
    2084          42 :     opt.pub_conninfo_str = NULL;
    2085          42 :     opt.socket_dir = NULL;
    2086          42 :     opt.sub_port = DEFAULT_SUB_PORT;
    2087          42 :     opt.sub_username = NULL;
    2088          42 :     opt.two_phase = false;
    2089          42 :     opt.database_names = (SimpleStringList)
    2090             :     {
    2091             :         0
    2092             :     };
    2093          42 :     opt.recovery_timeout = 0;
    2094          42 :     opt.all_dbs = false;
    2095             : 
    2096             :     /*
    2097             :      * Don't allow it to be run as root. It uses pg_ctl which does not allow
    2098             :      * it either.
    2099             :      */
    2100             : #ifndef WIN32
    2101          42 :     if (geteuid() == 0)
    2102             :     {
    2103           0 :         pg_log_error("cannot be executed by \"root\"");
    2104           0 :         pg_log_error_hint("You must run %s as the PostgreSQL superuser.",
    2105             :                           progname);
    2106           0 :         exit(1);
    2107             :     }
    2108             : #endif
    2109             : 
    2110          42 :     get_restricted_token();
    2111             : 
    2112         324 :     while ((c = getopt_long(argc, argv, "ad:D:np:P:R:s:t:TU:v",
    2113             :                             long_options, &option_index)) != -1)
    2114             :     {
    2115         288 :         switch (c)
    2116             :         {
    2117           6 :             case 'a':
    2118           6 :                 opt.all_dbs = true;
    2119           6 :                 break;
    2120          50 :             case 'd':
    2121          50 :                 if (!simple_string_list_member(&opt.database_names, optarg))
    2122             :                 {
    2123          48 :                     simple_string_list_append(&opt.database_names, optarg);
    2124          48 :                     num_dbs++;
    2125             :                 }
    2126             :                 else
    2127             :                 {
    2128           2 :                     pg_log_error("database \"%s\" specified more than once", optarg);
    2129           2 :                     exit(1);
    2130             :                 }
    2131          48 :                 break;
    2132          38 :             case 'D':
    2133          38 :                 subscriber_dir = pg_strdup(optarg);
    2134          38 :                 canonicalize_path(subscriber_dir);
    2135          38 :                 break;
    2136          18 :             case 'n':
    2137          18 :                 dry_run = true;
    2138          18 :                 break;
    2139          24 :             case 'p':
    2140          24 :                 opt.sub_port = pg_strdup(optarg);
    2141          24 :                 break;
    2142          36 :             case 'P':
    2143          36 :                 opt.pub_conninfo_str = pg_strdup(optarg);
    2144          36 :                 break;
    2145           2 :             case 'R':
    2146           2 :                 if (!simple_string_list_member(&opt.objecttypes_to_remove, optarg))
    2147           2 :                     simple_string_list_append(&opt.objecttypes_to_remove, optarg);
    2148             :                 else
    2149           0 :                     pg_fatal("object type \"%s\" is specified more than once for --remove", optarg);
    2150           2 :                 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             :                 {
    2178           2 :                     pg_log_error("publication \"%s\" specified more than once", optarg);
    2179           2 :                     exit(1);
    2180             :                 }
    2181          22 :                 break;
    2182           8 :             case 3:
    2183           8 :                 if (!simple_string_list_member(&opt.replslot_names, optarg))
    2184             :                 {
    2185           8 :                     simple_string_list_append(&opt.replslot_names, optarg);
    2186           8 :                     num_replslots++;
    2187             :                 }
    2188             :                 else
    2189             :                 {
    2190           0 :                     pg_log_error("replication slot \"%s\" specified more than once", optarg);
    2191           0 :                     exit(1);
    2192             :                 }
    2193           8 :                 break;
    2194          10 :             case 4:
    2195          10 :                 if (!simple_string_list_member(&opt.sub_names, optarg))
    2196             :                 {
    2197          10 :                     simple_string_list_append(&opt.sub_names, optarg);
    2198          10 :                     num_subs++;
    2199             :                 }
    2200             :                 else
    2201             :                 {
    2202           0 :                     pg_log_error("subscription \"%s\" specified more than once", optarg);
    2203           0 :                     exit(1);
    2204             :                 }
    2205          10 :                 break;
    2206           2 :             default:
    2207             :                 /* getopt_long already emitted a complaint */
    2208           2 :                 pg_log_error_hint("Try \"%s --help\" for more information.", progname);
    2209           2 :                 exit(1);
    2210             :         }
    2211             :     }
    2212             : 
    2213             :     /* Validate that --all is not used with incompatible options */
    2214          36 :     if (opt.all_dbs)
    2215             :     {
    2216           6 :         char       *bad_switch = NULL;
    2217             : 
    2218           6 :         if (num_dbs > 0)
    2219           2 :             bad_switch = "--database";
    2220           4 :         else if (num_pubs > 0)
    2221           2 :             bad_switch = "--publication";
    2222           2 :         else if (num_replslots > 0)
    2223           0 :             bad_switch = "--replication-slot";
    2224           2 :         else if (num_subs > 0)
    2225           0 :             bad_switch = "--subscription";
    2226             : 
    2227           6 :         if (bad_switch)
    2228             :         {
    2229           4 :             pg_log_error("%s cannot be used with --all", bad_switch);
    2230           4 :             pg_log_error_hint("Try \"%s --help\" for more information.", progname);
    2231           4 :             exit(1);
    2232             :         }
    2233             :     }
    2234             : 
    2235             :     /* Any non-option arguments? */
    2236          32 :     if (optind < argc)
    2237             :     {
    2238           0 :         pg_log_error("too many command-line arguments (first is \"%s\")",
    2239             :                      argv[optind]);
    2240           0 :         pg_log_error_hint("Try \"%s --help\" for more information.", progname);
    2241           0 :         exit(1);
    2242             :     }
    2243             : 
    2244             :     /* Required arguments */
    2245          32 :     if (subscriber_dir == NULL)
    2246             :     {
    2247           2 :         pg_log_error("no subscriber data directory specified");
    2248           2 :         pg_log_error_hint("Try \"%s --help\" for more information.", progname);
    2249           2 :         exit(1);
    2250             :     }
    2251             : 
    2252             :     /* If socket directory is not provided, use the current directory */
    2253          30 :     if (opt.socket_dir == NULL)
    2254             :     {
    2255             :         char        cwd[MAXPGPATH];
    2256             : 
    2257          10 :         if (!getcwd(cwd, MAXPGPATH))
    2258           0 :             pg_fatal("could not determine current directory");
    2259          10 :         opt.socket_dir = pg_strdup(cwd);
    2260          10 :         canonicalize_path(opt.socket_dir);
    2261             :     }
    2262             : 
    2263             :     /*
    2264             :      * Parse connection string. Build a base connection string that might be
    2265             :      * reused by multiple databases.
    2266             :      */
    2267          30 :     if (opt.pub_conninfo_str == NULL)
    2268             :     {
    2269             :         /*
    2270             :          * TODO use primary_conninfo (if available) from subscriber and
    2271             :          * extract publisher connection string. Assume that there are
    2272             :          * identical entries for physical and logical replication. If there is
    2273             :          * not, we would fail anyway.
    2274             :          */
    2275           2 :         pg_log_error("no publisher connection string specified");
    2276           2 :         pg_log_error_hint("Try \"%s --help\" for more information.", progname);
    2277           2 :         exit(1);
    2278             :     }
    2279          28 :     pg_log_info("validating publisher connection string");
    2280          28 :     pub_base_conninfo = get_base_conninfo(opt.pub_conninfo_str,
    2281             :                                           &dbname_conninfo);
    2282          28 :     if (pub_base_conninfo == NULL)
    2283           0 :         exit(1);
    2284             : 
    2285          28 :     pg_log_info("validating subscriber connection string");
    2286          28 :     sub_base_conninfo = get_sub_conninfo(&opt);
    2287             : 
    2288             :     /*
    2289             :      * Fetch all databases from the source (publisher) and treat them as if
    2290             :      * the user specified has multiple --database options, one for each source
    2291             :      * database.
    2292             :      */
    2293          28 :     if (opt.all_dbs)
    2294             :     {
    2295           2 :         bool        dbnamespecified = (dbname_conninfo != NULL);
    2296             : 
    2297           2 :         get_publisher_databases(&opt, dbnamespecified);
    2298             :     }
    2299             : 
    2300          28 :     if (opt.database_names.head == NULL)
    2301             :     {
    2302           4 :         pg_log_info("no database was specified");
    2303             : 
    2304             :         /*
    2305             :          * Try to obtain the dbname from the publisher conninfo. If dbname
    2306             :          * parameter is not available, error out.
    2307             :          */
    2308           4 :         if (dbname_conninfo)
    2309             :         {
    2310           2 :             simple_string_list_append(&opt.database_names, dbname_conninfo);
    2311           2 :             num_dbs++;
    2312             : 
    2313           2 :             pg_log_info("database name \"%s\" was extracted from the publisher connection string",
    2314             :                         dbname_conninfo);
    2315             :         }
    2316             :         else
    2317             :         {
    2318           2 :             pg_log_error("no database name specified");
    2319           2 :             pg_log_error_hint("Try \"%s --help\" for more information.",
    2320             :                               progname);
    2321           2 :             exit(1);
    2322             :         }
    2323             :     }
    2324             : 
    2325             :     /* Number of object names must match number of databases */
    2326          26 :     if (num_pubs > 0 && num_pubs != num_dbs)
    2327             :     {
    2328           2 :         pg_log_error("wrong number of publication names specified");
    2329           2 :         pg_log_error_detail("The number of specified publication names (%d) must match the number of specified database names (%d).",
    2330             :                             num_pubs, num_dbs);
    2331           2 :         exit(1);
    2332             :     }
    2333          24 :     if (num_subs > 0 && num_subs != num_dbs)
    2334             :     {
    2335           2 :         pg_log_error("wrong number of subscription names specified");
    2336           2 :         pg_log_error_detail("The number of specified subscription names (%d) must match the number of specified database names (%d).",
    2337             :                             num_subs, num_dbs);
    2338           2 :         exit(1);
    2339             :     }
    2340          22 :     if (num_replslots > 0 && num_replslots != num_dbs)
    2341             :     {
    2342           2 :         pg_log_error("wrong number of replication slot names specified");
    2343           2 :         pg_log_error_detail("The number of specified replication slot names (%d) must match the number of specified database names (%d).",
    2344             :                             num_replslots, num_dbs);
    2345           2 :         exit(1);
    2346             :     }
    2347             : 
    2348             :     /* Verify the object types specified for removal from the subscriber */
    2349          22 :     for (SimpleStringListCell *cell = opt.objecttypes_to_remove.head; cell; cell = cell->next)
    2350             :     {
    2351           2 :         if (pg_strcasecmp(cell->val, "publications") == 0)
    2352           2 :             dbinfos.objecttypes_to_remove |= OBJECTTYPE_PUBLICATIONS;
    2353             :         else
    2354             :         {
    2355           0 :             pg_log_error("invalid object type \"%s\" specified for --remove", cell->val);
    2356           0 :             pg_log_error_hint("The valid option is: \"publications\"");
    2357           0 :             exit(1);
    2358             :         }
    2359             :     }
    2360             : 
    2361             :     /* Get the absolute path of pg_ctl and pg_resetwal on the subscriber */
    2362          20 :     pg_ctl_path = get_exec_path(argv[0], "pg_ctl");
    2363          20 :     pg_resetwal_path = get_exec_path(argv[0], "pg_resetwal");
    2364             : 
    2365             :     /* Rudimentary check for a data directory */
    2366          20 :     check_data_directory(subscriber_dir);
    2367             : 
    2368          20 :     dbinfos.two_phase = opt.two_phase;
    2369             : 
    2370             :     /*
    2371             :      * Store database information for publisher and subscriber. It should be
    2372             :      * called before atexit() because its return is used in the
    2373             :      * cleanup_objects_atexit().
    2374             :      */
    2375          20 :     dbinfos.dbinfo = store_pub_sub_info(&opt, pub_base_conninfo, sub_base_conninfo);
    2376             : 
    2377             :     /* Register a function to clean up objects in case of failure */
    2378          20 :     atexit(cleanup_objects_atexit);
    2379             : 
    2380             :     /*
    2381             :      * Check if the subscriber data directory has the same system identifier
    2382             :      * than the publisher data directory.
    2383             :      */
    2384          20 :     pub_sysid = get_primary_sysid(dbinfos.dbinfo[0].pubconninfo);
    2385          20 :     sub_sysid = get_standby_sysid(subscriber_dir);
    2386          20 :     if (pub_sysid != sub_sysid)
    2387           2 :         pg_fatal("subscriber data directory is not a copy of the source database cluster");
    2388             : 
    2389             :     /* Subscriber PID file */
    2390          18 :     snprintf(pidfile, MAXPGPATH, "%s/postmaster.pid", subscriber_dir);
    2391             : 
    2392             :     /*
    2393             :      * The standby server must not be running. If the server is started under
    2394             :      * service manager and pg_createsubscriber stops it, the service manager
    2395             :      * might react to this action and start the server again. Therefore,
    2396             :      * refuse to proceed if the server is running to avoid possible failures.
    2397             :      */
    2398          18 :     if (stat(pidfile, &statbuf) == 0)
    2399             :     {
    2400           2 :         pg_log_error("standby server is running");
    2401           2 :         pg_log_error_hint("Stop the standby server and try again.");
    2402           2 :         exit(1);
    2403             :     }
    2404             : 
    2405             :     /*
    2406             :      * Start a short-lived standby server with temporary parameters (provided
    2407             :      * by command-line options). The goal is to avoid connections during the
    2408             :      * transformation steps.
    2409             :      */
    2410          16 :     pg_log_info("starting the standby server with command-line options");
    2411          16 :     start_standby_server(&opt, true, false);
    2412             : 
    2413             :     /* Check if the standby server is ready for logical replication */
    2414          16 :     check_subscriber(dbinfos.dbinfo);
    2415             : 
    2416             :     /* Check if the primary server is ready for logical replication */
    2417          12 :     check_publisher(dbinfos.dbinfo);
    2418             : 
    2419             :     /*
    2420             :      * Stop the target server. The recovery process requires that the server
    2421             :      * reaches a consistent state before targeting the recovery stop point.
    2422             :      * Make sure a consistent state is reached (stop the target server
    2423             :      * guarantees it) *before* creating the replication slots in
    2424             :      * setup_publisher().
    2425             :      */
    2426           8 :     pg_log_info("stopping the subscriber");
    2427           8 :     stop_standby_server(subscriber_dir);
    2428             : 
    2429             :     /* Create the required objects for each database on publisher */
    2430           8 :     consistent_lsn = setup_publisher(dbinfos.dbinfo);
    2431             : 
    2432             :     /* Write the required recovery parameters */
    2433           8 :     setup_recovery(dbinfos.dbinfo, subscriber_dir, consistent_lsn);
    2434             : 
    2435             :     /*
    2436             :      * Start subscriber so the recovery parameters will take effect. Wait
    2437             :      * until accepting connections. We don't want to start logical replication
    2438             :      * during setup.
    2439             :      */
    2440           8 :     pg_log_info("starting the subscriber");
    2441           8 :     start_standby_server(&opt, true, true);
    2442             : 
    2443             :     /* Waiting the subscriber to be promoted */
    2444           8 :     wait_for_end_recovery(dbinfos.dbinfo[0].subconninfo, &opt);
    2445             : 
    2446             :     /*
    2447             :      * Create the subscription for each database on subscriber. It does not
    2448             :      * enable it immediately because it needs to adjust the replication start
    2449             :      * point to the LSN reported by setup_publisher().  It also cleans up
    2450             :      * publications created by this tool and replication to the standby.
    2451             :      */
    2452           8 :     setup_subscriber(dbinfos.dbinfo, consistent_lsn);
    2453             : 
    2454             :     /* Remove primary_slot_name if it exists on primary */
    2455           8 :     drop_primary_replication_slot(dbinfos.dbinfo, primary_slot_name);
    2456             : 
    2457             :     /* Remove failover replication slots if they exist on subscriber */
    2458           8 :     drop_failover_replication_slots(dbinfos.dbinfo);
    2459             : 
    2460             :     /* Stop the subscriber */
    2461           8 :     pg_log_info("stopping the subscriber");
    2462           8 :     stop_standby_server(subscriber_dir);
    2463             : 
    2464             :     /* Change system identifier from subscriber */
    2465           8 :     modify_subscriber_sysid(&opt);
    2466             : 
    2467           8 :     success = true;
    2468             : 
    2469           8 :     pg_log_info("Done!");
    2470             : 
    2471           8 :     return 0;
    2472             : }

Generated by: LCOV version 1.14