Subject: Re: speed up a logical replica setup

From: Tomas Vondra <[email-masked]>
Date: Thu, 7 Mar 2024 18:14:14 +0100

Hi,

I decided to take a quick look on this patch today, to see how it works
and do some simple tests. I've only started to get familiar with it, so
I have only some comments / questions regarding usage, not on the code.
It's quite possible I didn't understand some finer points, or maybe it
was already discussed earlier in this very long thread, so please feel
free to push back or point me to the past discussion.

Also, some of this is rather opinionated, but considering I didn't see
this patch before, my opinions may easily be wrong ...

1) SGML docs

It seems the SGML docs are more about explaining how this works on the
inside, rather than how to use the tool. Maybe that's intentional, but
as someone who didn't work with pg_createsubscriber before I found it
confusing and not very helpful.

For example, the first half of the page is prerequisities+warning, and
sure those are useful details, but prerequisities are checked by the
tool (so I can't really miss this) and warnings go into a lot of details
about different places where things may go wrong. Sure, worth knowing
and including in the docs, but maybe not right at the beginning, before
I learn how to even run the tool?

Maybe that's just me, though. Also, I'm sure it's not the only part of
our docs like this. Perhaps it'd be good to reorganize the content a bit
to make the "how to use" stuff more prominent?

2) this is a bit vague

... pg_createsubscriber will check a few times if the connection has
been reestablished to stream the required WAL. After a few attempts, it
terminates with an error.

What does "a few times" mean, and how many is "a few attempts"? Seems
worth knowing when using this tool in environments where disconnections
can happen. Maybe this should be configurable?

3) publication info

For a while I was quite confused about which tables get replicated,
until I realized the publication is FOR ALL TABLES. But I learned that
from this thread, the docs say nothing about this. Surely that's an
important detail that should be mentioned?

4) Is FOR ALL TABLES a good idea?

I'm not sure FOR ALL TABLES is a good idea. Or said differently, I'm
sure it won't work for a number of use cases. I know large databases
it's common to create "work tables" (not necessarily temporary) as part
of a batch job, but there's no need to replicate those tables.

AFAIK that'd break this FOR ALL TABLES publication, because the tables
will qualify for replication, but won't be present on the subscriber. Or
did I miss something?

I do understand that FOR ALL TABLES is the simplest approach, and for v1
it may be an acceptable limitation, but maybe it'd be good to also
support restricting which tables should be replicated (e.g. blacklist or
whitelist based on table/schema name?).

BTW if I'm right and creating a table breaks the subscriber creation,
maybe it'd be good to explicitly mention that in the docs.

Note: I now realize this might fall under the warning about DDL, which
says this:

    Executing DDL commands on the source server while running
    pg_createsubscriber is not recommended. If the target server has
    already been converted to logical replica, the DDL commands must
    not be replicated so an error would occur.

But I find this confusing. Surely there are many DDL commands that have
absolutely no impact on logical replication (like creating an index or
view, various ALTER TABLE flavors, and so on). And running such DDL
certainly does not trigger error, right?

5) slot / publication / subscription name

I find it somewhat annoying it's not possible to specify names for
objects created by the tool - replication slots, publication and
subscriptions. If this is meant to be a replica running for a while,
after a while I'll have no idea what pg_createsubscriber_569853 or
pg_createsubscriber_459548_2348239 was meant for.

This is particularly annoying because renaming these objects later is
either not supported at all (e.g. for replication slots), or may be
quite difficult (e.g. publications).

I do realize there are challenges with custom names (say, if there are
multiple databases to replicate), but can't we support some simple
formatting with basic placeholders? So we could specify

--slot-name "myslot_%d_%p"

or something like that?

BTW what will happen if we convert multiple standbys? Can't they all get
the same slot name (they all have the same database OID, and I'm not
sure how much entropy the PID has)?

regards

-- 
Tomas Vondra
EnterpriseDB: http://www.enterprisedb.com
The Enterprise PostgreSQL Company

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Fri, 8 Mar 2024 09:44:19 +0000

Dear Tomas, Euler,

Thanks for starting to read the thread! Since I'm not an original author,
I want to reply partially.

> this patch before, my opinions may easily be wrong ...

I felt your comments were quit valuable.

> I learn how to even run the tool?

Hmm, right. I considered below improvements. Tomas and Euler, how do you think?

* Adds more descriptions in "Description" section.
* Moves prerequisities+warning to "Notes" section.
* Adds "Usage" section which describes from a single node.

> of a batch job, but there's no need to replicate those tables.

Indeed, the documentation does not describe that all tables in the database
would be included in the publication.

> whitelist based on table/schema name?).

May not directly related, but we considered that accepting options was a next-step [1].

>     not be replicated so an error would occur.

Yeah, they would not be replicated, but not lead ERROR.
So should we say like "Creating tables on the source server..."?

> --slot-name "myslot_%d_%p"

Not sure we can do in the first version, but looks nice. One concern is that I
cannot find applications which accepts escape strings like log_line_prefix.
(It may just because we do not have use-case.) Do you know examples?

> sure how much entropy the PID has)?

I tested and the second try did not work. The primal reason was the name of publication
- pg_createsubscriber_%u (oid).
FYI - previously we can reuse same publications, but based on my comment [2] the
feature was removed. It might be too optimistic.

[1]: https://www.postgresql.org/message-id/TY3PR01MB9889CCBD4D9DAF8BD2F18541F56F2%40TY3PR01MB9889.jpnprd01.prod.outlook.com
[2]: https://www.postgresql.org/message-id/TYCPR01MB12077756323B79042F29DDAEDF54C2%40TYCPR01MB12077.jpnprd01.prod.outlook.com

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/

---

From: Tomas Vondra <[email-masked]>
Date: Fri, 8 Mar 2024 20:26:44 +0100

> So should we say like "Creating tables on the source server..."?

Perhaps. Clarifying the docs would help, but it depends on the wording.
For example, I doubt this should talk about "creating tables" because
there are other DDL that (probably) could cause issues (like adding a
column to the table, or something like that).

> (It may just because we do not have use-case.) Do you know examples?

I can't think of a tool already doing that, but I think that's simply
because it was not needed. Why should we be concerned about this?

> feature was removed. It might be too optimistic.

OK. I could be convinced the other limitations are reasonable for v1 and
can be improved later, but this seems like something that needs fixing.

regards

---

From: vignesh C <[email-masked]>
Date: Mon, 11 Mar 2024 09:42:06 +0530

> can be improved later, but this seems like something that needs fixing.

+1 to handle this.
Currently, a) Publication name = pg_createsubscriber_%u, where %u is
database oid, b) Replication slot name =  pg_createsubscriber_%u_%d,
Where %u is database oid and %d is the pid and c) Subscription name =
pg_createsubscriber_%u_%d, Where %u is database oid and %d is the pid
How about we have a non mandatory option like
--prefix_object_name=mysetup1, which will create a) Publication name =
mysetup1_pg_createsubscriber_%u, and b) Replication slot name =
mysetup1_pg_createsubscriber_%u (here pid is also removed) c)
Subscription name = mysetup1_pg_createsubscriber_%u (here pid is also
removed).

In the default case where the user does not specify
--prefix_object_name the object names will be created without any
prefix names.

Regards,
Vignesh

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Mon, 11 Mar 2024 05:03:07 +0000

Dear Vignesh,

Thanks for updating the patch, but cfbot still got angry [1].
Note that two containers (autoconf and meson) failed at different place,
so I think it is intentional one. It seems that there may be a bug related with 32-bit build.

We should see and fix as soon as possible.

[1]: http://cfbot.cputube.org/highlights/all.html#4637

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/

---

From: Amit Kapila <[email-masked]>
Date: Tue, 12 Mar 2024 15:43:02 +0530

> will qualify for replication, but won't be present on the subscriber. Or

As the subscriber is created from standby, all the tables should be
present at least initially during and after creating the subscriber.
Users are later free to modify the publications/subscriptions.

> whitelist based on table/schema name?).

This would be useful, but OTOH could also be enhanced in a later
version unless we think it is a must for the first version.

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Tue, 12 Mar 2024 16:15:02 +0530

> --prefix_object_name the object names will be created without any

Tomas's idea is better in terms of useability. So, we should instead
have three switches --slot-name, --publication-name, and
--subscriber-name with some provision of appending dbid's and some
unique identifier for standby. The unique identifier can help in
creating multiple subscribers from different standbys.

-- 
With Regards,
Amit Kapila.

---

From: "Euler Taveira" <[email-masked]>
Date: Wed, 13 Mar 2024 03:10:21 -0300

> Hmm, right. I considered below improvements. Tomas and Euler, how do you think?

I'm posting a new patchset v28.

I changed the way that the check function works. From the usability
perspective, it is better to test all conditions and reports all errors (if
any) at once. It avoids multiple executions in dry run mode just to figure out
all of the issues in the initial phase. I also included tests for it using
Shlok's idea [1] although I didn't use v27-0004.

Shlok [1] reported that it was failing on Windows since the socket-directory
option was added. I added a fix for it.

Tomas pointed out the documentation [2] does not provide a good high level
explanation about pg_createsubscriber. I expanded the Description section and
moved the prerequisites to Nodes section. The prerequisites were grouped into
target and source conditions on their own paragraph instead of using a list. It
seems more in line with the style of some applications.

As I said in a previous email [3], I removed the retain option.

[1] https://www.postgresql.org/message-id/[email-masked]
[2] https://www.postgresql.org/message-id/6423dfeb-a729-45d3-b71e-7bf1b3adb0c9%40enterprisedb.com
[3] https://www.postgresql.org/message-id/e390e35e-508e-4eb8-92e4-e6b066407a41%40app.fastmail.com

--
Euler Taveira
EDB   https://www.enterprisedb.com/

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Wed, 13 Mar 2024 07:47:52 +0000

Dear Euler,

Thanks for updating the patch! I have not reviewed yet, but cfbot did not
accept your patch [1]. It seems that you missed updating the windows part
in 0003 patch:

```
#if !defined(WIN32)
-       sub_base_conninfo = psprintf("host=%s port=%u user=%s fallback_application_name=%s",
+       sub_base_conninfo = psprintf("host=%s port=%s user=%s fallback_application_name=%s",
                                                                 opt.socket_dir, opt.sub_port, opt.sub_username, progname);
 #else                                                  /* WIN32 */
        sub_base_conninfo = psprintf("port=%u user=%s fallback_application_name=%s"
```

Since the change is quite trivial, I felt no need to update the versioning.
PSA the new set to keep the cfbot quiet. These could pass tests on my CI.

[1]:  https://cirrus-ci.com/build/5253379782344704

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/global/

---

From: Shlok Kyal <[email-masked]>
Date: Wed, 13 Mar 2024 18:39:21 +0530

Hi,

Currently when the pg_createsubscriber error out and exit in between
execution, the standby server started by the pg_createsubscriber may
still be running. I think this standby server should be stopped at
exit of pg_createsubscriber(when it errors out), as this server is
started by pg_createsubscriber and it may be running with
configurations that may not be desired by the user (such as " -c
listen_addresses='' -c unix_socket_permissions=0700"). Thoughts?

Added a top-up patch v28-0005 to fix this issue.
I am not changing the version as v28-0001 to v28-0004 is the same as above.

Thanks and regards,
Shlok Kyal

---

From: "Euler Taveira" <[email-masked]>
Date: Fri, 15 Mar 2024 00:51:48 -0300

> I am not changing the version as v28-0001 to v28-0004 is the same as above.

Thanks for your review!

I'm posting a new patch (v29) that merges the previous patches (v28-0002 and
v28-0003). I applied the fix provided by Hayato [1]. It was an oversight during
a rebase. I also included the patch proposed by Shlok [2] that stops the target
server on error if it is running.

Tomas suggested in [3] that maybe the PID should be replaced with something
else that has more entropy. Instead of PID, it uses a random number for
replication slot and subscription. There is also a concern about converting
multiple standbys that will have the same publication name. It added the same
random number to the publication name so it doesn't fail because the
publication already exists. Documentation was changed based on Tomas feedback.

The user name was always included in the subscriber connection string. Let's
have the libpq to choose it. While on it, a new routine (get_sub_conninfo)
contains the code to build the subscriber connection string.

As I said in [4], there wasn't a way to inform a different configuration file.
If your cluster has a postgresql.conf outside PGDATA, when pg_createsubscriber
starts the server it will fail. The new --config-file option let you inform the
postgresql.conf location and the server is started just fine.

I also did some changes in the start_standby_server routine. I replaced the
strcat and snprintf with appendPQExpBuffer that has been used to store the
pg_ctl command.

[1] https://www.postgresql.org/message-id/TYCPR01MB12077FD21BB186C5A685C0BF3F52A2%40TYCPR01MB12077.jpnprd01.prod.outlook.com
[2] https://www.postgresql.org/message-id/CANhcyEW6-dH28gLbFc5XpDTJ6JPizU%2Bt5g-aKUWJBf5W_Zriqw%40mail.gmail.com
[3] https://www.postgresql.org/message-id/6423dfeb-a729-45d3-b71e-7bf1b3adb0c9%40enterprisedb.com
[4] https://www.postgresql.org/message-id/d898faad-f6d7-4b0d-b816-b9dcdf490685%40app.fastmail.com

---

From: Amit Kapila <[email-masked]>
Date: Fri, 15 Mar 2024 12:04:01 +0530

On Fri, Mar 15, 2024 at 9:23 AM Euler Taveira <[email-masked]> wrote:

Did you consider adding options for publication/subscription/slot
names as mentioned in my previous email? As discussed in a few emails
above, it would be quite confusing for users to identify the logical
replication objects once the standby is converted to subscriber.

*
+static void
+cleanup_objects_atexit(void)
{
...
conn = connect_database(dbinfo[i].pubconninfo, false);
+ if (conn != NULL)
+ {
+ if (dbinfo[i].made_publication)
+ drop_publication(conn, &dbinfo[i]);
+ if (dbinfo[i].made_replslot)
+ drop_replication_slot(conn, &dbinfo[i], dbinfo[i].subname);
+ disconnect_database(conn, false);
+ }
+ else
+ {
+ /*
+ * If a connection could not be established, inform the user
+ * that some objects were left on primary and should be
+ * removed before trying again.
+ */
+ if (dbinfo[i].made_publication)
+ {
+ pg_log_warning("There might be a publication \"%s\" in database
\"%s\" on primary",
+    dbinfo[i].pubname, dbinfo[i].dbname);
+ pg_log_warning_hint("Consider dropping this publication before
trying again.");
+ }

It seems we care only for publications created on the primary. Isn't
it possible that some of the publications have been replicated to
standby by that time, for example, in case failure happens after
creating a few publications? IIUC, we don't care for standby cleanup
after failure because it can't be used for streaming replication
anymore. So, the only choice the user has is to recreate the standby
and start the pg_createsubscriber again. This sounds questionable to
me as to whether users would like this behavior. Does anyone else have
an opinion on this point?

I see the below note in the patch:
+    If <application>pg_createsubscriber</application> fails while processing,
+    then the data directory is likely not in a state that can be recovered. It
+    is true if the target server was promoted. In such a case, creating a new
+    standby server is recommended.

By reading this it is not completely clear whether the standby is not
recoverable in case of any error or only an error after the target
server is promoted. If others agree with this behavior then we should
write the detailed reason for this somewhere in the comments as well
unless it is already explained.

-- 
With Regards,
Amit Kapila.

---

From: vignesh C <[email-masked]>
Date: Sat, 16 Mar 2024 19:01:02 +0530

> We should see and fix as soon as possible.

I was able to reproduce this random failure and found the following reason:
The Minimum recovery ending location 0/5000000 was more than the
recovery_target_lsn specified is "0/4001198". In few random cases the
standby applies a few more WAL records after the replication slot is
created; this leads to minimum recovery ending location being greater
than the recovery_target_lsn because of which the server will fail
with:
FATAL:  requested recovery stop point is before consistent recovery point

I have fixed it by pausing the replay in the standby server before the
replication slots get created.
The attached v29-0002-Keep-standby-server-s-minimum-recovery-point-les.patch
patch has the changes for the same.
Thoughts?

Regards,
Vignesh
From e98e1ba0661e3d658de8d46ff9082f6cd4040b41 Mon Sep 17 00:00:00 2001
From: Vignesh C <[email-masked]>
Date: Sat, 16 Mar 2024 18:47:35 +0530
Subject: [PATCH v29 2/2] Keep standby server's minimum recovery point less
 than the consistent_lsn.

Standby server should not get ahead of the replication slot's lsn. If
this happens we will not be able to promote the standby server as it
will not be able to reach the consistent point because the standby
server's Minimum recovery ending location will be greater than the
consistent_lsn. Fixed this by pausing the replay before the replication
slots are created.
---
 src/bin/pg_basebackup/pg_createsubscriber.c | 93 ++++++++++++++++++++-
 1 file changed, 92 insertions(+), 1 deletion(-)

diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index c565f8524a..18e8757403 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -95,6 +95,8 @@ static void drop_publication(PGconn *conn, struct LogicalRepInfo *dbinfo);
 static void create_subscription(PGconn *conn, struct LogicalRepInfo *dbinfo);
 static void set_replication_progress(PGconn *conn, struct LogicalRepInfo *dbinfo,
 									 const char *lsn);
+static void pause_replay_stantby_server(struct LogicalRepInfo *dbinfo,
+										struct CreateSubscriberOptions *opt);
 static void enable_subscription(PGconn *conn, struct LogicalRepInfo *dbinfo);
 
 #define	USEC_PER_SEC	1000000
@@ -917,7 +919,9 @@ check_subscriber(struct LogicalRepInfo *dbinfo)
 	appendPQExpBuffer(str,
 					  "SELECT pg_catalog.pg_has_role(current_user, %u, 'MEMBER'), "
 					  "pg_catalog.has_database_privilege(current_user, '%s', 'CREATE'), "
-					  "pg_catalog.has_function_privilege(current_user, 'pg_catalog.pg_replication_origin_advance(text, pg_lsn)', 'EXECUTE')",
+					  "pg_catalog.has_function_privilege(current_user, 'pg_catalog.pg_replication_origin_advance(text, pg_lsn)', 'EXECUTE'), "
+					  "pg_catalog.has_function_privilege(current_user, 'pg_catalog.pg_wal_replay_pause()', 'EXECUTE'), "
+					  "pg_catalog.has_function_privilege(current_user, 'pg_catalog.pg_get_wal_replay_pause_state()', 'EXECUTE')",
 					  ROLE_PG_CREATE_SUBSCRIPTION, dbinfo[0].dbname);
 
 	pg_log_debug("command is: %s", str->data);
@@ -949,6 +953,18 @@ check_subscriber(struct LogicalRepInfo *dbinfo)
 					 "pg_catalog.pg_replication_origin_advance(text, pg_lsn)");
 		failed = true;
 	}
+	if (strcmp(PQgetvalue(res, 0, 3), "t") != 0)
+	{
+		pg_log_error("permission denied for function \"%s\"",
+					 "pg_catalog.pg_wal_replay_pause()");
+		failed = true;
+	}
+	if (strcmp(PQgetvalue(res, 0, 4), "t") != 0)
+	{
+		pg_log_error("permission denied for function \"%s\"",
+					 "pg_catalog.pg_get_wal_replay_pause_state()");
+		failed = true;
+	}
 
 	destroyPQExpBuffer(str);
 	PQclear(res);
@@ -1026,6 +1042,72 @@ check_subscriber(struct LogicalRepInfo *dbinfo)
 		exit(1);
 }
 
+/*
+ * Pause replaying at the standby server.
+ */
+static void
+pause_replay_stantby_server(struct LogicalRepInfo *dbinfo,
+							struct CreateSubscriberOptions *opt)
+{
+	PGconn	   *conn;
+	PGresult   *res;
+
+	/* Connect to subscriber. */
+	conn = connect_database(dbinfo[0].subconninfo, true);
+
+	pg_log_info("Pausing the replay in standby server");
+	pg_log_debug("command is: SELECT pg_catalog.pg_wal_replay_pause()");
+
+	if (!dry_run)
+	{
+		int			timer = 0;
+
+		res = PQexec(conn, "SELECT pg_catalog.pg_wal_replay_pause()");
+		if (PQresultStatus(res) != PGRES_TUPLES_OK)
+		{
+			pg_log_error("could not pause replay in standby server: %s",
+						 PQresultErrorMessage(res));
+			disconnect_database(conn, true);
+		}
+		PQclear(res);
+
+		/* Wait till replay has paused */
+		for(;;)
+		{
+			pg_log_debug("command is: SELECT pg_catalog.pg_get_wal_replay_pause_state()");
+			res = PQexec(conn, "SELECT pg_catalog.pg_get_wal_replay_pause_state()");
+			if (PQresultStatus(res) != PGRES_TUPLES_OK)
+			{
+				pg_log_error("could not get pause replay state in standby server: %s",
+							PQresultErrorMessage(res));
+				disconnect_database(conn, true);
+			}
+
+			if (strcmp(PQgetvalue(res, 0, 0), "paused") == 0)
+			{
+				PQclear(res);
+				break;
+			}
+
+			PQclear(res);
+
+			/* Bail out after recovery_timeout seconds if this option is set */
+			if (opt->recovery_timeout > 0 && timer >= opt->recovery_timeout)
+			{
+				pg_log_error("timed out waiting to pause replay");
+				disconnect_database(conn, true);
+			}
+
+			/* Keep waiting */
+			pg_usleep(WAIT_INTERVAL * USEC_PER_SEC);
+
+			timer += WAIT_INTERVAL;
+		}
+	}
+
+	disconnect_database(conn, false);
+}
+
 /*
  * Create the subscriptions, adjust the initial location for logical
  * replication and enable the subscriptions. That's the last step for logical
@@ -1948,6 +2030,15 @@ main(int argc, char **argv)
 	 */
 	check_publisher(dbinfo);
 
+	/*
+	 * Standby server should not get ahead of the replication slot's lsn. If
+	 * this happens we will not be able to promote the standby server as it
+	 * will not be able to reach the consistent point because the standby
+	 * server's Minimum recovery ending location will be greater than the
+	 * consistent_lsn.
+	 */
+	pause_replay_stantby_server(dbinfo, &opt);
+
 	/*
 	 * Create the required objects for each database on publisher. This step
 	 * is here mainly because if we stop the standby we cannot verify if the
-- 
2.34.1

From a0aa20ff421a12e4f94358d4144c1fc3106a3009 Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email-masked]>
Date: Mon, 5 Jun 2023 14:39:40 -0400
Subject: [PATCH v29 1/2] pg_createsubscriber: creates a new logical replica
 from a standby server

It must be run on the target server and should be able to connect to the
source server (publisher) and the target server (subscriber). All tables
in the specified database(s) are included in the logical replication
setup. A pair of publication and subscription objects are created for
each database.

The main advantage of pg_createsubscriber over the common logical
replication setup is the initial data copy. It also reduces the catchup
phase.

Some prerequisites must be met to successfully run it. It is basically
the logical replication requirements. It starts creating a publication
using FOR ALL TABLES and a replication slot for each specified database.
Write recovery parameters into the target data directory and start the
target server. It specifies the LSN of the last replication slot
(replication start point) up to which the recovery will proceed. Wait
until the target server is promoted. Create one subscription per
specified database (using publication and replication slot created in a
previous step) on the target server. Set the replication progress to the
replication start point for each subscription. Enable the subscription
for each specified database on the target server. And finally, change
the system identifier on the target server.
---
 doc/src/sgml/ref/allfiles.sgml                |    1 +
 doc/src/sgml/ref/pg_createsubscriber.sgml     |  477 ++++
 doc/src/sgml/reference.sgml                   |    1 +
 src/bin/pg_basebackup/.gitignore              |    1 +
 src/bin/pg_basebackup/Makefile                |   11 +-
 src/bin/pg_basebackup/meson.build             |   19 +
 src/bin/pg_basebackup/pg_createsubscriber.c   | 2005 +++++++++++++++++
 .../t/040_pg_createsubscriber.pl              |  265 +++
 8 files changed, 2777 insertions(+), 3 deletions(-)
 create mode 100644 doc/src/sgml/ref/pg_createsubscriber.sgml
 create mode 100644 src/bin/pg_basebackup/pg_createsubscriber.c
 create mode 100644 src/bin/pg_basebackup/t/040_pg_createsubscriber.pl

diff --git a/doc/src/sgml/ref/allfiles.sgml b/doc/src/sgml/ref/allfiles.sgml
index 4a42999b18..f5be638867 100644
--- a/doc/src/sgml/ref/allfiles.sgml
+++ b/doc/src/sgml/ref/allfiles.sgml
@@ -205,6 +205,7 @@ Complete list of usable sgml source files in this directory.
 <!ENTITY pgCombinebackup    SYSTEM "pg_combinebackup.sgml">
 <!ENTITY pgConfig           SYSTEM "pg_config-ref.sgml">
 <!ENTITY pgControldata      SYSTEM "pg_controldata.sgml">
+<!ENTITY pgCreateSubscriber SYSTEM "pg_createsubscriber.sgml">
 <!ENTITY pgCtl              SYSTEM "pg_ctl-ref.sgml">
 <!ENTITY pgDump             SYSTEM "pg_dump.sgml">
 <!ENTITY pgDumpall          SYSTEM "pg_dumpall.sgml">
diff --git a/doc/src/sgml/ref/pg_createsubscriber.sgml b/doc/src/sgml/ref/pg_createsubscriber.sgml
new file mode 100644
index 0000000000..d038572eb0
--- /dev/null
+++ b/doc/src/sgml/ref/pg_createsubscriber.sgml
@@ -0,0 +1,477 @@
+<!--
+doc/src/sgml/ref/pg_createsubscriber.sgml
+PostgreSQL documentation
+-->
+
+<refentry id="app-pgcreatesubscriber">
+ <indexterm zone="app-pgcreatesubscriber">
+  <primary>pg_createsubscriber</primary>
+ </indexterm>
+
+ <refmeta>
+  <refentrytitle><application>pg_createsubscriber</application></refentrytitle>
+  <manvolnum>1</manvolnum>
+  <refmiscinfo>Application</refmiscinfo>
+ </refmeta>
+
+ <refnamediv>
+  <refname>pg_createsubscriber</refname>
+  <refpurpose>convert a physical replica into a new logical replica</refpurpose>
+ </refnamediv>
+
+ <refsynopsisdiv>
+  <cmdsynopsis>
+   <command>pg_createsubscriber</command>
+   <arg rep="repeat"><replaceable>option</replaceable></arg>
+   <group choice="plain">
+    <group choice="req">
+     <arg choice="plain"><option>-d</option></arg>
+     <arg choice="plain"><option>--database</option></arg>
+    </group>
+    <replaceable>dbname</replaceable>
+    <group choice="req">
+     <arg choice="plain"><option>-D</option> </arg>
+     <arg choice="plain"><option>--pgdata</option></arg>
+    </group>
+    <replaceable>datadir</replaceable>
+    <group choice="req">
+     <arg choice="plain"><option>-P</option></arg>
+     <arg choice="plain"><option>--publisher-server</option></arg>
+    </group>
+    <replaceable>connstr</replaceable>
+   </group>
+  </cmdsynopsis>
+ </refsynopsisdiv>
+
+ <refsect1>
+  <title>Description</title>
+  <para>
+    <application>pg_createsubscriber</application> creates a new logical
+    replica from a physical standby server. All tables in the specified
+    database are included in the logical replication setup. A pair of
+    publication and subscription objects are created for each database. It
+    must be run at the target server.
+  </para>
+
+  <para>
+    After a successful run, the state of the target server is analagous to a
+    fresh logical replication setup. The main difference between the logical
+    replication setup and <application>pg_createsubscriber</application>
+    is the initial data copy. Only the synchronization phase is done, which
+    ensures each table is brought up to a synchronized state.
+  </para>
+
+  <para>
+   The <application>pg_createsubscriber</application> targets large database
+   systems because most of the execution time is spent making the initial data
+   copy. For smaller databases,
+   <link linkend="logical-replication">initial data synchronization</link> is
+   recommended.
+  </para>
+
+ </refsect1>
+
+ <refsect1>
+  <title>Options</title>
+
+   <para>
+    <application>pg_createsubscriber</application> accepts the following
+    command-line arguments:
+
+    <variablelist>
+     <varlistentry>
+      <term><option>-d <replaceable class="parameter">dbname</replaceable></option></term>
+      <term><option>--database=<replaceable class="parameter">dbname</replaceable></option></term>
+      <listitem>
+       <para>
+        The database name to create the subscription. Multiple databases can be
+        selected by writing multiple <option>-d</option> switches.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-D <replaceable class="parameter">directory</replaceable></option></term>
+      <term><option>--pgdata=<replaceable class="parameter">directory</replaceable></option></term>
+      <listitem>
+       <para>
+        The target directory that contains a cluster directory from a physical
+        replica.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-n</option></term>
+      <term><option>--dry-run</option></term>
+      <listitem>
+       <para>
+        Do everything except actually modifying the target directory.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-p <replaceable class="parameter">port</replaceable></option></term>
+      <term><option>--subscriber-port=<replaceable class="parameter">port</replaceable></option></term>
+      <listitem>
+       <para>
+        The port number on which the target server is listening for connections.
+        Defaults to running the target server on port 50432 to avoid unintended
+        client connections.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-P <replaceable class="parameter">connstr</replaceable></option></term>
+      <term><option>--publisher-server=<replaceable class="parameter">connstr</replaceable></option></term>
+      <listitem>
+       <para>
+        The connection string to the publisher. For details see <xref linkend="libpq-connstring"/>.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-s <replaceable class="parameter">dir</replaceable></option></term>
+      <term><option>--socket-directory=<replaceable class="parameter">dir</replaceable></option></term>
+      <listitem>
+       <para>
+        The directory to use for postmaster sockets on target server. The
+        default is current directory.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+       <term><option>-t <replaceable class="parameter">seconds</replaceable></option></term>
+       <term><option>--recovery-timeout=<replaceable class="parameter">seconds</replaceable></option></term>
+       <listitem>
+       <para>
+        The maximum number of seconds to wait for recovery to end. Setting to 0
+        disables. The default is 0.
+       </para>
+       </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-U <replaceable class="parameter">username</replaceable></option></term>
+      <term><option>--subscriber-username=<replaceable class="parameter">username</replaceable></option></term>
+      <listitem>
+       <para>
+        The username to connect as on target server. Defaults to the current
+        operating system user name.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>-v</option></term>
+      <term><option>--verbose</option></term>
+      <listitem>
+       <para>
+        Enables verbose mode. This will cause
+        <application>pg_createsubscriber</application> to output progress messages
+        and detailed information about each step to standard error.
+        Repeating the option causes additional debug-level messages to appear on
+        standard error.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+      <term><option>--config-file=<replaceable class="parameter">filename</replaceable></option></term>
+      <listitem>
+       <para>
+        Use the specified main server configuration file for the target
+        data directory. The <application>pg_createsubscriber</application>
+        uses internally the <application>pg_ctl</application> command to start
+        and stop the target server. It allows you to specify the actual
+        <filename>postgresql.conf</filename> configuration file if it is stored
+        outside the data directory.
+       </para>
+      </listitem>
+     </varlistentry>
+
+     <varlistentry>
+       <term><option>-?</option></term>
+       <term><option>--help</option></term>
+       <listitem>
+       <para>
+       Show help about <application>pg_createsubscriber</application> command
+       line arguments, and exit.
+       </para>
+       </listitem>
+     </varlistentry>
+
+     <varlistentry>
+       <term><option>-V</option></term>
+       <term><option>--version</option></term>
+       <listitem>
+       <para>
+       Print the <application>pg_createsubscriber</application> version and exit.
+       </para>
+       </listitem>
+     </varlistentry>
+
+    </variablelist>
+   </para>
+
+ </refsect1>
+
+ <refsect1>
+  <title>Notes</title>
+
+  <para>
+   There are some prerequisites for
+   <application>pg_createsubscriber</application> to convert the target server
+   into a logical replica. If these are not met an error will be reported. The
+   source and target servers must have the same major version as the
+   <application>pg_createsubscriber</application>. The given target data
+   directory must have the same system identifier than the source data
+   directory. If a standby server is running on the target data directory or it
+   is a base backup from the source data directory, system identifiers are the
+   same. The given database user for the target data directory must have
+   privileges for creating
+   <link linkend="sql-createsubscription">subscriptions</link> and using
+   <link linkend="pg-replication-origin-advance"><function>pg_replication_origin_advance()</function></link>.
+  </para>
+
+  <para>
+   The target server must be used as a physical standby. The target server
+   must have
+   <link linkend="guc-max-replication-slots"><varname>max_replication_slots</varname></link>
+   and
+   <link linkend="guc-max-logical-replication-workers"><varname>max_logical_replication_workers</varname></link>
+   configured to a value greater than or equal to the number of specified
+   databases. The target server must have
+   <link linkend="guc-max-worker-processes"><varname>max_worker_processes</varname></link>
+   configured to a value greater than the number of specified databases. The
+   target server must accept local connections.
+  </para>
+
+  <para>
+   The source server must accept connections from the target server. The
+   source server must not be in recovery. Publications cannot be created in a
+   read-only cluster. The source server must have
+   <link linkend="guc-wal-level"><varname>wal_level</varname></link> as
+   <literal>logical</literal>.
+   The source server must have
+   <link linkend="guc-max-replication-slots"><varname>max_replication_slots</varname></link>
+   configured to a value greater than or equal to the number of specified
+   databases plus existing replication slots. The source server must have
+   <link linkend="guc-max-wal-senders"><varname>max_wal_senders</varname></link>
+   configured to a value greater than or equal to the number of specified
+   databases and existing <literal>walsender</literal> processes.
+  </para>
+
+  <warning>
+   <title>Warning</title>
+   <para>
+    If <application>pg_createsubscriber</application> fails while processing,
+    then the data directory is likely not in a state that can be recovered. It
+    is true if the target server was promoted. In such a case, creating a new
+    standby server is recommended.
+   </para>
+
+   <para>
+    <application>pg_createsubscriber</application> usually starts the target
+    server with different connection settings during the transformation steps.
+    Hence, regular connections to the target server should fail.
+   </para>
+
+   <para>
+    During the recovery process, if the target server disconnects from the
+    source server, <application>pg_createsubscriber</application> will check
+    a few times if the connection has been reestablished to stream the required
+    WAL. After a few attempts, it terminates with an error.
+   </para>
+
+   <para>
+    Since DDL commands are not replicated by logical replication, avoid
+    executing DDL commands that change the database schema while running
+    <application>pg_createsubscriber</application>. If the target server has
+    already been converted to logical replica, the DDL commands must not be
+    replicated which might cause an error.
+   </para>
+
+   <para>
+    If <application>pg_createsubscriber</application> fails while processing,
+    objects (publications, replication slots) created on the source server
+    are removed. The removal might fail if the target server cannot connect to
+    the source server. In such a case, a warning message will inform the
+    objects left. If the target server is running, it will be stopped.
+   </para>
+
+   <para>
+    If the replication is using a
+    <link linkend="guc-primary-slot-name"><varname>primary_slot_name</varname></link>,
+    it will be removed from the source server after the logical replication setup.
+   </para>
+
+   <para>
+    If the target server is a synchronous replica, transaction commits on the
+    primary might wait for replication while running
+    <application>pg_createsubscriber</application>.
+   </para>
+
+   <para>
+    <application>pg_createsubscriber</application> changes the system identifier
+    using <application>pg_resetwal</application>. It would avoid situations in
+    which WAL files from the source server might be used by the target server.
+    If the target server has a standby, replication will break and a fresh
+    standby should be created.
+   </para>
+  </warning>
+
+ </refsect1>
+
+ <refsect1>
+  <title>How It Works</title>
+
+  <para>
+    The basic idea is to have a replication start point from the source server
+    and set up a logical replication to start from this point:
+  </para>
+
+  <procedure>
+   <step>
+    <para>
+     Start the target server with the specified command-line options. If the
+     target server is already running, stop it because some parameters can only
+     be set at server start.
+    </para>
+   </step>
+
+   <step>
+    <para>
+     Check if the target server can be converted. There are also a few checks
+     on the source server. If any of the prerequisites are not met,
+     <application>pg_createsubscriber</application> will terminate with an
+     error.
+    </para>
+   </step>
+
+   <step>
+    <para>
+     Create a publication and replication slot for each specified database on
+     the source server. Each publication is created using
+     <link linkend="sql-createpublication-params-for-all-tables"><literal>FOR ALL TABLES</literal></link>.
+     It has the following name pattern:
+     <quote><literal>pg_createsubscriber_%u_%x</literal></quote> (parameter:
+     database <parameter>oid</parameter>, random <parameter>int</parameter>). Each
+     replication slot has the following name pattern:
+     <quote><literal>pg_createsubscriber_%u_%x</literal></quote> (parameters:
+     database <parameter>oid</parameter>, random <parameter>int</parameter>).
+     These replication slots will be used by the subscriptions in a future step.
+     The last replication slot LSN is used as a stopping point in the
+     <xref linkend="guc-recovery-target-lsn"/> parameter and by the
+     subscriptions as a replication start point. It guarantees that no
+     transaction will be lost.
+    </para>
+   </step>
+
+   <step>
+    <para>
+     Write recovery parameters into the target data directory and restart the
+     target server. It specifies a LSN (<xref linkend="guc-recovery-target-lsn"/>)
+     of write-ahead log location up to which recovery will proceed. It also
+     specifies <literal>promote</literal> as the action that the server should
+     take once the recovery target is reached. Additional
+     <link linkend="runtime-config-wal-recovery-target">recovery parameters</link>
+     are also added so it avoids unexpected behavior during the recovery
+     process such as end of the recovery as soon as a consistent state is
+     reached (WAL should be applied until the replication start location) and
+     multiple recovery targets that can cause a failure. This step finishes
+     once the server ends standby mode and is accepting read-write transactions.
+     If <option>--recovery-timeout</option> option is set,
+     <application>pg_createsubscriber</application> terminates if recovery does
+     not end until the given number of seconds.
+    </para>
+   </step>
+
+   <step>
+    <para>
+     Create a subscription for each specified database on the target server.
+     The subscription has the following name pattern:
+     <quote><literal>pg_createsubscriber_%u_%x</literal></quote> (parameters:
+     database <parameter>oid</parameter>, random <parameter>int</parameter>).
+     The replication slot name is identical to the subscription name. It does
+     not copy existing data from the source server. It does not create a
+     replication slot. Instead, it uses the replication slot that was created
+     in a previous step. The subscription is created but it is not enabled yet.
+     The reason is the replication progress must be set to the replication
+     start point before starting the replication.
+    </para>
+   </step>
+
+   <step>
+    <para>
+     Drop publications on the target server that were replicated because they
+     were created before the replication start location. It has no use on the
+     subscriber.
+    </para>
+   </step>
+
+   <step>
+    <para>
+     Set the replication progress to the replication start point for each
+     subscription. When the target server starts the recovery process, it
+     catches up to the replication start point. This is the exact LSN to be used
+     as a initial replication location for each subscription. The replication
+     origin name is obtained since the subscription was created. The replication
+     origin name and the replication start point are used in
+     <link linkend="pg-replication-origin-advance"><function>pg_replication_origin_advance()</function></link>
+     to set up the initial replication location.
+    </para>
+   </step>
+
+   <step>
+    <para>
+     Enable the subscription for each specified database on the target server.
+     The subscription starts applying transactions from the replication start
+     point.
+    </para>
+   </step>
+
+   <step>
+    <para>
+     If the standby server was using
+     <link linkend="guc-primary-slot-name"><varname>primary_slot_name</varname></link>,
+     it has no use from now on so drop it.
+    </para>
+   </step>
+
+   <step>
+    <para>
+     Update the system identifier on the target server. The
+     <xref linkend="app-pgresetwal"/> is run to modify the system identifier.
+     The target server is stopped as a <command>pg_resetwal</command> requirement.
+    </para>
+   </step>
+  </procedure>
+ </refsect1>
+
+ <refsect1>
+  <title>Examples</title>
+
+  <para>
+   To create a logical replica for databases <literal>hr</literal> and
+   <literal>finance</literal> from a physical replica at <literal>foo</literal>:
+<screen>
+<prompt>$</prompt> <userinput>pg_createsubscriber -D /usr/local/pgsql/data -P "host=foo" -d hr -d finance</userinput>
+</screen>
+  </para>
+
+ </refsect1>
+
+ <refsect1>
+  <title>See Also</title>
+
+  <simplelist type="inline">
+   <member><xref linkend="app-pgbasebackup"/></member>
+  </simplelist>
+ </refsect1>
+
+</refentry>
diff --git a/doc/src/sgml/reference.sgml b/doc/src/sgml/reference.sgml
index aa94f6adf6..ff85ace83f 100644
--- a/doc/src/sgml/reference.sgml
+++ b/doc/src/sgml/reference.sgml
@@ -282,6 +282,7 @@
    &pgarchivecleanup;
    &pgChecksums;
    &pgControldata;
+   &pgCreateSubscriber;
    &pgCtl;
    &pgResetwal;
    &pgRewind;
diff --git a/src/bin/pg_basebackup/.gitignore b/src/bin/pg_basebackup/.gitignore
index 26048bdbd8..14d5de6c01 100644
--- a/src/bin/pg_basebackup/.gitignore
+++ b/src/bin/pg_basebackup/.gitignore
@@ -1,4 +1,5 @@
 /pg_basebackup
+/pg_createsubscriber
 /pg_receivewal
 /pg_recvlogical
 
diff --git a/src/bin/pg_basebackup/Makefile b/src/bin/pg_basebackup/Makefile
index abfb6440ec..e9a920dbcd 100644
--- a/src/bin/pg_basebackup/Makefile
+++ b/src/bin/pg_basebackup/Makefile
@@ -44,11 +44,14 @@ BBOBJS = \
 	bbstreamer_tar.o \
 	bbstreamer_zstd.o
 
-all: pg_basebackup pg_receivewal pg_recvlogical
+all: pg_basebackup pg_createsubscriber pg_receivewal pg_recvlogical
 
 pg_basebackup: $(BBOBJS) $(OBJS) | submake-libpq submake-libpgport submake-libpgfeutils
 	$(CC) $(CFLAGS) $(BBOBJS) $(OBJS) $(LDFLAGS) $(LDFLAGS_EX) $(LIBS) -o $@$(X)
 
+pg_createsubscriber: $(WIN32RES) pg_createsubscriber.o | submake-libpq submake-libpgport submake-libpgfeutils
+	$(CC) $(CFLAGS) $^ $(LDFLAGS) $(LDFLAGS_EX) $(LIBS) -o $@$(X)
+
 pg_receivewal: pg_receivewal.o $(OBJS) | submake-libpq submake-libpgport submake-libpgfeutils
 	$(CC) $(CFLAGS) pg_receivewal.o $(OBJS) $(LDFLAGS) $(LDFLAGS_EX) $(LIBS) -o $@$(X)
 
@@ -57,6 +60,7 @@ pg_recvlogical: pg_recvlogical.o $(OBJS) | submake-libpq submake-libpgport subma
 
 install: all installdirs
 	$(INSTALL_PROGRAM) pg_basebackup$(X) '$(DESTDIR)$(bindir)/pg_basebackup$(X)'
+	$(INSTALL_PROGRAM) pg_createsubscriber$(X) '$(DESTDIR)$(bindir)/pg_createsubscriber$(X)'
 	$(INSTALL_PROGRAM) pg_receivewal$(X) '$(DESTDIR)$(bindir)/pg_receivewal$(X)'
 	$(INSTALL_PROGRAM) pg_recvlogical$(X) '$(DESTDIR)$(bindir)/pg_recvlogical$(X)'
 
@@ -65,12 +69,13 @@ installdirs:
 
 uninstall:
 	rm -f '$(DESTDIR)$(bindir)/pg_basebackup$(X)'
+	rm -f '$(DESTDIR)$(bindir)/pg_createsubscriber$(X)'
 	rm -f '$(DESTDIR)$(bindir)/pg_receivewal$(X)'
 	rm -f '$(DESTDIR)$(bindir)/pg_recvlogical$(X)'
 
 clean distclean:
-	rm -f pg_basebackup$(X) pg_receivewal$(X) pg_recvlogical$(X) \
-		$(BBOBJS) pg_receivewal.o pg_recvlogical.o \
+	rm -f pg_basebackup$(X) pg_createsubscriber$(X) pg_receivewal$(X) pg_recvlogical$(X) \
+		$(BBOBJS) pg_createsubscriber.o pg_receivewal.o pg_recvlogical.o \
 		$(OBJS)
 	rm -rf tmp_check
 
diff --git a/src/bin/pg_basebackup/meson.build b/src/bin/pg_basebackup/meson.build
index f7e60e6670..c00acd5e11 100644
--- a/src/bin/pg_basebackup/meson.build
+++ b/src/bin/pg_basebackup/meson.build
@@ -38,6 +38,24 @@ pg_basebackup = executable('pg_basebackup',
 bin_targets += pg_basebackup

+pg_createsubscriber_sources = files(
+  'pg_createsubscriber.c'
+)
+
+if host_system == 'windows'
+  pg_createsubscriber_sources += rc_bin_gen.process(win32ver_rc, extra_args: [
+	'--NAME', 'pg_createsubscriber',
+	'--FILEDESC', 'pg_createsubscriber - create a new logical replica from a standby server',])
+endif
+
+pg_createsubscriber = executable('pg_createsubscriber',
+  pg_createsubscriber_sources,
+  dependencies: [frontend_code, libpq],
+  kwargs: default_bin_args,
+)
+bin_targets += pg_createsubscriber
+
+
 pg_receivewal_sources = files(
   'pg_receivewal.c',
 )
@@ -89,6 +107,7 @@ tests += {
       't/011_in_place_tablespace.pl',
       't/020_pg_receivewal.pl',
       't/030_pg_recvlogical.pl',
+      't/040_pg_createsubscriber.pl',
     ],
   },
 }
diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
new file mode 100644
index 0000000000..c565f8524a
--- /dev/null
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -0,0 +1,2005 @@
+/*-------------------------------------------------------------------------
+ *
+ * pg_createsubscriber.c
+ *	  Create a new logical replica from a standby server
+ *
+ * Copyright (C) 2024, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ *	  src/bin/pg_basebackup/pg_createsubscriber.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres_fe.h"
+
+#include <sys/time.h>
+#include <sys/wait.h>
+#include <time.h>
+
+#include "catalog/pg_authid_d.h"
+#include "common/connect.h"
+#include "common/controldata_utils.h"
+#include "common/file_perm.h"
+#include "common/logging.h"
+#include "common/pg_prng.h"
+#include "common/restricted_token.h"
+#include "fe_utils/recovery_gen.h"
+#include "fe_utils/simple_list.h"
+#include "getopt_long.h"
+
+#define	DEFAULT_SUB_PORT	"50432"
+
+/* Command-line options */
+struct CreateSubscriberOptions
+{
+	char	   *config_file;	/* configuration file */
+	char	   *pub_conninfo_str;	/* publisher connection string */
+	char	   *socket_dir;		/* directory for Unix-domain socket, if any */
+	char	   *sub_port;		/* subscriber port number */
+	const char *sub_username;	/* subscriber username */
+	SimpleStringList database_names;	/* list of database names */
+	int			recovery_timeout;	/* stop recovery after this time */
+}			CreateSubscriberOptions;
+
+struct LogicalRepInfo
+{
+	Oid			oid;			/* database OID */
+	char	   *dbname;			/* database name */
+	char	   *pubconninfo;	/* publisher connection string */
+	char	   *subconninfo;	/* subscriber connection string */
+	char	   *pubname;		/* publication name */
+	char	   *subname;		/* subscription name / replication slot name */
+
+	bool		made_replslot;	/* replication slot was created */
+	bool		made_publication;	/* publication was created */
+}			LogicalRepInfo;
+
+static void cleanup_objects_atexit(void);
+static void usage();
+static char *get_base_conninfo(char *conninfo, char **dbname);
+static char *get_sub_conninfo(struct CreateSubscriberOptions *opt);
+static char *get_exec_path(const char *argv0, const char *progname);
+static void check_data_directory(const char *datadir);
+static char *concat_conninfo_dbname(const char *conninfo, const char *dbname);
+static struct LogicalRepInfo *store_pub_sub_info(SimpleStringList dbnames,
+												 const char *pub_base_conninfo,
+												 const char *sub_base_conninfo);
+static PGconn *connect_database(const char *conninfo, bool exit_on_error);
+static void disconnect_database(PGconn *conn, bool exit_on_error);
+static uint64 get_primary_sysid(const char *conninfo);
+static uint64 get_standby_sysid(const char *datadir);
+static void modify_subscriber_sysid(struct CreateSubscriberOptions *opt);
+static bool server_is_in_recovery(PGconn *conn);
+static void check_publisher(struct LogicalRepInfo *dbinfo);
+static char *setup_publisher(struct LogicalRepInfo *dbinfo);
+static void check_subscriber(struct LogicalRepInfo *dbinfo);
+static void setup_subscriber(struct LogicalRepInfo *dbinfo,
+							 const char *consistent_lsn);
+static void setup_recovery(struct LogicalRepInfo *dbinfo, char *datadir,
+						   char *lsn);
+static void drop_primary_replication_slot(struct LogicalRepInfo *dbinfo,
+										  char *slotname);
+static char *create_logical_replication_slot(PGconn *conn,
+											 struct LogicalRepInfo *dbinfo);
+static void drop_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo,
+								  const char *slot_name);
+static void pg_ctl_status(const char *pg_ctl_cmd, int rc);
+static void start_standby_server(struct CreateSubscriberOptions *opt,
+								 bool restricted_access);
+static void stop_standby_server(const char *datadir);
+static void wait_for_end_recovery(const char *conninfo,
+								  struct CreateSubscriberOptions *opt);
+static void create_publication(PGconn *conn, struct LogicalRepInfo *dbinfo);
+static void drop_publication(PGconn *conn, struct LogicalRepInfo *dbinfo);
+static void create_subscription(PGconn *conn, struct LogicalRepInfo *dbinfo);
+static void set_replication_progress(PGconn *conn, struct LogicalRepInfo *dbinfo,
+									 const char *lsn);
+static void enable_subscription(PGconn *conn, struct LogicalRepInfo *dbinfo);
+
+#define	USEC_PER_SEC	1000000
+#define	WAIT_INTERVAL	1		/* 1 second */
+
+static const char *progname;
+
+static char *primary_slot_name = NULL;
+static bool dry_run = false;
+
+static bool success = false;
+
+static struct LogicalRepInfo *dbinfo;
+static int	num_dbs = 0;
+
+static char *pg_ctl_path = NULL;
+static char *pg_resetwal_path = NULL;
+
+/* standby / subscriber data directory */
+static char *subscriber_dir = NULL;
+
+static bool recovery_ended = false;
+static bool standby_running = false;
+
+enum WaitPMResult
+{
+	POSTMASTER_READY,
+	POSTMASTER_STILL_STARTING
+};
+
+
+/*
+ * Cleanup objects that were created by pg_createsubscriber if there is an
+ * error.
+ *
+ * Replication slots, publications and subscriptions are created. Depending on
+ * the step it failed, it should remove the already created objects if it is
+ * possible (sometimes it won't work due to a connection issue).
+ */
+static void
+cleanup_objects_atexit(void)
+{
+	if (success)
+		return;
+
+	/*
+	 * If the server is promoted, there is no way to use the current setup
+	 * again. Warn the user that a new replication setup should be done before
+	 * trying again.
+	 */
+	if (recovery_ended)
+	{
+		pg_log_warning("pg_createsubscriber failed after the end of recovery");
+		pg_log_warning_hint("The target server cannot be used as a physical replica anymore.");
+		pg_log_warning_hint("You must recreate the physical replica before continuing.");
+	}
+
+	for (int i = 0; i < num_dbs; i++)
+	{
+		if (dbinfo[i].made_publication || dbinfo[i].made_replslot)
+		{
+			PGconn	   *conn;
+
+			conn = connect_database(dbinfo[i].pubconninfo, false);
+			if (conn != NULL)
+			{
+				if (dbinfo[i].made_publication)
+					drop_publication(conn, &dbinfo[i]);
+				if (dbinfo[i].made_replslot)
+					drop_replication_slot(conn, &dbinfo[i], dbinfo[i].subname);
+				disconnect_database(conn, false);
+			}
+			else
+			{
+				/*
+				 * If a connection could not be established, inform the user
+				 * that some objects were left on primary and should be
+				 * removed before trying again.
+				 */
+				if (dbinfo[i].made_publication)
+				{
+					pg_log_warning("There might be a publication \"%s\" in database \"%s\" on primary",
+								   dbinfo[i].pubname, dbinfo[i].dbname);
+					pg_log_warning_hint("Consider dropping this publication before trying again.");
+				}
+				if (dbinfo[i].made_replslot)
+				{
+					pg_log_warning("There might be a replication slot \"%s\" in database \"%s\" on primary",
+								   dbinfo[i].subname, dbinfo[i].dbname);
+					pg_log_warning_hint("Drop this replication slot soon to avoid retention of WAL files.");
+				}
+			}
+		}
+	}
+
+	if (standby_running)
+		stop_standby_server(subscriber_dir);
+}
+
+static void
+usage(void)
+{
+	printf(_("%s creates a new logical replica from a standby server.\n\n"),
+		   progname);
+	printf(_("Usage:\n"));
+	printf(_("  %s [OPTION]...\n"), progname);
+	printf(_("\nOptions:\n"));
+	printf(_(" -d, --database=DBNAME               database to create a subscription\n"));
+	printf(_(" -D, --pgdata=DATADIR                location for the subscriber data directory\n"));
+	printf(_(" -n, --dry-run                       dry run, just show what would be done\n"));
+	printf(_(" -p, --subscriber-port=PORT          subscriber port number (default %s)\n"), DEFAULT_SUB_PORT);
+	printf(_(" -P, --publisher-server=CONNSTR      publisher connection string\n"));
+	printf(_(" -s, --socket-directory=DIR          socket directory to use (default current directory)\n"));
+	printf(_(" -t, --recovery-timeout=SECS         seconds to wait for recovery to end\n"));
+	printf(_(" -U, --subscriber-username=NAME      subscriber username\n"));
+	printf(_(" -v, --verbose                       output verbose messages\n"));
+	printf(_("      --config-file=FILENAME         use specified main server configuration\n"
+			 "                                     file when running target cluster\n"));
+	printf(_(" -V, --version                       output version information, then exit\n"));
+	printf(_(" -?, --help                          show this help, then exit\n"));
+	printf(_("\nReport bugs to <%s>.\n"), PACKAGE_BUGREPORT);
+	printf(_("%s home page: <%s>\n"), PACKAGE_NAME, PACKAGE_URL);
+}
+
+/*
+ * Validate a connection string. Returns a base connection string that is a
+ * connection string without a database name.
+ *
+ * Since we might process multiple databases, each database name will be
+ * appended to this base connection string to provide a final connection
+ * string. If the second argument (dbname) is not null, returns dbname if the
+ * provided connection string contains it. If option --database is not
+ * provided, uses dbname as the only database to setup the logical replica.
+ *
+ * It is the caller's responsibility to free the returned connection string and
+ * dbname.
+ */
+static char *
+get_base_conninfo(char *conninfo, char **dbname)
+{
+	PQExpBuffer buf = createPQExpBuffer();
+	PQconninfoOption *conn_opts = NULL;
+	PQconninfoOption *conn_opt;
+	char	   *errmsg = NULL;
+	char	   *ret;
+	int			i;
+
+	conn_opts = PQconninfoParse(conninfo, &errmsg);
+	if (conn_opts == NULL)
+	{
+		pg_log_error("could not parse connection string: %s", errmsg);
+		return NULL;
+	}
+
+	i = 0;
+	for (conn_opt = conn_opts; conn_opt->keyword != NULL; conn_opt++)
+	{
+		if (strcmp(conn_opt->keyword, "dbname") == 0 && conn_opt->val != NULL)
+		{
+			if (dbname)
+				*dbname = pg_strdup(conn_opt->val);
+			continue;
+		}
+
+		if (conn_opt->val != NULL && conn_opt->val[0] != '\0')
+		{
+			if (i > 0)
+				appendPQExpBufferChar(buf, ' ');
+			appendPQExpBuffer(buf, "%s=%s", conn_opt->keyword, conn_opt->val);
+			i++;
+		}
+	}
+
+	ret = pg_strdup(buf->data);
+
+	destroyPQExpBuffer(buf);
+	PQconninfoFree(conn_opts);
+
+	return ret;
+}
+
+/*
+ * Build a subscriber connection string. Only a few parameters are supported
+ * since it starts a server with restricted access.
+ */
+static char *
+get_sub_conninfo(struct CreateSubscriberOptions *opt)
+{
+	PQExpBuffer buf = createPQExpBuffer();
+	char	   *ret;
+
+	appendPQExpBuffer(buf, "port=%s", opt->sub_port);
+#if !defined(WIN32)
+	appendPQExpBuffer(buf, " host=%s", opt->socket_dir);
+#endif
+	if (opt->sub_username != NULL)
+		appendPQExpBuffer(buf, " user=%s", opt->sub_username);
+	appendPQExpBuffer(buf, " fallback_application_name=%s", progname);
+
+	ret = pg_strdup(buf->data);
+
+	destroyPQExpBuffer(buf);
+
+	return ret;
+}
+
+/*
+ * Verify if a PostgreSQL binary (progname) is available in the same directory as
+ * pg_createsubscriber and it has the same version.  It returns the absolute
+ * path of the progname.
+ */
+static char *
+get_exec_path(const char *argv0, const char *progname)
+{
+	char	   *versionstr;
+	char	   *exec_path;
+	int			ret;
+
+	versionstr = psprintf("%s (PostgreSQL) %s\n", progname, PG_VERSION);
+	exec_path = pg_malloc(MAXPGPATH);
+	ret = find_other_exec(argv0, progname, versionstr, exec_path);
+
+	if (ret < 0)
+	{
+		char		full_path[MAXPGPATH];
+
+		if (find_my_exec(argv0, full_path) < 0)
+			strlcpy(full_path, progname, sizeof(full_path));
+
+		if (ret == -1)
+			pg_fatal("program \"%s\" is needed by %s but was not found in the same directory as \"%s\"",
+					 progname, "pg_createsubscriber", full_path);
+		else
+			pg_fatal("program \"%s\" was found by \"%s\" but was not the same version as %s",
+					 progname, full_path, "pg_createsubscriber");
+	}
+
+	pg_log_debug("%s path is:  %s", progname, exec_path);
+
+	return exec_path;
+}
+
+/*
+ * Is it a cluster directory? These are preliminary checks. It is far from
+ * making an accurate check. If it is not a clone from the publisher, it will
+ * eventually fail in a future step.
+ */
+static void
+check_data_directory(const char *datadir)
+{
+	struct stat statbuf;
+	char		versionfile[MAXPGPATH];
+
+	pg_log_info("checking if directory \"%s\" is a cluster data directory",
+				datadir);
+
+	if (stat(datadir, &statbuf) != 0)
+	{
+		if (errno == ENOENT)
+			pg_fatal("data directory \"%s\" does not exist", datadir);
+		else
+			pg_fatal("could not access directory \"%s\": %s", datadir,
+					 strerror(errno));
+	}
+
+	snprintf(versionfile, MAXPGPATH, "%s/PG_VERSION", datadir);
+	if (stat(versionfile, &statbuf) != 0 && errno == ENOENT)
+	{
+		pg_fatal("directory \"%s\" is not a database cluster directory",
+				 datadir);
+	}
+}
+
+/*
+ * Append database name into a base connection string.
+ *
+ * dbname is the only parameter that changes so it is not included in the base
+ * connection string. This function concatenates dbname to build a "real"
+ * connection string.
+ */
+static char *
+concat_conninfo_dbname(const char *conninfo, const char *dbname)
+{
+	PQExpBuffer buf = createPQExpBuffer();
+	char	   *ret;
+
+	Assert(conninfo != NULL);
+
+	appendPQExpBufferStr(buf, conninfo);
+	appendPQExpBuffer(buf, " dbname=%s", dbname);
+
+	ret = pg_strdup(buf->data);
+	destroyPQExpBuffer(buf);
+
+	return ret;
+}
+
+/*
+ * Store publication and subscription information.
+ */
+static struct LogicalRepInfo *
+store_pub_sub_info(SimpleStringList dbnames, const char *pub_base_conninfo,
+				   const char *sub_base_conninfo)
+{
+	struct LogicalRepInfo *dbinfo;
+	int			i = 0;
+
+	dbinfo = (struct LogicalRepInfo *) pg_malloc(num_dbs * sizeof(struct LogicalRepInfo));
+
+	for (SimpleStringListCell *cell = dbnames.head; cell; cell = cell->next)
+	{
+		char	   *conninfo;
+
+		/* Fill publisher attributes */
+		conninfo = concat_conninfo_dbname(pub_base_conninfo, cell->val);
+		dbinfo[i].pubconninfo = conninfo;
+		dbinfo[i].dbname = cell->val;
+		dbinfo[i].made_replslot = false;
+		dbinfo[i].made_publication = false;
+		/* Fill subscriber attributes */
+		conninfo = concat_conninfo_dbname(sub_base_conninfo, cell->val);
+		dbinfo[i].subconninfo = conninfo;
+		/* Other fields will be filled later */
+
+		pg_log_debug("publisher(%d): connection string: %s", i, dbinfo[i].pubconninfo);
+		pg_log_debug("subscriber(%d): connection string: %s", i, dbinfo[i].subconninfo);
+
+		i++;
+	}
+
+	return dbinfo;
+}
+
+/*
+ * Open a new connection. If exit_on_error is true, it has an undesired
+ * condition and it should exit immediately.
+ */
+static PGconn *
+connect_database(const char *conninfo, bool exit_on_error)
+{
+	PGconn	   *conn;
+	PGresult   *res;
+
+	conn = PQconnectdb(conninfo);
+	if (PQstatus(conn) != CONNECTION_OK)
+	{
+		pg_log_error("connection to database failed: %s",
+					 PQerrorMessage(conn));
+		if (exit_on_error)
+			exit(1);
+
+		return NULL;
+	}
+
+	/* Secure search_path */
+	res = PQexec(conn, ALWAYS_SECURE_SEARCH_PATH_SQL);
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		pg_log_error("could not clear search_path: %s",
+					 PQresultErrorMessage(res));
+		if (exit_on_error)
+			exit(1);
+
+		return NULL;
+	}
+	PQclear(res);
+
+	return conn;
+}
+
+/*
+ * Close the connection. If exit_on_error is true, it has an undesired
+ * condition and it should exit immediately.
+ */
+static void
+disconnect_database(PGconn *conn, bool exit_on_error)
+{
+	Assert(conn != NULL);
+
+	PQfinish(conn);
+
+	if (exit_on_error)
+		exit(1);
+}
+
+/*
+ * Obtain the system identifier using the provided connection. It will be used
+ * to compare if a data directory is a clone of another one.
+ */
+static uint64
+get_primary_sysid(const char *conninfo)
+{
+	PGconn	   *conn;
+	PGresult   *res;
+	uint64		sysid;
+
+	pg_log_info("getting system identifier from publisher");
+
+	conn = connect_database(conninfo, true);
+
+	res = PQexec(conn, "SELECT system_identifier FROM pg_catalog.pg_control_system()");
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		pg_log_error("could not get system identifier: %s",
+					 PQresultErrorMessage(res));
+		disconnect_database(conn, true);
+	}
+	if (PQntuples(res) != 1)
+	{
+		pg_log_error("could not get system identifier: got %d rows, expected %d row",
+					 PQntuples(res), 1);
+		disconnect_database(conn, true);
+	}
+
+	sysid = strtou64(PQgetvalue(res, 0, 0), NULL, 10);
+
+	pg_log_info("system identifier is %llu on publisher",
+				(unsigned long long) sysid);
+
+	PQclear(res);
+	disconnect_database(conn, false);
+
+	return sysid;
+}
+
+/*
+ * Obtain the system identifier from control file. It will be used to compare
+ * if a data directory is a clone of another one. This routine is used locally
+ * and avoids a connection.
+ */
+static uint64
+get_standby_sysid(const char *datadir)
+{
+	ControlFileData *cf;
+	bool		crc_ok;
+	uint64		sysid;
+
+	pg_log_info("getting system identifier from subscriber");
+
+	cf = get_controlfile(datadir, &crc_ok);
+	if (!crc_ok)
+		pg_fatal("control file appears to be corrupt");
+
+	sysid = cf->system_identifier;
+
+	pg_log_info("system identifier is %llu on subscriber",
+				(unsigned long long) sysid);
+
+	pg_free(cf);
+
+	return sysid;
+}
+
+/*
+ * Modify the system identifier. Since a standby server preserves the system
+ * identifier, it makes sense to change it to avoid situations in which WAL
+ * files from one of the systems might be used in the other one.
+ */
+static void
+modify_subscriber_sysid(struct CreateSubscriberOptions *opt)
+{
+	ControlFileData *cf;
+	bool		crc_ok;
+	struct timeval tv;
+
+	char	   *cmd_str;
+
+	pg_log_info("modifying system identifier from subscriber");
+
+	cf = get_controlfile(subscriber_dir, &crc_ok);
+	if (!crc_ok)
+		pg_fatal("control file appears to be corrupt");
+
+	/*
+	 * Select a new system identifier.
+	 *
+	 * XXX this code was extracted from BootStrapXLOG().
+	 */
+	gettimeofday(&tv, NULL);
+	cf->system_identifier = ((uint64) tv.tv_sec) << 32;
+	cf->system_identifier |= ((uint64) tv.tv_usec) << 12;
+	cf->system_identifier |= getpid() & 0xFFF;
+
+	if (!dry_run)
+		update_controlfile(subscriber_dir, cf, true);
+
+	pg_log_info("system identifier is %llu on subscriber",
+				(unsigned long long) cf->system_identifier);
+
+	pg_log_info("running pg_resetwal on the subscriber");
+
+	cmd_str = psprintf("\"%s\" -D \"%s\" > \"%s\"", pg_resetwal_path,
+					   subscriber_dir, DEVNULL);
+
+	pg_log_debug("pg_resetwal command is: %s", cmd_str);
+
+	if (!dry_run)
+	{
+		int			rc = system(cmd_str);
+
+		if (rc == 0)
+			pg_log_info("subscriber successfully changed the system identifier");
+		else
+			pg_fatal("subscriber failed to change system identifier: exit code: %d", rc);
+	}
+
+	pg_free(cf);
+}
+
+/*
+ * Create the publications and replication slots in preparation for logical
+ * replication. Returns the LSN from latest replication slot. It will be the
+ * replication start point that is used to adjust the subscriptions (see
+ * set_replication_progress).
+ */
+static char *
+setup_publisher(struct LogicalRepInfo *dbinfo)
+{
+	pg_prng_state prng_state;
+	char	   *lsn = NULL;
+
+	pg_prng_seed(&prng_state, (uint64) (getpid() ^ time(NULL)));
+
+	for (int i = 0; i < num_dbs; i++)
+	{
+		PGconn	   *conn;
+		PGresult   *res;
+		char		pubname[NAMEDATALEN];
+		char		replslotname[NAMEDATALEN];
+		uint32		rand;
+
+		conn = connect_database(dbinfo[i].pubconninfo, true);
+
+		res = PQexec(conn,
+					 "SELECT oid FROM pg_catalog.pg_database "
+					 "WHERE datname = pg_catalog.current_database()");
+		if (PQresultStatus(res) != PGRES_TUPLES_OK)
+		{
+			pg_log_error("could not obtain database OID: %s",
+						 PQresultErrorMessage(res));
+			disconnect_database(conn, true);
+		}
+
+		if (PQntuples(res) != 1)
+		{
+			pg_log_error("could not obtain database OID: got %d rows, expected %d rows",
+						 PQntuples(res), 1);
+			disconnect_database(conn, true);
+		}
+
+		/* Remember database OID */
+		dbinfo[i].oid = strtoul(PQgetvalue(res, 0, 0), NULL, 10);
+
+		PQclear(res);
+
+		/* Random unsigned integer */
+		rand = pg_prng_uint32(&prng_state);
+
+		/*
+		 * Build the publication name. The name must not exceed NAMEDATALEN -
+		 * 1. This current schema uses a maximum of 40 characters (20 + 10 + 1
+		 * + 8 + '\0').
+		 */
+		snprintf(pubname, sizeof(pubname), "pg_createsubscriber_%u_%x",
+				 dbinfo[i].oid, rand);
+		dbinfo[i].pubname = pg_strdup(pubname);
+
+		/*
+		 * Create publication on publisher. This step should be executed
+		 * *before* promoting the subscriber to avoid any transactions between
+		 * consistent LSN and the new publication rows (such transactions
+		 * wouldn't see the new publication rows resulting in an error).
+		 */
+		create_publication(conn, &dbinfo[i]);
+
+		/*
+		 * Build the replication slot name. The name must not exceed
+		 * NAMEDATALEN - 1. This current schema uses a maximum of 40
+		 * characters (20 + 10 + 1 + 8 + '\0'). A random number is included to
+		 * reduce the probability of collision. By default, subscription name
+		 * is used as replication slot name.
+		 */
+		snprintf(replslotname, sizeof(replslotname),
+				 "pg_createsubscriber_%u_%x",
+				 dbinfo[i].oid,
+				 rand);
+		dbinfo[i].subname = pg_strdup(replslotname);
+
+		/* Create replication slot on publisher */
+		if (lsn)
+			pg_free(lsn);
+		lsn = create_logical_replication_slot(conn, &dbinfo[i]);
+		if (lsn != NULL || dry_run)
+			pg_log_info("create replication slot \"%s\" on publisher",
+						replslotname);
+		else
+			exit(1);
+
+		disconnect_database(conn, false);
+	}
+
+	return lsn;
+}
+
+/*
+ * Is recovery still in progress?
+ */
+static bool
+server_is_in_recovery(PGconn *conn)
+{
+	PGresult   *res;
+	int			ret;
+
+	res = PQexec(conn, "SELECT pg_catalog.pg_is_in_recovery()");
+
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		pg_log_error("could not obtain recovery progress: %s",
+					 PQresultErrorMessage(res));
+		disconnect_database(conn, true);
+	}
+
+
+	ret = strcmp("t", PQgetvalue(res, 0, 0));
+
+	PQclear(res);
+
+	return ret == 0;
+}
+
+/*
+ * Is the primary server ready for logical replication?
+ *
+ * XXX Does it not allow a synchronous replica?
+ */
+static void
+check_publisher(struct LogicalRepInfo *dbinfo)
+{
+	PGconn	   *conn;
+	PGresult   *res;
+	bool		failed = false;
+
+	char	   *wal_level;
+	int			max_repslots;
+	int			cur_repslots;
+	int			max_walsenders;
+	int			cur_walsenders;
+
+	pg_log_info("checking settings on publisher");
+
+	conn = connect_database(dbinfo[0].pubconninfo, true);
+
+	/*
+	 * If the primary server is in recovery (i.e. cascading replication),
+	 * objects (publication) cannot be created because it is read only.
+	 */
+	if (server_is_in_recovery(conn))
+	{
+		pg_log_error("primary server cannot be in recovery");
+		disconnect_database(conn, true);
+	}
+
+	/*------------------------------------------------------------------------
+	 * Logical replication requires a few parameters to be set on publisher.
+	 * Since these parameters are not a requirement for physical replication,
+	 * we should check it to make sure it won't fail.
+	 *
+	 * - wal_level = logical
+	 * - max_replication_slots >= current + number of dbs to be converted
+	 * - max_wal_senders >= current + number of dbs to be converted
+	 * -----------------------------------------------------------------------
+	 */
+	res = PQexec(conn,
+				 "WITH wl AS "
+				 "(SELECT setting AS wallevel FROM pg_catalog.pg_settings "
+				 "WHERE name = 'wal_level'), "
+				 "total_mrs AS "
+				 "(SELECT setting AS tmrs FROM pg_catalog.pg_settings "
+				 "WHERE name = 'max_replication_slots'), "
+				 "cur_mrs AS "
+				 "(SELECT count(*) AS cmrs "
+				 "FROM pg_catalog.pg_replication_slots), "
+				 "total_mws AS "
+				 "(SELECT setting AS tmws FROM pg_catalog.pg_settings "
+				 "WHERE name = 'max_wal_senders'), "
+				 "cur_mws AS "
+				 "(SELECT count(*) AS cmws FROM pg_catalog.pg_stat_activity "
+				 "WHERE backend_type = 'walsender') "
+				 "SELECT wallevel, tmrs, cmrs, tmws, cmws "
+				 "FROM wl, total_mrs, cur_mrs, total_mws, cur_mws");
+
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		pg_log_error("could not obtain publisher settings: %s",
+					 PQresultErrorMessage(res));
+		disconnect_database(conn, true);
+	}
+
+	wal_level = strdup(PQgetvalue(res, 0, 0));
+	max_repslots = atoi(PQgetvalue(res, 0, 1));
+	cur_repslots = atoi(PQgetvalue(res, 0, 2));
+	max_walsenders = atoi(PQgetvalue(res, 0, 3));
+	cur_walsenders = atoi(PQgetvalue(res, 0, 4));
+
+	PQclear(res);
+
+	pg_log_debug("publisher: wal_level: %s", wal_level);
+	pg_log_debug("publisher: max_replication_slots: %d", max_repslots);
+	pg_log_debug("publisher: current replication slots: %d", cur_repslots);
+	pg_log_debug("publisher: max_wal_senders: %d", max_walsenders);
+	pg_log_debug("publisher: current wal senders: %d", cur_walsenders);
+
+	/*
+	 * If standby sets primary_slot_name, check if this replication slot is in
+	 * use on primary for WAL retention purposes. This replication slot has no
+	 * use after the transformation, hence, it will be removed at the end of
+	 * this process.
+	 */
+	if (primary_slot_name)
+	{
+		PQExpBuffer str = createPQExpBuffer();
+
+		appendPQExpBuffer(str,
+						  "SELECT 1 FROM pg_catalog.pg_replication_slots "
+						  "WHERE active AND slot_name = '%s'",
+						  primary_slot_name);
+
+		pg_log_debug("command is: %s", str->data);
+
+		res = PQexec(conn, str->data);
+		if (PQresultStatus(res) != PGRES_TUPLES_OK)
+		{
+			pg_log_error("could not obtain replication slot information: %s",
+						 PQresultErrorMessage(res));
+			disconnect_database(conn, true);
+		}
+
+		if (PQntuples(res) != 1)
+		{
+			pg_log_error("could not obtain replication slot information: got %d rows, expected %d row",
+						 PQntuples(res), 1);
+			disconnect_database(conn, true);
+		}
+		else
+			pg_log_info("primary has replication slot \"%s\"",
+						primary_slot_name);
+
+		PQclear(res);
+	}
+
+	disconnect_database(conn, false);
+
+	if (strcmp(wal_level, "logical") != 0)
+	{
+		pg_log_error("publisher requires wal_level >= logical");
+		failed = true;
+	}
+
+	if (max_repslots - cur_repslots < num_dbs)
+	{
+		pg_log_error("publisher requires %d replication slots, but only %d remain",
+					 num_dbs, max_repslots - cur_repslots);
+		pg_log_error_hint("Consider increasing max_replication_slots to at least %d.",
+						  cur_repslots + num_dbs);
+		failed = true;
+	}
+
+	if (max_walsenders - cur_walsenders < num_dbs)
+	{
+		pg_log_error("publisher requires %d wal sender processes, but only %d remain",
+					 num_dbs, max_walsenders - cur_walsenders);
+		pg_log_error_hint("Consider increasing max_wal_senders to at least %d.",
+						  cur_walsenders + num_dbs);
+		failed = true;
+	}
+
+	if (failed)
+		exit(1);
+}
+
+/*
+ * Is the standby server ready for logical replication?
+ *
+ * XXX Does it not allow a time-delayed replica?
+ *
+ * XXX In a cascaded replication scenario (P -> S -> C), if the target server
+ * is S, it cannot detect there is a replica (server C) because server S starts
+ * accepting only local connections and server C cannot connect to it. Hence,
+ * there is not a reliable way to provide a suitable error saying the server C
+ * will be broken at the end of this process (due to pg_resetwal).
+ */
+static void
+check_subscriber(struct LogicalRepInfo *dbinfo)
+{
+	PGconn	   *conn;
+	PGresult   *res;
+	PQExpBuffer str = createPQExpBuffer();
+	bool		failed = false;
+
+	int			max_lrworkers;
+	int			max_repslots;
+	int			max_wprocs;
+
+	pg_log_info("checking settings on subscriber");
+
+	conn = connect_database(dbinfo[0].subconninfo, true);
+
+	/* The target server must be a standby */
+	if (!server_is_in_recovery(conn))
+	{
+		pg_log_error("The target server must be a standby");
+		disconnect_database(conn, true);
+	}
+
+	/*
+	 * Subscriptions can only be created by roles that have the privileges of
+	 * pg_create_subscription role and CREATE privileges on the specified
+	 * database.
+	 */
+	appendPQExpBuffer(str,
+					  "SELECT pg_catalog.pg_has_role(current_user, %u, 'MEMBER'), "
+					  "pg_catalog.has_database_privilege(current_user, '%s', 'CREATE'), "
+					  "pg_catalog.has_function_privilege(current_user, 'pg_catalog.pg_replication_origin_advance(text, pg_lsn)', 'EXECUTE')",
+					  ROLE_PG_CREATE_SUBSCRIPTION, dbinfo[0].dbname);
+
+	pg_log_debug("command is: %s", str->data);
+
+	res = PQexec(conn, str->data);
+
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		pg_log_error("could not obtain access privilege information: %s",
+					 PQresultErrorMessage(res));
+		disconnect_database(conn, true);
+	}
+
+	if (strcmp(PQgetvalue(res, 0, 0), "t") != 0)
+	{
+		pg_log_error("permission denied to create subscription");
+		pg_log_error_hint("Only roles with privileges of the \"%s\" role may create subscriptions.",
+						  "pg_create_subscription");
+		failed = true;
+	}
+	if (strcmp(PQgetvalue(res, 0, 1), "t") != 0)
+	{
+		pg_log_error("permission denied for database %s", dbinfo[0].dbname);
+		failed = true;
+	}
+	if (strcmp(PQgetvalue(res, 0, 2), "t") != 0)
+	{
+		pg_log_error("permission denied for function \"%s\"",
+					 "pg_catalog.pg_replication_origin_advance(text, pg_lsn)");
+		failed = true;
+	}
+
+	destroyPQExpBuffer(str);
+	PQclear(res);
+
+	/*------------------------------------------------------------------------
+	 * Logical replication requires a few parameters to be set on subscriber.
+	 * Since these parameters are not a requirement for physical replication,
+	 * we should check it to make sure it won't fail.
+	 *
+	 * - max_replication_slots >= number of dbs to be converted
+	 * - max_logical_replication_workers >= number of dbs to be converted
+	 * - max_worker_processes >= 1 + number of dbs to be converted
+	 *------------------------------------------------------------------------
+	 */
+	res = PQexec(conn,
+				 "SELECT setting FROM pg_catalog.pg_settings WHERE name IN ("
+				 "'max_logical_replication_workers', "
+				 "'max_replication_slots', "
+				 "'max_worker_processes', "
+				 "'primary_slot_name') "
+				 "ORDER BY name");
+
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		pg_log_error("could not obtain subscriber settings: %s",
+					 PQresultErrorMessage(res));
+		disconnect_database(conn, true);
+	}
+
+	max_lrworkers = atoi(PQgetvalue(res, 0, 0));
+	max_repslots = atoi(PQgetvalue(res, 1, 0));
+	max_wprocs = atoi(PQgetvalue(res, 2, 0));
+	if (strcmp(PQgetvalue(res, 3, 0), "") != 0)
+		primary_slot_name = pg_strdup(PQgetvalue(res, 3, 0));
+
+	pg_log_debug("subscriber: max_logical_replication_workers: %d",
+				 max_lrworkers);
+	pg_log_debug("subscriber: max_replication_slots: %d", max_repslots);
+	pg_log_debug("subscriber: max_worker_processes: %d", max_wprocs);
+	if (primary_slot_name)
+		pg_log_debug("subscriber: primary_slot_name: %s", primary_slot_name);
+
+	PQclear(res);
+
+	disconnect_database(conn, false);
+
+	if (max_repslots < num_dbs)
+	{
+		pg_log_error("subscriber requires %d replication slots, but only %d remain",
+					 num_dbs, max_repslots);
+		pg_log_error_hint("Consider increasing max_replication_slots to at least %d.",
+						  num_dbs);
+		failed = true;
+	}
+
+	if (max_lrworkers < num_dbs)
+	{
+		pg_log_error("subscriber requires %d logical replication workers, but only %d remain",
+					 num_dbs, max_lrworkers);
+		pg_log_error_hint("Consider increasing max_logical_replication_workers to at least %d.",
+						  num_dbs);
+		failed = true;
+	}
+
+	if (max_wprocs < num_dbs + 1)
+	{
+		pg_log_error("subscriber requires %d worker processes, but only %d remain",
+					 num_dbs + 1, max_wprocs);
+		pg_log_error_hint("Consider increasing max_worker_processes to at least %d.",
+						  num_dbs + 1);
+		failed = true;
+	}
+
+	if (failed)
+		exit(1);
+}
+
+/*
+ * Create the subscriptions, adjust the initial location for logical
+ * replication and enable the subscriptions. That's the last step for logical
+ * replication setup.
+ */
+static void
+setup_subscriber(struct LogicalRepInfo *dbinfo, const char *consistent_lsn)
+{
+	for (int i = 0; i < num_dbs; i++)
+	{
+		PGconn	   *conn;
+
+		/* Connect to subscriber. */
+		conn = connect_database(dbinfo[i].subconninfo, true);
+
+		/*
+		 * Since the publication was created before the consistent LSN, it is
+		 * available on the subscriber when the physical replica is promoted.
+		 * Remove publications from the subscriber because it has no use.
+		 */
+		drop_publication(conn, &dbinfo[i]);
+
+		create_subscription(conn, &dbinfo[i]);
+
+		/* Set the replication progress to the correct LSN */
+		set_replication_progress(conn, &dbinfo[i], consistent_lsn);
+
+		/* Enable subscription */
+		enable_subscription(conn, &dbinfo[i]);
+
+		disconnect_database(conn, false);
+	}
+}
+
+/*
+ * Write the required recovery parameters.
+ */
+static void
+setup_recovery(struct LogicalRepInfo *dbinfo, char *datadir, char *lsn)
+{
+	PGconn	   *conn;
+	PQExpBuffer recoveryconfcontents;
+
+	/*
+	 * Despite of the recovery parameters will be written to the subscriber,
+	 * use a publisher connection. The primary_conninfo is generated using the
+	 * connection settings.
+	 */
+	conn = connect_database(dbinfo[0].pubconninfo, true);
+
+	/*
+	 * Write recovery parameters.
+	 *
+	 * The subscriber is not running yet. In dry run mode, the recovery
+	 * parameters *won't* be written. An invalid LSN is used for printing
+	 * purposes. Additional recovery parameters are added here. It avoids
+	 * unexpected behavior such as end of recovery as soon as a consistent
+	 * state is reached (recovery_target) and failure due to multiple recovery
+	 * targets (name, time, xid, LSN).
+	 */
+	recoveryconfcontents = GenerateRecoveryConfig(conn, NULL);
+	appendPQExpBuffer(recoveryconfcontents, "recovery_target = ''\n");
+	appendPQExpBuffer(recoveryconfcontents,
+					  "recovery_target_timeline = 'latest'\n");
+	appendPQExpBuffer(recoveryconfcontents,
+					  "recovery_target_inclusive = true\n");
+	appendPQExpBuffer(recoveryconfcontents,
+					  "recovery_target_action = promote\n");
+	appendPQExpBuffer(recoveryconfcontents, "recovery_target_name = ''\n");
+	appendPQExpBuffer(recoveryconfcontents, "recovery_target_time = ''\n");
+	appendPQExpBuffer(recoveryconfcontents, "recovery_target_xid = ''\n");
+
+	if (dry_run)
+	{
+		appendPQExpBuffer(recoveryconfcontents, "# dry run mode");
+		appendPQExpBuffer(recoveryconfcontents,
+						  "recovery_target_lsn = '%X/%X'\n",
+						  LSN_FORMAT_ARGS((XLogRecPtr) InvalidXLogRecPtr));
+	}
+	else
+	{
+		appendPQExpBuffer(recoveryconfcontents, "recovery_target_lsn = '%s'\n",
+						  lsn);
+		WriteRecoveryConfig(conn, datadir, recoveryconfcontents);
+	}
+	disconnect_database(conn, false);
+
+	pg_log_debug("recovery parameters:\n%s", recoveryconfcontents->data);
+}
+
+/*
+ * Drop physical replication slot on primary if the standby was using it. After
+ * the transformation, it has no use.
+ *
+ * XXX we might not fail here. Instead, we provide a warning so the user
+ * eventually drops this replication slot later.
+ */
+static void
+drop_primary_replication_slot(struct LogicalRepInfo *dbinfo, char *slotname)
+{
+	PGconn	   *conn;
+
+	/* Replication slot does not exist, do nothing */
+	if (!primary_slot_name)
+		return;
+
+	conn = connect_database(dbinfo[0].pubconninfo, false);
+	if (conn != NULL)
+	{
+		drop_replication_slot(conn, &dbinfo[0], slotname);
+		disconnect_database(conn, false);
+	}
+	else
+	{
+		pg_log_warning("could not drop replication slot \"%s\" on primary",
+					   slotname);
+		pg_log_warning_hint("Drop this replication slot soon to avoid retention of WAL files.");
+	}
+}
+
+/*
+ * Create a logical replication slot and returns a LSN.
+ *
+ * CreateReplicationSlot() is not used because it does not provide the one-row
+ * result set that contains the LSN.
+ */
+static char *
+create_logical_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo)
+{
+	PQExpBuffer str = createPQExpBuffer();
+	PGresult   *res = NULL;
+	char		slot_name[NAMEDATALEN];
+	char	   *lsn = NULL;
+
+	Assert(conn != NULL);
+
+	snprintf(slot_name, NAMEDATALEN, "%s", dbinfo->subname);
+
+	pg_log_info("creating the replication slot \"%s\" on database \"%s\"",
+				slot_name, dbinfo->dbname);
+
+	appendPQExpBuffer(str,
+					  "SELECT lsn FROM pg_catalog.pg_create_logical_replication_slot('%s', '%s', %s, false, false)",
+					  slot_name, "pgoutput", "false");
+
+	pg_log_debug("command is: %s", str->data);
+
+	if (!dry_run)
+	{
+		res = PQexec(conn, str->data);
+		if (PQresultStatus(res) != PGRES_TUPLES_OK)
+		{
+			pg_log_error("could not create replication slot \"%s\" on database \"%s\": %s",
+						 slot_name, dbinfo->dbname,
+						 PQresultErrorMessage(res));
+			return NULL;
+		}
+
+		lsn = pg_strdup(PQgetvalue(res, 0, 0));
+		PQclear(res);
+	}
+
+	/* For cleanup purposes */
+	dbinfo->made_replslot = true;
+
+	destroyPQExpBuffer(str);
+
+	return lsn;
+}
+
+static void
+drop_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo,
+					  const char *slot_name)
+{
+	PQExpBuffer str = createPQExpBuffer();
+	PGresult   *res;
+
+	Assert(conn != NULL);
+
+	pg_log_info("dropping the replication slot \"%s\" on database \"%s\"",
+				slot_name, dbinfo->dbname);
+
+	appendPQExpBuffer(str, "SELECT pg_catalog.pg_drop_replication_slot('%s')", slot_name);
+
+	pg_log_debug("command is: %s", str->data);
+
+	if (!dry_run)
+	{
+		res = PQexec(conn, str->data);
+		if (PQresultStatus(res) != PGRES_TUPLES_OK)
+		{
+			pg_log_error("could not drop replication slot \"%s\" on database \"%s\": %s",
+						 slot_name, dbinfo->dbname, PQresultErrorMessage(res));
+			dbinfo->made_replslot = false;	/* don't try again. */
+		}
+
+		PQclear(res);
+	}
+
+	destroyPQExpBuffer(str);
+}
+
+/*
+ * Reports a suitable message if pg_ctl fails.
+ */
+static void
+pg_ctl_status(const char *pg_ctl_cmd, int rc)
+{
+	if (rc != 0)
+	{
+		if (WIFEXITED(rc))
+		{
+			pg_log_error("pg_ctl failed with exit code %d", WEXITSTATUS(rc));
+		}
+		else if (WIFSIGNALED(rc))
+		{
+#if defined(WIN32)
+			pg_log_error("pg_ctl was terminated by exception 0x%X",
+						 WTERMSIG(rc));
+			pg_log_error_detail("See C include file \"ntstatus.h\" for a description of the hexadecimal value.");
+#else
+			pg_log_error("pg_ctl was terminated by signal %d: %s",
+						 WTERMSIG(rc), pg_strsignal(WTERMSIG(rc)));
+#endif
+		}
+		else
+		{
+			pg_log_error("pg_ctl exited with unrecognized status %d", rc);
+		}
+
+		pg_log_error_detail("The failed command was: %s", pg_ctl_cmd);
+		exit(1);
+	}
+}
+
+static void
+start_standby_server(struct CreateSubscriberOptions *opt, bool restricted_access)
+{
+	PQExpBuffer pg_ctl_cmd = createPQExpBuffer();
+	int			rc;
+
+	appendPQExpBuffer(pg_ctl_cmd, "\"%s\" start -D \"%s\" -s",
+					  pg_ctl_path, subscriber_dir);
+	if (restricted_access)
+	{
+		appendPQExpBuffer(pg_ctl_cmd, " -o \"-p %s\"", opt->sub_port);
+#if !defined(WIN32)
+
+		/*
+		 * An empty listen_addresses list means the server does not listen on
+		 * any IP interfaces; only Unix-domain sockets can be used to connect
+		 * to the server. Prevent external connections to minimize the chance
+		 * of failure.
+		 */
+		appendPQExpBufferStr(pg_ctl_cmd, " -o \"-c listen_addresses='' -c unix_socket_permissions=0700");
+		if (opt->socket_dir)
+			appendPQExpBuffer(pg_ctl_cmd, " -c unix_socket_directories='%s'",
+							  opt->socket_dir);
+		appendPQExpBufferChar(pg_ctl_cmd, '"');
+#endif
+	}
+	if (opt->config_file != NULL)
+		appendPQExpBuffer(pg_ctl_cmd, " -o \"-c config_file=%s\"",
+						  opt->config_file);
+	pg_log_debug("pg_ctl command is: %s", pg_ctl_cmd->data);
+	rc = system(pg_ctl_cmd->data);
+	pg_ctl_status(pg_ctl_cmd->data, rc);
+	standby_running = true;
+	destroyPQExpBuffer(pg_ctl_cmd);
+	pg_log_info("server was started");
+}
+
+static void
+stop_standby_server(const char *datadir)
+{
+	char	   *pg_ctl_cmd;
+	int			rc;
+
+	pg_ctl_cmd = psprintf("\"%s\" stop -D \"%s\" -s", pg_ctl_path,
+						  datadir);
+	pg_log_debug("pg_ctl command is: %s", pg_ctl_cmd);
+	rc = system(pg_ctl_cmd);
+	pg_ctl_status(pg_ctl_cmd, rc);
+	standby_running = false;
+	pg_log_info("server was stopped");
+}
+
+/*
+ * Returns after the server finishes the recovery process.
+ *
+ * If recovery_timeout option is set, terminate abnormally without finishing
+ * the recovery process. By default, it waits forever.
+ */
+static void
+wait_for_end_recovery(const char *conninfo, struct CreateSubscriberOptions *opt)
+{
+	PGconn	   *conn;
+	int			status = POSTMASTER_STILL_STARTING;
+	int			timer = 0;
+	int			count = 0;		/* number of consecutive connection attempts */
+
+#define NUM_CONN_ATTEMPTS	10
+
+	pg_log_info("waiting for the target server to reach the consistent state");
+
+	conn = connect_database(conninfo, true);
+
+	for (;;)
+	{
+		PGresult   *res;
+		bool		in_recovery = server_is_in_recovery(conn);
+
+		/*
+		 * Does the recovery process finish? In dry run mode, there is no
+		 * recovery mode. Bail out as the recovery process has ended.
+		 */
+		if (!in_recovery || dry_run)
+		{
+			status = POSTMASTER_READY;
+			recovery_ended = true;
+			break;
+		}
+
+		/*
+		 * If it is still in recovery, make sure the target server is
+		 * connected to the primary so it can receive the required WAL to
+		 * finish the recovery process. If it is disconnected try
+		 * NUM_CONN_ATTEMPTS in a row and bail out if not succeed.
+		 */
+		res = PQexec(conn,
+					 "SELECT 1 FROM pg_catalog.pg_stat_wal_receiver");
+		if (PQntuples(res) == 0)
+		{
+			if (++count > NUM_CONN_ATTEMPTS)
+			{
+				stop_standby_server(subscriber_dir);
+				pg_log_error("standby server disconnected from the primary");
+				break;
+			}
+		}
+		else
+			count = 0;			/* reset counter if it connects again */
+
+		PQclear(res);
+
+		/* Bail out after recovery_timeout seconds if this option is set */
+		if (opt->recovery_timeout > 0 && timer >= opt->recovery_timeout)
+		{
+			stop_standby_server(subscriber_dir);
+			pg_log_error("recovery timed out");
+			disconnect_database(conn, true);
+		}
+
+		/* Keep waiting */
+		pg_usleep(WAIT_INTERVAL * USEC_PER_SEC);
+
+		timer += WAIT_INTERVAL;
+	}
+
+	disconnect_database(conn, false);
+
+	if (status == POSTMASTER_STILL_STARTING)
+		pg_fatal("server did not end recovery");
+
+	pg_log_info("target server reached the consistent state");
+	pg_log_info_hint("If pg_createsubscriber fails after this point, "
+					 "you must recreate the physical replica before continuing.");
+}
+
+/*
+ * Create a publication that includes all tables in the database.
+ */
+static void
+create_publication(PGconn *conn, struct LogicalRepInfo *dbinfo)
+{
+	PQExpBuffer str = createPQExpBuffer();
+	PGresult   *res;
+
+	Assert(conn != NULL);
+
+	/* Check if the publication already exists */
+	appendPQExpBuffer(str,
+					  "SELECT 1 FROM pg_catalog.pg_publication "
+					  "WHERE pubname = '%s'",
+					  dbinfo->pubname);
+	res = PQexec(conn, str->data);
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		pg_log_error("could not obtain publication information: %s",
+					 PQresultErrorMessage(res));
+		disconnect_database(conn, true);
+	}
+
+	if (PQntuples(res) == 1)
+	{
+		/*
+		 * Unfortunately, if it reaches this code path, it will always fail
+		 * (unless you decide to change the existing publication name). That's
+		 * bad but it is very unlikely that the user will choose a name with
+		 * pg_createsubscriber_ prefix followed by the exact database oid and a
+		 * random number.
+		 */
+		pg_log_error("publication \"%s\" already exists", dbinfo->pubname);
+		pg_log_error_hint("Consider renaming this publication before continuing.");
+		disconnect_database(conn, true);
+	}
+
+	PQclear(res);
+	resetPQExpBuffer(str);
+
+	pg_log_info("creating publication \"%s\" on database \"%s\"",
+				dbinfo->pubname, dbinfo->dbname);
+
+	appendPQExpBuffer(str, "CREATE PUBLICATION %s FOR ALL TABLES",
+					  dbinfo->pubname);
+
+	pg_log_debug("command is: %s", str->data);
+
+	if (!dry_run)
+	{
+		res = PQexec(conn, str->data);
+		if (PQresultStatus(res) != PGRES_COMMAND_OK)
+		{
+			pg_log_error("could not create publication \"%s\" on database \"%s\": %s",
+						 dbinfo->pubname, dbinfo->dbname, PQresultErrorMessage(res));
+			disconnect_database(conn, true);
+		}
+		PQclear(res);
+	}
+
+	/* For cleanup purposes */
+	dbinfo->made_publication = true;
+
+	destroyPQExpBuffer(str);
+}
+
+/*
+ * Remove publication if it couldn't finish all steps.
+ */
+static void
+drop_publication(PGconn *conn, struct LogicalRepInfo *dbinfo)
+{
+	PQExpBuffer str = createPQExpBuffer();
+	PGresult   *res;
+
+	Assert(conn != NULL);
+
+	pg_log_info("dropping publication \"%s\" on database \"%s\"",
+				dbinfo->pubname, dbinfo->dbname);
+
+	appendPQExpBuffer(str, "DROP PUBLICATION %s", dbinfo->pubname);
+
+	pg_log_debug("command is: %s", str->data);
+
+	if (!dry_run)
+	{
+		res = PQexec(conn, str->data);
+		if (PQresultStatus(res) != PGRES_COMMAND_OK)
+		{
+			pg_log_error("could not drop publication \"%s\" on database \"%s\": %s",
+						 dbinfo->pubname, dbinfo->dbname, PQresultErrorMessage(res));
+			dbinfo->made_publication = false;	/* don't try again. */
+
+			/*
+			 * Don't disconnect and exit here. This routine is used by primary
+			 * (cleanup publication / replication slot due to an error) and
+			 * subscriber (remove the replicated publications). In both cases,
+			 * it can continue and provide instructions for the user to remove
+			 * it later if cleanup fails.
+			 */
+		}
+		PQclear(res);
+	}
+
+	destroyPQExpBuffer(str);
+}
+
+/*
+ * Create a subscription with some predefined options.
+ *
+ * A replication slot was already created in a previous step. Let's use it. By
+ * default, the subscription name is used as replication slot name. It is
+ * not required to copy data. The subscription will be created but it will not
+ * be enabled now. That's because the replication progress must be set and the
+ * replication origin name (one of the function arguments) contains the
+ * subscription OID in its name. Once the subscription is created,
+ * set_replication_progress() can obtain the chosen origin name and set up its
+ * initial location.
+ */
+static void
+create_subscription(PGconn *conn, struct LogicalRepInfo *dbinfo)
+{
+	PQExpBuffer str = createPQExpBuffer();
+	PGresult   *res;
+
+	Assert(conn != NULL);
+
+	pg_log_info("creating subscription \"%s\" on database \"%s\"",
+				dbinfo->subname, dbinfo->dbname);
+
+	appendPQExpBuffer(str,
+					  "CREATE SUBSCRIPTION %s CONNECTION '%s' PUBLICATION %s "
+					  "WITH (create_slot = false, copy_data = false, enabled = false)",
+					  dbinfo->subname, dbinfo->pubconninfo, dbinfo->pubname);
+
+	pg_log_debug("command is: %s", str->data);
+
+	if (!dry_run)
+	{
+		res = PQexec(conn, str->data);
+		if (PQresultStatus(res) != PGRES_COMMAND_OK)
+		{
+			pg_log_error("could not create subscription \"%s\" on database \"%s\": %s",
+						 dbinfo->subname, dbinfo->dbname, PQresultErrorMessage(res));
+			disconnect_database(conn, true);
+		}
+		PQclear(res);
+	}
+
+	destroyPQExpBuffer(str);
+}
+
+/*
+ * Sets the replication progress to the consistent LSN.
+ *
+ * The subscriber caught up to the consistent LSN provided by the last
+ * replication slot that was created. The goal is to set up the initial
+ * location for the logical replication that is the exact LSN that the
+ * subscriber was promoted. Once the subscription is enabled it will start
+ * streaming from that location onwards.  In dry run mode, the subscription OID
+ * and LSN are set to invalid values for printing purposes.
+ */
+static void
+set_replication_progress(PGconn *conn, struct LogicalRepInfo *dbinfo, const char *lsn)
+{
+	PQExpBuffer str = createPQExpBuffer();
+	PGresult   *res;
+	Oid			suboid;
+	char		originname[NAMEDATALEN];
+	char		lsnstr[17 + 1]; /* MAXPG_LSNLEN = 17 */
+
+	Assert(conn != NULL);
+
+	appendPQExpBuffer(str,
+					  "SELECT oid FROM pg_catalog.pg_subscription "
+					  "WHERE subname = '%s'",
+					  dbinfo->subname);
+
+	res = PQexec(conn, str->data);
+	if (PQresultStatus(res) != PGRES_TUPLES_OK)
+	{
+		pg_log_error("could not obtain subscription OID: %s",
+					 PQresultErrorMessage(res));
+		disconnect_database(conn, true);
+	}
+
+	if (PQntuples(res) != 1 && !dry_run)
+	{
+		pg_log_error("could not obtain subscription OID: got %d rows, expected %d rows",
+					 PQntuples(res), 1);
+		disconnect_database(conn, true);
+	}
+
+	if (dry_run)
+	{
+		suboid = InvalidOid;
+		snprintf(lsnstr, sizeof(lsnstr), "%X/%X",
+				 LSN_FORMAT_ARGS((XLogRecPtr) InvalidXLogRecPtr));
+	}
+	else
+	{
+		suboid = strtoul(PQgetvalue(res, 0, 0), NULL, 10);
+		snprintf(lsnstr, sizeof(lsnstr), "%s", lsn);
+	}
+
+	PQclear(res);
+
+	/*
+	 * The origin name is defined as pg_%u. %u is the subscription OID. See
+	 * ApplyWorkerMain().
+	 */
+	snprintf(originname, sizeof(originname), "pg_%u", suboid);
+
+	pg_log_info("setting the replication progress (node name \"%s\" ; LSN %s) on database \"%s\"",
+				originname, lsnstr, dbinfo->dbname);
+
+	resetPQExpBuffer(str);
+	appendPQExpBuffer(str,
+					  "SELECT pg_catalog.pg_replication_origin_advance('%s', '%s')",
+					  originname, lsnstr);
+
+	pg_log_debug("command is: %s", str->data);
+
+	if (!dry_run)
+	{
+		res = PQexec(conn, str->data);
+		if (PQresultStatus(res) != PGRES_TUPLES_OK)
+		{
+			pg_log_error("could not set replication progress for the subscription \"%s\": %s",
+						 dbinfo->subname, PQresultErrorMessage(res));
+			disconnect_database(conn, true);
+		}
+		PQclear(res);
+	}
+
+	destroyPQExpBuffer(str);
+}
+
+/*
+ * Enables the subscription.
+ *
+ * The subscription was created in a previous step but it was disabled. After
+ * adjusting the initial logical replication location, enable the subscription.
+ */
+static void
+enable_subscription(PGconn *conn, struct LogicalRepInfo *dbinfo)
+{
+	PQExpBuffer str = createPQExpBuffer();
+	PGresult   *res;
+
+	Assert(conn != NULL);
+
+	pg_log_info("enabling subscription \"%s\" on database \"%s\"",
+				dbinfo->subname, dbinfo->dbname);
+
+	appendPQExpBuffer(str, "ALTER SUBSCRIPTION %s ENABLE", dbinfo->subname);
+
+	pg_log_debug("command is: %s", str->data);
+
+	if (!dry_run)
+	{
+		res = PQexec(conn, str->data);
+		if (PQresultStatus(res) != PGRES_COMMAND_OK)
+		{
+			pg_log_error("could not enable subscription \"%s\": %s",
+						 dbinfo->subname, PQresultErrorMessage(res));
+			disconnect_database(conn, true);
+		}
+
+		PQclear(res);
+	}
+
+	destroyPQExpBuffer(str);
+}
+
+int
+main(int argc, char **argv)
+{
+	static struct option long_options[] =
+	{
+		{"config-file", required_argument, NULL, 1},
+		{"database", required_argument, NULL, 'd'},
+		{"pgdata", required_argument, NULL, 'D'},
+		{"dry-run", no_argument, NULL, 'n'},
+		{"subscriber-port", required_argument, NULL, 'p'},
+		{"publisher-server", required_argument, NULL, 'P'},
+		{"socket-directory", required_argument, NULL, 's'},
+		{"recovery-timeout", required_argument, NULL, 't'},
+		{"subscriber-username", required_argument, NULL, 'U'},
+		{"verbose", no_argument, NULL, 'v'},
+		{"version", no_argument, NULL, 'V'},
+		{"help", no_argument, NULL, '?'},
+		{NULL, 0, NULL, 0}
+	};
+
+	struct CreateSubscriberOptions opt = {0};
+
+	int			c;
+	int			option_index;
+
+	char	   *pub_base_conninfo;
+	char	   *sub_base_conninfo;
+	char	   *dbname_conninfo = NULL;
+
+	uint64		pub_sysid;
+	uint64		sub_sysid;
+	struct stat statbuf;
+
+	char	   *consistent_lsn;
+
+	char		pidfile[MAXPGPATH];
+
+	pg_logging_init(argv[0]);
+	pg_logging_set_level(PG_LOG_WARNING);
+	progname = get_progname(argv[0]);
+	set_pglocale_pgservice(argv[0], PG_TEXTDOMAIN("pg_createsubscriber"));
+
+	if (argc > 1)
+	{
+		if (strcmp(argv[1], "--help") == 0 || strcmp(argv[1], "-?") == 0)
+		{
+			usage();
+			exit(0);
+		}
+		else if (strcmp(argv[1], "-V") == 0
+				 || strcmp(argv[1], "--version") == 0)
+		{
+			puts("pg_createsubscriber (PostgreSQL) " PG_VERSION);
+			exit(0);
+		}
+	}
+
+	/* Default settings */
+	subscriber_dir = NULL;
+	opt.config_file = NULL;
+	opt.pub_conninfo_str = NULL;
+	opt.socket_dir = NULL;
+	opt.sub_port = palloc(16);
+	strcpy(opt.sub_port, DEFAULT_SUB_PORT);
+	opt.sub_username = NULL;
+	opt.database_names = (SimpleStringList)
+	{
+		NULL, NULL
+	};
+	opt.recovery_timeout = 0;
+
+	/*
+	 * Don't allow it to be run as root. It uses pg_ctl which does not allow
+	 * it either.
+	 */
+#ifndef WIN32
+	if (geteuid() == 0)
+	{
+		pg_log_error("cannot be executed by \"root\"");
+		pg_log_error_hint("You must run %s as the PostgreSQL superuser.",
+						  progname);
+		exit(1);
+	}
+#endif
+
+	get_restricted_token();
+
+	while ((c = getopt_long(argc, argv, "d:D:np:P:s:t:U:v",
+							long_options, &option_index)) != -1)
+	{
+		switch (c)
+		{
+			case 'd':
+				/* Ignore duplicated database names */
+				if (!simple_string_list_member(&opt.database_names, optarg))
+				{
+					simple_string_list_append(&opt.database_names, optarg);
+					num_dbs++;
+				}
+				break;
+			case 'D':
+				subscriber_dir = pg_strdup(optarg);
+				canonicalize_path(subscriber_dir);
+				break;
+			case 'n':
+				dry_run = true;
+				break;
+			case 'p':
+				pg_free(opt.sub_port);
+				opt.sub_port = pg_strdup(optarg);
+				break;
+			case 'P':
+				opt.pub_conninfo_str = pg_strdup(optarg);
+				break;
+			case 's':
+				opt.socket_dir = pg_strdup(optarg);
+				canonicalize_path(opt.socket_dir);
+				break;
+			case 't':
+				opt.recovery_timeout = atoi(optarg);
+				break;
+			case 'U':
+				opt.sub_username = pg_strdup(optarg);
+				break;
+			case 'v':
+				pg_logging_increase_verbosity();
+				break;
+			case 1:
+				opt.config_file = pg_strdup(optarg);
+				break;
+			default:
+				/* getopt_long already emitted a complaint */
+				pg_log_error_hint("Try \"%s --help\" for more information.", progname);
+				exit(1);
+		}
+	}
+
+	/* Any non-option arguments? */
+	if (optind < argc)
+	{
+		pg_log_error("too many command-line arguments (first is \"%s\")",
+					 argv[optind]);
+		pg_log_error_hint("Try \"%s --help\" for more information.", progname);
+		exit(1);
+	}
+
+	/* Required arguments */
+	if (subscriber_dir == NULL)
+	{
+		pg_log_error("no subscriber data directory specified");
+		pg_log_error_hint("Try \"%s --help\" for more information.", progname);
+		exit(1);
+	}
+
+	/* If socket directory is not provided, use the current directory */
+	if (opt.socket_dir == NULL)
+	{
+		char		cwd[MAXPGPATH];
+
+		if (!getcwd(cwd, MAXPGPATH))
+			pg_fatal("could not determine current directory");
+		opt.socket_dir = pg_strdup(cwd);
+		canonicalize_path(opt.socket_dir);
+	}
+
+	/*
+	 * Parse connection string. Build a base connection string that might be
+	 * reused by multiple databases.
+	 */
+	if (opt.pub_conninfo_str == NULL)
+	{
+		/*
+		 * TODO use primary_conninfo (if available) from subscriber and
+		 * extract publisher connection string. Assume that there are
+		 * identical entries for physical and logical replication. If there is
+		 * not, we would fail anyway.
+		 */
+		pg_log_error("no publisher connection string specified");
+		pg_log_error_hint("Try \"%s --help\" for more information.", progname);
+		exit(1);
+	}
+	pg_log_info("validating connection string on publisher");
+	pub_base_conninfo = get_base_conninfo(opt.pub_conninfo_str,
+										  &dbname_conninfo);
+	if (pub_base_conninfo == NULL)
+		exit(1);
+
+	pg_log_info("validating connection string on subscriber");
+	sub_base_conninfo = get_sub_conninfo(&opt);
+
+	if (opt.database_names.head == NULL)
+	{
+		pg_log_info("no database was specified");
+
+		/*
+		 * If --database option is not provided, try to obtain the dbname from
+		 * the publisher conninfo. If dbname parameter is not available, error
+		 * out.
+		 */
+		if (dbname_conninfo)
+		{
+			simple_string_list_append(&opt.database_names, dbname_conninfo);
+			num_dbs++;
+
+			pg_log_info("database \"%s\" was extracted from the publisher connection string",
+						dbname_conninfo);
+		}
+		else
+		{
+			pg_log_error("no database name specified");
+			pg_log_error_hint("Try \"%s --help\" for more information.",
+							  progname);
+			exit(1);
+		}
+	}
+
+	/* Get the absolute path of pg_ctl and pg_resetwal on the subscriber */
+	pg_ctl_path = get_exec_path(argv[0], "pg_ctl");
+	pg_resetwal_path = get_exec_path(argv[0], "pg_resetwal");
+
+	/* Rudimentary check for a data directory */
+	check_data_directory(subscriber_dir);
+
+	/*
+	 * Store database information for publisher and subscriber. It should be
+	 * called before atexit() because its return is used in the
+	 * cleanup_objects_atexit().
+	 */
+	dbinfo = store_pub_sub_info(opt.database_names, pub_base_conninfo,
+								sub_base_conninfo);
+
+	/* Register a function to clean up objects in case of failure */
+	atexit(cleanup_objects_atexit);
+
+	/*
+	 * Check if the subscriber data directory has the same system identifier
+	 * than the publisher data directory.
+	 */
+	pub_sysid = get_primary_sysid(dbinfo[0].pubconninfo);
+	sub_sysid = get_standby_sysid(subscriber_dir);
+	if (pub_sysid != sub_sysid)
+		pg_fatal("subscriber data directory is not a copy of the source database cluster");
+
+	/* Subscriber PID file */
+	snprintf(pidfile, MAXPGPATH, "%s/postmaster.pid", subscriber_dir);
+
+	/*
+	 * If the standby server is running, stop it. Some parameters (that can
+	 * only be set at server start) are informed by command-line options.
+	 */
+	if (stat(pidfile, &statbuf) == 0)
+	{
+
+		pg_log_info("standby is up and running");
+		pg_log_info("stopping the server to start the transformation steps");
+		stop_standby_server(subscriber_dir);
+	}
+
+	/*
+	 * Start a short-lived standby server with temporary parameters (provided
+	 * by command-line options). The goal is to avoid connections during the
+	 * transformation steps.
+	 */
+	pg_log_info("starting the standby with command-line options");
+	start_standby_server(&opt, true);
+
+	/* Check if the standby server is ready for logical replication */
+	check_subscriber(dbinfo);
+
+	/*
+	 * Check if the primary server is ready for logical replication. This
+	 * routine checks if a replication slot is in use on primary so it relies
+	 * on check_subscriber() to obtain the primary_slot_name. That's why it is
+	 * called after it.
+	 */
+	check_publisher(dbinfo);
+
+	/*
+	 * Create the required objects for each database on publisher. This step
+	 * is here mainly because if we stop the standby we cannot verify if the
+	 * primary slot is in use. We could use an extra connection for it but it
+	 * doesn't seem worth.
+	 */
+	consistent_lsn = setup_publisher(dbinfo);
+
+	/* Write the required recovery parameters */
+	setup_recovery(dbinfo, subscriber_dir, consistent_lsn);
+
+	/*
+	 * Restart subscriber so the recovery parameters will take effect. Wait
+	 * until accepting connections.
+	 */
+	pg_log_info("stopping and starting the subscriber");
+	stop_standby_server(subscriber_dir);
+	start_standby_server(&opt, true);
+
+	/* Waiting the subscriber to be promoted */
+	wait_for_end_recovery(dbinfo[0].subconninfo, &opt);
+
+	/*
+	 * Create the subscription for each database on subscriber. It does not
+	 * enable it immediately because it needs to adjust the replication start
+	 * point to the LSN reported by setup_publisher().  It also cleans up
+	 * publications created by this tool and replication to the standby.
+	 */
+	setup_subscriber(dbinfo, consistent_lsn);
+
+	/* Remove primary_slot_name if it exists on primary */
+	drop_primary_replication_slot(dbinfo, primary_slot_name);
+
+	/* Stop the subscriber */
+	pg_log_info("stopping the subscriber");
+	stop_standby_server(subscriber_dir);
+
+	/* Change system identifier from subscriber */
+	modify_subscriber_sysid(&opt);
+
+	/*
+	 * In dry run mode, the server is restarted with the provided command-line
+	 * options so validation can be applied in the target server. In order to
+	 * preserve the initial state of the server (running), start it without
+	 * the command-line options.
+	 */
+	if (dry_run)
+		start_standby_server(&opt, false);
+
+	success = true;
+
+	pg_log_info("Done!");
+
+	return 0;
+}
diff --git a/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
new file mode 100644
index 0000000000..e1294f9307
--- /dev/null
+++ b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
@@ -0,0 +1,265 @@
+# Copyright (c) 2024, PostgreSQL Global Development Group
+
+#
+# Test using a standby server as the subscriber.
+
+use strict;
+use warnings FATAL => 'all';
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+program_help_ok('pg_createsubscriber');
+program_version_ok('pg_createsubscriber');
+program_options_handling_ok('pg_createsubscriber');
+
+my $datadir = PostgreSQL::Test::Utils::tempdir;
+
+#
+# Test mandatory options
+command_fails(['pg_createsubscriber'],
+	'no subscriber data directory specified');
+command_fails(
+	[ 'pg_createsubscriber', '--pgdata', $datadir ],
+	'no publisher connection string specified');
+command_fails(
+	[
+		'pg_createsubscriber', '--verbose',
+		'--pgdata', $datadir,
+		'--publisher-server', 'port=5432'
+	],
+	'no database name specified');
+
+# Set up node P as primary
+my $node_p = PostgreSQL::Test::Cluster->new('node_p');
+$node_p->init(allows_streaming => 'logical');
+$node_p->start;
+
+# Set up node F as about-to-fail node
+# Force it to initialize a new cluster instead of copying a
+# previously initdb'd cluster. New cluster has a different system identifier so
+# we can test if the target cluster is a copy of the source cluster.
+my $node_f = PostgreSQL::Test::Cluster->new('node_f');
+$node_f->init(force_initdb => 1, allows_streaming => 'logical');
+$node_f->start;
+
+# On node P
+# - create databases
+# - create test tables
+# - insert a row
+# - create a physical replication slot
+$node_p->safe_psql(
+	'postgres', q(
+	CREATE DATABASE pg1;
+	CREATE DATABASE pg2;
+));
+$node_p->safe_psql('pg1', 'CREATE TABLE tbl1 (a text)');
+$node_p->safe_psql('pg1', "INSERT INTO tbl1 VALUES('first row')");
+$node_p->safe_psql('pg2', 'CREATE TABLE tbl2 (a text)');
+my $slotname = 'physical_slot';
+$node_p->safe_psql('pg2',
+	"SELECT pg_create_physical_replication_slot('$slotname')");
+
+# Set up node S as standby linking to node P
+$node_p->backup('backup_1');
+my $node_s = PostgreSQL::Test::Cluster->new('node_s');
+$node_s->init_from_backup($node_p, 'backup_1', has_streaming => 1);
+$node_s->append_conf(
+	'postgresql.conf', qq[
+primary_slot_name = '$slotname'
+]);
+$node_s->set_standby_mode();
+$node_s->start;
+
+# Run pg_createsubscriber on about-to-fail node F
+command_fails(
+	[
+		'pg_createsubscriber', '--verbose',
+		'--pgdata', $node_f->data_dir,
+		'--publisher-server', $node_p->connstr('pg1'),
+		'--socket-directory', $node_f->host,
+		'--subscriber-port', $node_f->port,
+		'--database', 'pg1',
+		'--database', 'pg2'
+	],
+	'subscriber data directory is not a copy of the source database cluster');
+
+# Set up node C as standby linking to node S
+$node_s->backup('backup_2');
+my $node_c = PostgreSQL::Test::Cluster->new('node_c');
+$node_c->init_from_backup($node_s, 'backup_2', has_streaming => 1);
+$node_c->adjust_conf('postgresql.conf', 'primary_slot_name', undef);
+$node_c->set_standby_mode();
+$node_c->start;
+
+# Run pg_createsubscriber on node C (P -> S -> C)
+command_fails(
+	[
+		'pg_createsubscriber', '--verbose',
+		'--dry-run', '--pgdata',
+		$node_c->data_dir, '--publisher-server',
+		$node_s->connstr('pg1'),
+		'--socket-directory', $node_c->host,
+		'--subscriber-port', $node_c->port,
+		'--database', 'pg1',
+		'--database', 'pg2'
+	],
+	'primary server is in recovery');
+
+# Insert another row on node P and wait node S to catch up
+$node_p->safe_psql('pg1', "INSERT INTO tbl1 VALUES('second row')");
+$node_p->wait_for_replay_catchup($node_s);
+
+# Check some unmet conditions on node P
+$node_p->append_conf('postgresql.conf', q{
+wal_level = replica
+max_replication_slots = 1
+max_wal_senders = 1
+max_worker_processes = 2
+});
+$node_p->restart;
+command_fails(
+	[
+		'pg_createsubscriber', '--verbose',
+		'--dry-run', '--pgdata',
+		$node_s->data_dir, '--publisher-server',
+		$node_p->connstr('pg1'),
+		'--socket-directory', $node_s->host,
+		'--subscriber-port', $node_s->port,
+		'--database', 'pg1',
+		'--database', 'pg2'
+	],
+	'primary contains unmet conditions on node P');
+# Restore default settings here but only apply it after testing standby. Some
+# standby settings should not be a lower setting than on the primary.
+$node_p->append_conf('postgresql.conf', q{
+wal_level = logical
+max_replication_slots = 10
+max_wal_senders = 10
+max_worker_processes = 8
+});
+
+# Check some unmet conditions on node S
+$node_s->append_conf('postgresql.conf', q{
+max_replication_slots = 1
+max_logical_replication_workers = 1
+max_worker_processes = 2
+});
+$node_s->restart;
+command_fails(
+	[
+		'pg_createsubscriber', '--verbose',
+		'--dry-run', '--pgdata',
+		$node_s->data_dir, '--publisher-server',
+		$node_p->connstr('pg1'),
+		'--socket-directory', $node_s->host,
+		'--subscriber-port', $node_s->port,
+		'--database', 'pg1',
+		'--database', 'pg2'
+	],
+	'standby contains unmet conditions on node S');
+$node_s->append_conf('postgresql.conf', q{
+max_replication_slots = 10
+max_logical_replication_workers = 4
+max_worker_processes = 8
+});
+# Restore default settings on both servers
+$node_s->restart;
+$node_p->restart;
+
+# dry run mode on node S
+command_ok(
+	[
+		'pg_createsubscriber', '--verbose',
+		'--dry-run', '--pgdata',
+		$node_s->data_dir, '--publisher-server',
+		$node_p->connstr('pg1'),
+		'--socket-directory', $node_s->host,
+		'--subscriber-port', $node_s->port,
+		'--database', 'pg1',
+		'--database', 'pg2'
+	],
+	'run pg_createsubscriber --dry-run on node S');
+
+# Check if node S is still a standby
+is($node_s->safe_psql('postgres', 'SELECT pg_catalog.pg_is_in_recovery()'),
+	't', 'standby is in recovery');
+
+# pg_createsubscriber can run without --databases option
+command_ok(
+	[
+		'pg_createsubscriber', '--verbose',
+		'--dry-run', '--pgdata',
+		$node_s->data_dir, '--publisher-server',
+		$node_p->connstr('pg1'),
+		'--socket-directory', $node_s->host,
+		'--subscriber-port', $node_s->port
+	],
+	'run pg_createsubscriber without --databases');
+
+# Run pg_createsubscriber on node S
+command_ok(
+	[
+		'pg_createsubscriber', '--verbose',
+		'--verbose', '--pgdata',
+		$node_s->data_dir, '--publisher-server',
+		$node_p->connstr('pg1'),
+		'--socket-directory', $node_s->host,
+		'--subscriber-port', $node_s->port,
+		'--database', 'pg1',
+		'--database', 'pg2'
+	],
+	'run pg_createsubscriber on node S');
+
+# Confirm the physical replication slot has been removed
+my $result = $node_p->safe_psql('pg1',
+	"SELECT count(*) FROM pg_replication_slots WHERE slot_name = '$slotname'"
+);
+is($result, qq(0),
+	'the physical replication slot used as primary_slot_name has been removed'
+);
+
+# Insert rows on P
+$node_p->safe_psql('pg1', "INSERT INTO tbl1 VALUES('third row')");
+$node_p->safe_psql('pg2', "INSERT INTO tbl2 VALUES('row 1')");
+
+# PID sets to undefined because subscriber was stopped behind the scenes.
+# Start subscriber
+$node_s->{_pid} = undef;
+$node_s->start;
+
+# Get subscription names
+$result = $node_s->safe_psql(
+	'postgres', qq(
+	SELECT subname FROM pg_subscription WHERE subname ~ '^pg_createsubscriber_'
+));
+my @subnames = split("\n", $result);
+
+# Wait subscriber to catch up
+$node_s->wait_for_subscription_sync($node_p, $subnames[0]);
+$node_s->wait_for_subscription_sync($node_p, $subnames[1]);
+
+# Check result on database pg1
+$result = $node_s->safe_psql('pg1', 'SELECT * FROM tbl1');
+is( $result, qq(first row
+second row
+third row),
+	'logical replication works on database pg1');
+
+# Check result on database pg2
+$result = $node_s->safe_psql('pg2', 'SELECT * FROM tbl2');
+is($result, qq(row 1), 'logical replication works on database pg2');
+
+# Different system identifier?
+my $sysid_p = $node_p->safe_psql('postgres',
+	'SELECT system_identifier FROM pg_control_system()');
+my $sysid_s = $node_s->safe_psql('postgres',
+	'SELECT system_identifier FROM pg_control_system()');
+ok($sysid_p != $sysid_s, 'system identifier was changed');
+
+# clean up
+$node_p->teardown_node;
+$node_s->teardown_node;
+$node_f->teardown_node;
+
+done_testing();
-- 
2.34.1

---

From: "Euler Taveira" <[email-masked]>
Date: Sat, 16 Mar 2024 12:42:51 -0300

> replication objects once the standby is converted to subscriber.

Yes. I was wondering to implement after v1 is pushed. I started to write a code
for it but I wasn't sure about the UI. The best approach I came up with was
multiple options in the same order. (I don't provide short options to avoid
possible portability issues with the order.) It means if I have 3 databases and
the following command-line:

pg_createsubscriber ... --database pg1 --database pg2 --database3 --publication
pubx --publication puby --publication pubz

pubx, puby and pubz are created in the database pg1, pg2, and pg3 respectively.

> an opinion on this point?

If it happens after creating a publication and before promotion, the cleanup
routine will drop the publications on primary and it will eventually be applied
to the standby via replication later.

> unless it is already explained.

I rewrote the sentence to make it clear that only if the server is promoted,
the target server will be in a state that cannot be reused. It provides a
message saying it too.

pg_createsubscriber: target server reached the consistent state
pg_createsubscriber: hint: If pg_createsubscriber fails after this point, you
must recreate the physical replica before continuing.

I'm attaching a new version (v30) that adds:

* 3 new options (--publication, --subscription, --replication-slot) to assign
  names to the objects. The --database option used to ignore duplicate names,
  however, since these new options rely on the number of database options to
  match the number of object name options, it is forbidden from now on. The
  duplication is also forbidden for the object names to avoid errors earlier.
* rewrite the paragraph related to unusuable target server after
  pg_createsubscriber fails.
* Vignesh reported an issue [1] related to reaching the recovery stop point
  before the consistent state is reached. I proposed a simple patch that fixes
  the issue.

[1] https://www.postgresql.org/message-id/CALDaNm3VMOi0GugGvhk3motghaFRKSWMCSE2t3YX1e%2BMttToxg%40mail.gmail.com

---

From: "Euler Taveira" <[email-masked]>
Date: Sat, 16 Mar 2024 12:46:07 -0300

> FATAL:  requested recovery stop point is before consistent recovery point

Thanks for checking. I proposed an alternative patch for it [1]. Can you check
it?

[1] https://www.postgresql.org/message-id/34637e7f-0330-420d-8f45-1d022962d2fe%40app.fastmail.com

---

From: vignesh C <[email-masked]>
Date: Sat, 16 Mar 2024 22:38:30 +0530

> Thanks for checking. I proposed an alternative patch for it [1]. Can you check

This approach looks good to me.

Regards,
Vignesh

---

From: Amit Kapila <[email-masked]>
Date: Mon, 18 Mar 2024 09:37:27 +0530

> pubx, puby and pubz are created in the database pg1, pg2, and pg3 respectively.

With this syntax, you want to give the user the option to specify
publication/subscription/slot name for each database? If so, won't it
be too much verbose?

> to the standby via replication later.

Do you mean to say that the next time if user uses
pg_createsubscriber, it will be ensured that all the prior WAL will be
replicated? I think we need to ensure that after the restart of this
tool and before it attempts to create the publications again, the WAL
of the previous drop has to be replayed.

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Mon, 18 Mar 2024 10:35:05 +0530

> of the previous drop has to be replayed.

On further thinking, the WAL for such dropped publications should get
replayed eventually before the WAL for new publications (the
publications which will be created after restart) unless the required
WAL is removed on primary due to some reason.

-- 
With Regards,
Amit Kapila.

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Mon, 18 Mar 2024 05:43:59 +0000

Dear Euler,

Thanks for updating the patch. Here are my comments.
I used Grammarly to proofread sentences.
(The tool strongly recommends to use active voice, but I can ignore for now)

01.

"After a successful run, the state of the target server is analagous to a fresh
logical replication setup."
a/analagous/analogous

02.

"The main difference between the logical replication setup and pg_createsubscriber
is the initial data copy."

Grammarly suggests:
"The initial data copy is the main difference between the logical replication
setup and pg_createsubscriber."

03.

"Only the synchronization phase is done, which ensures each table is brought up
to a synchronized state."

This sentence is not very clear to me. How about:
"pg_createsubscriber does only the synchronization phase, ensuring each table's
replication state is ready."

04.

"The pg_createsubscriber targets large database systems because most of the
execution time is spent making the initial data copy."

Hmm, but initial data sync by logical replication also spends most of time to
make the initial data copy. IIUC bottlenecks are a) this application must stop
and start server several times, and b) only the serial copy works. Can you
clarify them?

05.

It is better to say the internal difference between pg_createsubscriber and the
initial sync by logical replication. For example:
pg_createsubscriber uses a physical replication mechanism to ensure the standby
catches up until a certain point. Then, it converts to the standby to the
subscriber by promoting and creating subscriptions.

06.

"If these are not met an error will be reported."

Grammarly suggests:
"If these are not met, an error will be reported."

07.

"The given target data directory must have the same system identifier than the
source data directory."

Grammarly suggests:
"The given target data directory must have the same system identifier as the
source data directory."

08.

"If a standby server is running on the target data directory or it is a base
backup from the source data directory, system identifiers are the same."

This line is not needed if bullet-style is not used. The line is just a supplement,
not prerequisite.

09.

"The source server must accept connections from the target server. The source server must not be in recovery."

Grammarly suggests:
"The source server must accept connections from the target server and not be in recovery."

10.

"Publications cannot be created in a read-only cluster."

Same as 08, this line is not needed if bullet-style is not used.

11.

"pg_createsubscriber usually starts the target server with different connection
settings during the transformation steps. Hence, connections to target server
might fail."
 
Grammarly suggests:
"pg_createsubscriber usually starts the target server with different connection
settings during transformation. Hence, connections to the target server might fail."

12.

"During the recovery process,"

Grammarly suggests:
"During recovery,"

13.

"replicated so an error would occur."

Grammarly suggests:
"replicated, so an error would occur."

14.

"It would avoid situations in which WAL files from the source server might be
used by the target server."

Grammarly suggests:
"It would avoid situations in which the target server might use WAL files from
the source server."

15.

"a LSN"

s/a/an

16.

"of write-ahead"

s/of/of the/

17.

"specifies promote"

We can do double-quote for the word promote.

18.

"are also added so it avoids"

Grammarly suggests:
"are added to avoid"

19.

"is accepting read-write transactions"

Grammarly suggests:
"accepts read-write transactions"

20.

New options must be also documented as well. This helps not only users but also
reviewers.
(Sometimes we cannot identify that the implementation is intentinal or not.)

21.

Also, not sure the specification is good. I preferred to specify them by format
string. Because it can reduce the number of arguments and I cannot find use cases
which user want to control the name of objects.

However, your approach has a benefit which users can easily identify the generated
objects by pg_createsubscriber. How do other think?

22.

```
#define	BASE_OUTPUT_DIR		"pg_createsubscriber_output.d"
```

No one refers the define.

23.
 
```
}			CreateSubscriberOptions;
...
}			LogicalRepInfo;
```

Declarations after the "{" are not needed, because we do not do typedef.

22.

While seeing definitions of functions, I found that some pointers are declared
as const, but others are not. E.g., "char *lsn" in setup_recovery() won' be
changed but not the constant. Is it just missing or is there another rule?

23.

```
static int	num_dbs = 0;
static int	num_pubs = 0;
static int	num_subs = 0;
static int	num_replslots = 0;
```

I think the name is bit confusing. The number of generating publications/subscriptions/replication slots
are always same as the number of databases. They just indicate the number of
specified.

My idea is num_custom_pubs or something. Thought?

24.

```
/* standby / subscriber data directory */
static char *subscriber_dir = NULL;
```

It is bit strange that only subscriber_dir is a global variable. Caller requires
the CreateSubscriberOptions as an argument, except cleanup_objects_atexit() and
main. So, how about makeing `CreateSubscriberOptions opt` to global one?

25.

```
 * Replication slots, publications and subscriptions are created. Depending on
 * the step it failed, it should remove the already created objects if it is
 * possible (sometimes it won't work due to a connection issue).
```

I think it should be specified here that subscriptions won't be removed with the
reason. 

26.

```

	/*
	 * If the server is promoted, there is no way to use the current setup
	 * again. Warn the user that a new replication setup should be done before
	 * trying again.
	 */
```

Per comment 25, we can add a reference like "See comments atop the function"

27.

usage() was not updated based on recent changes.

28.

```
		if (strcmp(conn_opt->keyword, "dbname") == 0 && conn_opt->val != NULL)
		{
			if (dbname)
				*dbname = pg_strdup(conn_opt->val);
			continue;
		}
```

There is a memory-leak if multiple dbname are specified in the conninfo.

29.

```
	pg_prng_seed(&prng_state, (uint64) (getpid() ^ time(NULL)));
```

No need to initialize the seed every time. Can you reuse pg_prng_state?

30.

```
		if (num_replslots == 0)
			dbinfo[i].replslotname = pg_strdup(genname);
```

I think the straightforward way is to use the name of subscription if no name
is specified. This follows the rule for CREATE SUBSCRIPTION.

31.

```
		/* Create replication slot on publisher */
		if (lsn)
			pg_free(lsn);
```

I think allocating/freeing memory is not so efficient.
Can we add a flag to create_logical_replication_slot() for controlling the
returning value (NULL or duplicated string)? We can use the condition (i == num_dbs-1)
as flag.

32.

```
/*
 * Close the connection. If exit_on_error is true, it has an undesired
 * condition and it should exit immediately.
 */
static void
disconnect_database(PGconn *conn, bool exit_on_error)
```

In case of disconnect_database(), the second argument should have different name.
If it is true, the process exits unconditionally.
Also, comments atop the function must be fixed.

33.

```
	wal_level = strdup(PQgetvalue(res, 0, 0));
```

pg_strdup should be used here.

34.

```
		{"config-file", required_argument, NULL, 1},
		{"publication", required_argument, NULL, 2},
		{"replication-slot", required_argument, NULL, 3},
		{"subscription", required_argument, NULL, 4},
```

The ordering looks strange for me. According to pg_upgarade and pg_basebackup,
options which do not have short notation are listed behind.

35.

```
	opt.sub_port = palloc(16);
```

Per other lines, pg_alloc() should be used.

36.

```
				pg_free(opt.sub_port);
```

You said that the leak won't be concerned here. If so, why only 'p' has pg_free()?

37.

```
	/* Register a function to clean up objects in case of failure */
	atexit(cleanup_objects_atexit);
```

Sorry if we have already discussed. I think the registration can be moved just
before the boot of the standby. Before that, the callback will be no-op.

38.

```
	/* Subscriber PID file */
	snprintf(pidfile, MAXPGPATH, "%s/postmaster.pid", subscriber_dir);

	/*
	 * If the standby server is running, stop it. Some parameters (that can
	 * only be set at server start) are informed by command-line options.
	 */
	if (stat(pidfile, &statbuf) == 0)
```

Hmm. pidfile is used only here, but it is declared in main(). Can it be
separated into another funtion like is_standby_started()?

39.

Or, we may able to introcue "restart_standby_if_needed" or something.

40.

```
	 * XXX this code was extracted from BootStrapXLOG().
```

So, can we extract the common part to somewhere? Since system identifier is related
with the controldata file, I think it can be located in controldata_util.c.

41.

You said like below in [1], but I could not find the related fix. Can you clarify?

> That's a good point. We should state in the documentation that GUCs specified in
> the command-line options are ignored during the execution.

[1]: https://www.postgresql.org/message-id/40595e73-c7e1-463a-b8be-49792e870007%40app.fastmail.com

Best Regards,
Hayato Kuroda
FUJITSU LIMITED

---

From: vignesh C <[email-masked]>
Date: Mon, 18 Mar 2024 12:48:28 +0530

>   pg_createsubscriber fails.

1) Maximum size of the object name is 64, we can have a check so that
we don't specify more than the maximum allowed length:
+ case 3:
+ if (!simple_string_list_member(&opt.replslot_names, optarg))
+ {
+ simple_string_list_append(&opt.replslot_names, optarg);
+ num_replslots++;
+ }
+ else
+ {
+ pg_log_error("duplicate replication slot \"%s\"", optarg);
+ exit(1);
+ }
+ break;

If we allow something like this:
 ./pg_createsubscriber -U postgres -D data_N2/ -P "port=5431
user=postgres"  -p 5432 -s /home/vignesh/postgres/inst/bin/ -d db1 -d
db2 -d db3 --replication-slot="testtesttesttesttesttesttesttesttesttesttesttesttesttesttesttes1"
--replication-slot="testtesttesttesttesttesttesttesttesttesttesttesttesttesttesttes2"
--replication-slot="testtesttesttesttesttesttesttesttesttesttesttesttesttesttesttes3"
In this case creation of replication slot will fail:
pg_createsubscriber: error: could not create replication slot
"testtesttesttesttesttesttesttesttesttesttesttesttesttesttesttes" on
database "db2": ERROR:  replication slot
"testtesttesttesttesttesttesttesttesttesttesttesttesttesttesttes"
already exists

2) Similarly here too:
+ case 4:
+ if (!simple_string_list_member(&opt.sub_names, optarg))
+ {
+ simple_string_list_append(&opt.sub_names, optarg);
+ num_subs++;
+ }
+ else
+ {
+ pg_log_error("duplicate subscription \"%s\"", optarg);
+ exit(1);
+ }
+ break;

If we allow something like this:
./pg_createsubscriber -U postgres -D data_N2/ -P "port=5431
user=postgres"  -p 5432 -s /home/vignesh/postgres/inst/bin/ -d db1 -d
db2 -d db3 --subscription=testtesttesttesttesttesttesttesttesttesttesttesttesttesttesttes1
--subscription=testtesttesttesttesttesttesttesttesttesttesttesttesttesttesttes2
--subscription=testtesttesttesttesttesttesttesttesttesttesttesttesttesttesttes3

Subscriptions will be created with the same name and later there will
be a problem when setting replication progress as there will be
multiple subscriptions with the same name.

Regards,
Vignesh

---

From: Peter Eisentraut <[email-masked]>
Date: Mon, 18 Mar 2024 12:06:30 +0100

> "testtesttesttesttesttesttesttesttesttesttesttesttesttesttesttes"

I think this is fine.  The server can check whether the names it is 
given are of the right size.  We don't need to check it again in the client.

> multiple subscriptions with the same name.

Could you clarify this problem?

---

From: Peter Eisentraut <[email-masked]>
Date: Mon, 18 Mar 2024 12:15:16 +0100

> setup and pg_createsubscriber."

I think that change is worse.

> "The source server must accept connections from the target server and not be in recovery."

I think the previous version is better.

> We can do double-quote for the word promote.

The v30 patch has <literal>promote</literal>, which I think is adequate.

> "accepts read-write transactions"

I like the first one better.

> (Sometimes we cannot identify that the implementation is intentinal or not.)

Which ones are missing?

> objects by pg_createsubscriber. How do other think?

I think listing them explicitly is better for the first version.  It's 
simpler to implement and more flexible.

> No one refers the define.

This is gone in v30.

> Declarations after the "{" are not needed, because we do not do typedef.

Yeah, this is actually wrong, because as it is written now, it defines 
global variables.

> changed but not the constant. Is it just missing or is there another rule?

Yes, more could be done here.  I have attached a patch for this.  (This 
also requires the just-committed 48018f1d8c.)

> main. So, how about makeing `CreateSubscriberOptions opt` to global one?

Fewer global variables seem preferable.  Only make global as needed.

> is specified. This follows the rule for CREATE SUBSCRIPTION.

right

> returning value (NULL or duplicated string)? We can use the condition (i == num_dbs-1)

Nothing is even using the return value of 
create_logical_replication_slot().  I think this can be removed altogether.

> Per other lines, pg_alloc() should be used.

Even better psprintf().

> before the boot of the standby. Before that, the callback will be no-op.

But it can also stay where it is.  What is the advantage of moving it later?

> with the controldata file, I think it can be located in controldata_util.c.

Let's leave it as is for this PG release.
From d951a9f186b2162aa241f7554908b236c718154f Mon Sep 17 00:00:00 2001
From: Peter Eisentraut <[email-masked]>
Date: Mon, 18 Mar 2024 11:54:03 +0100
Subject: [PATCH] fixup! pg_createsubscriber: creates a new logical replica
 from a standby server

Add const decorations.
---
 src/bin/pg_basebackup/pg_createsubscriber.c | 54 ++++++++++-----------
 1 file changed, 27 insertions(+), 27 deletions(-)

diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index 6cc1c341214..e24ed7ef506 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -61,46 +61,46 @@ struct LogicalRepInfo
 
 static void cleanup_objects_atexit(void);
 static void usage();
-static char *get_base_conninfo(char *conninfo, char **dbname);
-static char *get_sub_conninfo(struct CreateSubscriberOptions *opt);
+static char *get_base_conninfo(const char *conninfo, char **dbname);
+static char *get_sub_conninfo(const struct CreateSubscriberOptions *opt);
 static char *get_exec_path(const char *argv0, const char *progname);
 static void check_data_directory(const char *datadir);
 static char *concat_conninfo_dbname(const char *conninfo, const char *dbname);
-static struct LogicalRepInfo *store_pub_sub_info(struct CreateSubscriberOptions *opt,
+static struct LogicalRepInfo *store_pub_sub_info(const struct CreateSubscriberOptions *opt,
 												 const char *pub_base_conninfo,
 												 const char *sub_base_conninfo);
 static PGconn *connect_database(const char *conninfo, bool exit_on_error);
 static void disconnect_database(PGconn *conn, bool exit_on_error);
 static uint64 get_primary_sysid(const char *conninfo);
 static uint64 get_standby_sysid(const char *datadir);
-static void modify_subscriber_sysid(struct CreateSubscriberOptions *opt);
+static void modify_subscriber_sysid(const struct CreateSubscriberOptions *opt);
 static bool server_is_in_recovery(PGconn *conn);
 static char *generate_object_name(PGconn *conn);
-static void check_publisher(struct LogicalRepInfo *dbinfo);
+static void check_publisher(const struct LogicalRepInfo *dbinfo);
 static char *setup_publisher(struct LogicalRepInfo *dbinfo);
-static void check_subscriber(struct LogicalRepInfo *dbinfo);
+static void check_subscriber(const struct LogicalRepInfo *dbinfo);
 static void setup_subscriber(struct LogicalRepInfo *dbinfo,
 							 const char *consistent_lsn);
-static void setup_recovery(struct LogicalRepInfo *dbinfo, char *datadir,
-						   char *lsn);
+static void setup_recovery(const struct LogicalRepInfo *dbinfo, const char *datadir,
+						   const char *lsn);
 static void drop_primary_replication_slot(struct LogicalRepInfo *dbinfo,
-										  char *slotname);
+										  const char *slotname);
 static char *create_logical_replication_slot(PGconn *conn,
 											 struct LogicalRepInfo *dbinfo);
 static void drop_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo,
 								  const char *slot_name);
 static void pg_ctl_status(const char *pg_ctl_cmd, int rc);
-static void start_standby_server(struct CreateSubscriberOptions *opt,
+static void start_standby_server(const struct CreateSubscriberOptions *opt,
 								 bool restricted_access);
 static void stop_standby_server(const char *datadir);
 static void wait_for_end_recovery(const char *conninfo,
-								  struct CreateSubscriberOptions *opt);
+								  const struct CreateSubscriberOptions *opt);
 static void create_publication(PGconn *conn, struct LogicalRepInfo *dbinfo);
 static void drop_publication(PGconn *conn, struct LogicalRepInfo *dbinfo);
-static void create_subscription(PGconn *conn, struct LogicalRepInfo *dbinfo);
-static void set_replication_progress(PGconn *conn, struct LogicalRepInfo *dbinfo,
+static void create_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo);
+static void set_replication_progress(PGconn *conn, const struct LogicalRepInfo *dbinfo,
 									 const char *lsn);
-static void enable_subscription(PGconn *conn, struct LogicalRepInfo *dbinfo);
+static void enable_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo);
 
 #define	USEC_PER_SEC	1000000
 #define	WAIT_INTERVAL	1		/* 1 second */
@@ -244,7 +244,7 @@ usage(void)
  * dbname.
  */
 static char *
-get_base_conninfo(char *conninfo, char **dbname)
+get_base_conninfo(const char *conninfo, char **dbname)
 {
 	PQExpBuffer buf = createPQExpBuffer();
 	PQconninfoOption *conn_opts = NULL;
@@ -292,7 +292,7 @@ get_base_conninfo(char *conninfo, char **dbname)
  * since it starts a server with restricted access.
  */
 static char *
-get_sub_conninfo(struct CreateSubscriberOptions *opt)
+get_sub_conninfo(const struct CreateSubscriberOptions *opt)
 {
 	PQExpBuffer buf = createPQExpBuffer();
 	char	   *ret;
@@ -411,7 +411,7 @@ concat_conninfo_dbname(const char *conninfo, const char *dbname)
  * setup_publisher().
  */
 static struct LogicalRepInfo *
-store_pub_sub_info(struct CreateSubscriberOptions *opt, const char *pub_base_conninfo,
+store_pub_sub_info(const struct CreateSubscriberOptions *opt, const char *pub_base_conninfo,
 				   const char *sub_base_conninfo)
 {
 	struct LogicalRepInfo *dbinfo;
@@ -603,7 +603,7 @@ get_standby_sysid(const char *datadir)
  * files from one of the systems might be used in the other one.
  */
 static void
-modify_subscriber_sysid(struct CreateSubscriberOptions *opt)
+modify_subscriber_sysid(const struct CreateSubscriberOptions *opt)
 {
 	ControlFileData *cf;
 	bool		crc_ok;
@@ -791,7 +791,7 @@ server_is_in_recovery(PGconn *conn)
  * XXX Does it not allow a synchronous replica?
  */
 static void
-check_publisher(struct LogicalRepInfo *dbinfo)
+check_publisher(const struct LogicalRepInfo *dbinfo)
 {
 	PGconn	   *conn;
 	PGresult   *res;
@@ -947,7 +947,7 @@ check_publisher(struct LogicalRepInfo *dbinfo)
  * will be broken at the end of this process (due to pg_resetwal).
  */
 static void
-check_subscriber(struct LogicalRepInfo *dbinfo)
+check_subscriber(const struct LogicalRepInfo *dbinfo)
 {
 	PGconn	   *conn;
 	PGresult   *res;
@@ -1124,7 +1124,7 @@ setup_subscriber(struct LogicalRepInfo *dbinfo, const char *consistent_lsn)
  * Write the required recovery parameters.
  */
 static void
-setup_recovery(struct LogicalRepInfo *dbinfo, char *datadir, char *lsn)
+setup_recovery(const struct LogicalRepInfo *dbinfo, const char *datadir, const char *lsn)
 {
 	PGconn	   *conn;
 	PQExpBuffer recoveryconfcontents;
@@ -1184,7 +1184,7 @@ setup_recovery(struct LogicalRepInfo *dbinfo, char *datadir, char *lsn)
  * eventually drops this replication slot later.
  */
 static void
-drop_primary_replication_slot(struct LogicalRepInfo *dbinfo, char *slotname)
+drop_primary_replication_slot(struct LogicalRepInfo *dbinfo, const char *slotname)
 {
 	PGconn	   *conn;
 
@@ -1322,7 +1322,7 @@ pg_ctl_status(const char *pg_ctl_cmd, int rc)
 }
 
 static void
-start_standby_server(struct CreateSubscriberOptions *opt, bool restricted_access)
+start_standby_server(const struct CreateSubscriberOptions *opt, bool restricted_access)
 {
 	PQExpBuffer pg_ctl_cmd = createPQExpBuffer();
 	int			rc;
@@ -1380,7 +1380,7 @@ stop_standby_server(const char *datadir)
  * the recovery process. By default, it waits forever.
  */
 static void
-wait_for_end_recovery(const char *conninfo, struct CreateSubscriberOptions *opt)
+wait_for_end_recovery(const char *conninfo, const struct CreateSubscriberOptions *opt)
 {
 	PGconn	   *conn;
 	int			status = POSTMASTER_STILL_STARTING;
@@ -1575,7 +1575,7 @@ drop_publication(PGconn *conn, struct LogicalRepInfo *dbinfo)
  * initial location.
  */
 static void
-create_subscription(PGconn *conn, struct LogicalRepInfo *dbinfo)
+create_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo)
 {
 	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res;
@@ -1620,7 +1620,7 @@ create_subscription(PGconn *conn, struct LogicalRepInfo *dbinfo)
  * and LSN are set to invalid values for printing purposes.
  */
 static void
-set_replication_progress(PGconn *conn, struct LogicalRepInfo *dbinfo, const char *lsn)
+set_replication_progress(PGconn *conn, const struct LogicalRepInfo *dbinfo, const char *lsn)
 {
 	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res;
@@ -1702,7 +1702,7 @@ set_replication_progress(PGconn *conn, struct LogicalRepInfo *dbinfo, const char
  * adjusting the initial logical replication location, enable the subscription.
  */
 static void
-enable_subscription(PGconn *conn, struct LogicalRepInfo *dbinfo)
+enable_subscription(PGconn *conn, const struct LogicalRepInfo *dbinfo)
 {
 	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res;

base-commit: 48018f1d8c12d42b53c2a855626ee1ceb7f4ca71
prerequisite-patch-id: e4c7223061ca1e12dfa84f4a5ccdc3a9ab0c268a
-- 
2.44.0

---

From: Peter Eisentraut <[email-masked]>
Date: Mon, 18 Mar 2024 14:52:28 +0100

> I'm attaching a new version (v30) that adds:

I have some review comments and attached a patch with some smaller 
fixups (mainly message wording and avoid fixed-size string buffers).

* doc/src/sgml/ref/pg_createsubscriber.sgml

I would remove the "How It Works" section.  This is not relevant to
users, and it is very detailed and will require updating whenever the
implementation changes.  It could be a source code comment instead.

* src/bin/pg_basebackup/pg_createsubscriber.c

I think the connection string handling is not robust against funny
characters, like spaces, in database names etc.

Most SQL commands need to be amended for proper identifier or string
literal quoting and/or escaping.

In check_subscriber(): All these permissions checks seem problematic
to me.  We shouldn't reimplement our own copy of the server's
permission checks.  The server can check the permissions.  And if the
permission checking in the server ever changes, then we have
inconsistencies to take care of.  Also, the error messages "permission
denied" are inappropriate, because we are not doing the actual thing.
Maybe we want to do a dry-run for the benefit of the user, but then we
should do the actual thing, like try to create a replication slot, or
whatever.  But I would rather just remove all this, it seems too
problematic.

In main(): The first check if the standby is running is problematic.
I think it would be better to require that the standby is initially
shut down.  Consider, the standby might be running under systemd.
This tool will try to stop it, systemd will try to restart it.  Let's
avoid these kinds of battles.  It's also safer if we don't try to
touch running servers.

The -p option (--subscriber-port) doesn't seem to do anything.  In my
testing, it always uses the compiled-in default port.

Printing all the server log lines to the terminal doesn't seem very
user-friendly.  Not sure what to do about that, short of keeping a 
pg_upgrade-style directory of log files.  But it's ugly.

From ec8e6ed6c3325a6f9fde2d1632346e212ade9c9f Mon Sep 17 00:00:00 2001
From: Peter Eisentraut <[email-masked]>
Date: Mon, 18 Mar 2024 14:48:55 +0100
Subject: [PATCH] Various improvements

---
 src/bin/pg_basebackup/Makefile              |  2 +-
 src/bin/pg_basebackup/nls.mk                |  1 +
 src/bin/pg_basebackup/pg_createsubscriber.c | 49 +++++++++------------
 3 files changed, 23 insertions(+), 29 deletions(-)

diff --git a/src/bin/pg_basebackup/Makefile b/src/bin/pg_basebackup/Makefile
index e9a920dbcda..26c53e473f5 100644
--- a/src/bin/pg_basebackup/Makefile
+++ b/src/bin/pg_basebackup/Makefile
@@ -49,7 +49,7 @@ all: pg_basebackup pg_createsubscriber pg_receivewal pg_recvlogical
 pg_basebackup: $(BBOBJS) $(OBJS) | submake-libpq submake-libpgport submake-libpgfeutils
 	$(CC) $(CFLAGS) $(BBOBJS) $(OBJS) $(LDFLAGS) $(LDFLAGS_EX) $(LIBS) -o $@$(X)
 
-pg_createsubscriber: $(WIN32RES) pg_createsubscriber.o | submake-libpq submake-libpgport submake-libpgfeutils
+pg_createsubscriber: pg_createsubscriber.o $(WIN32RES) | submake-libpq submake-libpgport submake-libpgfeutils
 	$(CC) $(CFLAGS) $^ $(LDFLAGS) $(LDFLAGS_EX) $(LIBS) -o $@$(X)
 
 pg_receivewal: pg_receivewal.o $(OBJS) | submake-libpq submake-libpgport submake-libpgfeutils
diff --git a/src/bin/pg_basebackup/nls.mk b/src/bin/pg_basebackup/nls.mk
index fc475003e8e..7870cea71ce 100644
--- a/src/bin/pg_basebackup/nls.mk
+++ b/src/bin/pg_basebackup/nls.mk
@@ -8,6 +8,7 @@ GETTEXT_FILES    = $(FRONTEND_COMMON_GETTEXT_FILES) \
                    bbstreamer_tar.c \
                    bbstreamer_zstd.c \
                    pg_basebackup.c \
+                   pg_createsubscriber.c \
                    pg_receivewal.c \
                    pg_recvlogical.c \
                    receivelog.c \
diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index e24ed7ef506..91c3a2f0036 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -43,7 +43,7 @@ struct CreateSubscriberOptions
 	SimpleStringList sub_names; /* list of subscription names */
 	SimpleStringList replslot_names;	/* list of replication slot names */
 	int			recovery_timeout;	/* stop recovery after this time */
-}			CreateSubscriberOptions;
+};
 
 struct LogicalRepInfo
 {
@@ -57,7 +57,7 @@ struct LogicalRepInfo
 
 	bool		made_replslot;	/* replication slot was created */
 	bool		made_publication;	/* publication was created */
-}			LogicalRepInfo;
+};
 
 static void cleanup_objects_atexit(void);
 static void usage();
@@ -155,9 +155,9 @@ cleanup_objects_atexit(void)
 	 */
 	if (recovery_ended)
 	{
-		pg_log_warning("pg_createsubscriber failed after the end of recovery");
-		pg_log_warning_hint("The target server cannot be used as a physical replica anymore.");
-		pg_log_warning_hint("You must recreate the physical replica before continuing.");
+		pg_log_warning("failed after the end of recovery");
+		pg_log_warning_hint("The target server cannot be used as a physical replica anymore.  "
+							"You must recreate the physical replica before continuing.");
 	}
 
 	for (int i = 0; i < num_dbs; i++)
@@ -184,13 +184,13 @@ cleanup_objects_atexit(void)
 				 */
 				if (dbinfo[i].made_publication)
 				{
-					pg_log_warning("There might be a publication \"%s\" in database \"%s\" on primary",
+					pg_log_warning("publication \"%s\" in database \"%s\" on primary might be left behind",
 								   dbinfo[i].pubname, dbinfo[i].dbname);
 					pg_log_warning_hint("Consider dropping this publication before trying again.");
 				}
 				if (dbinfo[i].made_replslot)
 				{
-					pg_log_warning("There might be a replication slot \"%s\" in database \"%s\" on primary",
+					pg_log_warning("replication slot \"%s\" in database \"%s\" on primary might be left behind",
 								   dbinfo[i].replslotname, dbinfo[i].dbname);
 					pg_log_warning_hint("Drop this replication slot soon to avoid retention of WAL files.");
 				}
@@ -420,7 +420,7 @@ store_pub_sub_info(const struct CreateSubscriberOptions *opt, const char *pub_ba
 	SimpleStringListCell *replslotcell = NULL;
 	int			i = 0;
 
-	dbinfo = (struct LogicalRepInfo *) pg_malloc(num_dbs * sizeof(struct LogicalRepInfo));
+	dbinfo = pg_malloc_array(struct LogicalRepInfo, num_dbs);
 
 	if (num_pubs > 0)
 		pubcell = opt->pub_names.head;
@@ -611,7 +611,7 @@ modify_subscriber_sysid(const struct CreateSubscriberOptions *opt)
 
 	char	   *cmd_str;
 
-	pg_log_info("modifying system identifier from subscriber");
+	pg_log_info("modifying system identifier of subscriber");
 
 	cf = get_controlfile(subscriber_dir, &crc_ok);
 	if (!crc_ok)
@@ -965,7 +965,7 @@ check_subscriber(const struct LogicalRepInfo *dbinfo)
 	/* The target server must be a standby */
 	if (!server_is_in_recovery(conn))
 	{
-		pg_log_error("The target server must be a standby");
+		pg_log_error("target server must be a standby");
 		disconnect_database(conn, true);
 	}
 
@@ -1217,13 +1217,11 @@ create_logical_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo)
 {
 	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res = NULL;
-	char		slot_name[NAMEDATALEN];
+	const char *slot_name = dbinfo->replslotname;
 	char	   *lsn = NULL;
 
 	Assert(conn != NULL);
 
-	snprintf(slot_name, NAMEDATALEN, "%s", dbinfo->replslotname);
-
 	pg_log_info("creating the replication slot \"%s\" on database \"%s\"",
 				slot_name, dbinfo->dbname);
 
@@ -1451,8 +1449,7 @@ wait_for_end_recovery(const char *conninfo, const struct CreateSubscriberOptions
 		pg_fatal("server did not end recovery");
 
 	pg_log_info("target server reached the consistent state");
-	pg_log_info_hint("If pg_createsubscriber fails after this point, "
-					 "you must recreate the physical replica before continuing.");
+	pg_log_info_hint("If pg_createsubscriber fails after this point, you must recreate the physical replica before continuing.");
 }
 
 /*
@@ -1625,8 +1622,8 @@ set_replication_progress(PGconn *conn, const struct LogicalRepInfo *dbinfo, cons
 	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res;
 	Oid			suboid;
-	char		originname[NAMEDATALEN];
-	char		lsnstr[17 + 1]; /* MAXPG_LSNLEN = 17 */
+	char	   *originname;
+	char	   *lsnstr;
 
 	Assert(conn != NULL);
 
@@ -1653,13 +1650,12 @@ set_replication_progress(PGconn *conn, const struct LogicalRepInfo *dbinfo, cons
 	if (dry_run)
 	{
 		suboid = InvalidOid;
-		snprintf(lsnstr, sizeof(lsnstr), "%X/%X",
-				 LSN_FORMAT_ARGS((XLogRecPtr) InvalidXLogRecPtr));
+		lsnstr = psprintf("%X/%X", LSN_FORMAT_ARGS((XLogRecPtr) InvalidXLogRecPtr));
 	}
 	else
 	{
 		suboid = strtoul(PQgetvalue(res, 0, 0), NULL, 10);
-		snprintf(lsnstr, sizeof(lsnstr), "%s", lsn);
+		lsnstr = psprintf("%s", lsn);
 	}
 
 	PQclear(res);
@@ -1668,7 +1664,7 @@ set_replication_progress(PGconn *conn, const struct LogicalRepInfo *dbinfo, cons
 	 * The origin name is defined as pg_%u. %u is the subscription OID. See
 	 * ApplyWorkerMain().
 	 */
-	snprintf(originname, sizeof(originname), "pg_%u", suboid);
+	originname = psprintf("pg_%u", suboid);
 
 	pg_log_info("setting the replication progress (node name \"%s\" ; LSN %s) on database \"%s\"",
 				originname, lsnstr, dbinfo->dbname);
@@ -1692,6 +1688,8 @@ set_replication_progress(PGconn *conn, const struct LogicalRepInfo *dbinfo, cons
 		PQclear(res);
 	}
 
+	pg_free(originname);
+	pg_free(lsnstr);
 	destroyPQExpBuffer(str);
 }
 
@@ -1797,13 +1795,9 @@ main(int argc, char **argv)
 	opt.config_file = NULL;
 	opt.pub_conninfo_str = NULL;
 	opt.socket_dir = NULL;
-	opt.sub_port = palloc(16);
-	strcpy(opt.sub_port, DEFAULT_SUB_PORT);
+	opt.sub_port = DEFAULT_SUB_PORT;
 	opt.sub_username = NULL;
-	opt.database_names = (SimpleStringList)
-	{
-		NULL, NULL
-	};
+	opt.database_names = (SimpleStringList){0};
 	opt.recovery_timeout = 0;
 
 	/*
@@ -1847,7 +1841,6 @@ main(int argc, char **argv)
 				dry_run = true;
 				break;
 			case 'p':
-				pg_free(opt.sub_port);
 				opt.sub_port = pg_strdup(optarg);
 				break;
 			case 'P':

base-commit: 48018f1d8c12d42b53c2a855626ee1ceb7f4ca71
prerequisite-patch-id: e4c7223061ca1e12dfa84f4a5ccdc3a9ab0c268a
prerequisite-patch-id: 6cc7583799a18d1119c8ecbb400b4471c6873768
-- 
2.44.0

---

From: Alvaro Herrera <[email-masked]>
Date: Mon, 18 Mar 2024 15:09:23 +0100

> characters, like spaces, in database names etc.

Maybe it would be easier to deal with this by passing around a struct
with keyword/value pairs that can be given to PQconnectdbParams (and
keeping dbname as a param that's passed separately, so that it can be
added when one is needed), instead of messing with the string conninfos;
then you don't have to worry about quoting.

-- 
Álvaro Herrera               48°01'N 7°57'E  —  https://www.EnterpriseDB.com/
"If you have nothing to say, maybe you need just the right tool to help you
not say it."                   (New York Times, about Microsoft PowerPoint)

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Tue, 19 Mar 2024 04:54:23 +0000

Dear Peter,

Thanks for giving comments. I want to reply some of them.

> The v30 patch has <literal>promote</literal>, which I think is adequate.

Opps. Actually I did look v29 patch while firstly reviewing. Sorry for noise.

> > (Sometimes we cannot identify that the implementation is intentinal or not.)

In v29, newly added options (publication/subscription/replication-slot) was not added.
Since they have been added, please ignore.

> simpler to implement and more flexible.

OK.

> > No one refers the define.

I wrote due to the above reason. Please ignore...

> But it can also stay where it is.  What is the advantage of moving it later?

I thought we could reduce the risk of bugs. Previously some bugs were reported
because the registration is too early. However, this is not a strong opinion.

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/

---

From: Amit Kapila <[email-masked]>
Date: Tue, 19 Mar 2024 12:35:17 +0530

> whatever.  But I would rather just remove all this, it seems too

If we remove all the checks then there is a possibility that we can
fail later while creating the actual subscription. For example, if
there are not sufficient max_replication_slots, then it is bound to
fail in the later steps which would be a costlier affair because by
that time the standby would have been promoted and the user won't have
any way to move forward but to re-create standby and then use this
tool again. I think here the patch tries to mimic pg_upgrade style
checks where we do some pre-checks.

This raises a question in my mind how are we expecting users to know
all these required values and configure it properly before using this
tool?  IIUC, we are expecting that the user should figure out the
appropriate values for max_replication_slots,
max_logical_replication_workers, etc. by querying the number of
databases in the primary. Then either stop the standby to change these
parameters or use ALTER SYSTEM depending on the required parameters.
Similarly, there are some config requirements (like max_wal_senders,
max_replication_slots) for the primary which would be difficult for
users to know as these are tools are internal requirements.

The two possibilities that come to my mind are (a) pg_createsubscriber
should have some special mode/option using which the user can find out
all the required config settings, or (b) document how a user can find
the required settings. There are good chances of mistakes with option
(b).

-- 
With Regards,
Amit Kapila.

---

From: Shlok Kyal <[email-masked]>
Date: Tue, 19 Mar 2024 16:56:10 +0530

Hi,

> [1] https://www.postgresql.org/message-id/CALDaNm3VMOi0GugGvhk3motghaFRKSWMCSE2t3YX1e%2BMttToxg%40mail.gmail.com

I have added a top-up patch v30-0003. The issue in [1] still exists in
the v30 patch. I was not able to come up with an approach to handle it
in the code, so I have added it to the documentation in the warning
section. Thoughts?
I am not changing the version as I have not made any changes in
v30-0001 and v30-0002.

[1]: https://www.postgresql.org/message-id/[email-masked]

Thanks and regards,
Shlok Kyal

---

From: Peter Eisentraut <[email-masked]>
Date: Tue, 19 Mar 2024 12:56:10 +0100

> checks where we do some pre-checks.

I think checking for required parameter settings is fine.  My concern is 
with the code before that, that does 
pg_has_role/has_database_privilege/has_function_privilege.

---

From: Peter Eisentraut <[email-masked]>
Date: Tue, 19 Mar 2024 12:57:36 +0100

> in the code, so I have added it to the documentation in the warning

Seems acceptable to me.  pg_createsubscriber will probably always have 
some restrictions and unsupported edge cases like that.  We can't 
support everything, so documenting is ok.

---

From: vignesh C <[email-masked]>
Date: Tue, 19 Mar 2024 20:54:19 +0530

> Could you clarify this problem?

In this case the subscriptions name specified is more than the allowed
name, the subscription name will be truncated and both the
subscription for db1 and db2 will have same name like below:
db2=# select subname, subdbid from pg_subscription;
                             subname                             | subdbid
-----------------------------------------------------------------+---------
 testtesttesttesttesttesttesttesttesttesttesttesttesttesttesttes |   16384
 testtesttesttesttesttesttesttesttesttesttesttesttesttesttesttes |   16385

Now we try to set the replication origin of the subscriptions to a
consistent lsn from the following:
+set_replication_progress(PGconn *conn, struct LogicalRepInfo *dbinfo,
const char *lsn)
+{
+       PQExpBuffer str = createPQExpBuffer();
+       PGresult   *res;
+       Oid                     suboid;
+       char            originname[NAMEDATALEN];
+       char            lsnstr[17 + 1]; /* MAXPG_LSNLEN = 17 */
+
+       Assert(conn != NULL);
+
+       appendPQExpBuffer(str,
+                                         "SELECT oid FROM
pg_catalog.pg_subscription "
+                                         "WHERE subname = '%s'",
+                                         dbinfo->subname);
+
+       res = PQexec(conn, str->data);
+       if (PQresultStatus(res) != PGRES_TUPLES_OK)
+       {
+               pg_log_error("could not obtain subscription OID: %s",
+                                        PQresultErrorMessage(res));
+               disconnect_database(conn, true);
+       }
+
+       if (PQntuples(res) != 1 && !dry_run)
+       {
+               pg_log_error("could not obtain subscription OID: got
%d rows, expected %d rows",
+                                        PQntuples(res), 1);
+               disconnect_database(conn, true);
+       }

Since the subscription name is truncated, we will have multiple
records returned for the above query which results in failure with:
pg_createsubscriber: error: could not obtain subscription OID: got 2
rows, expected 1 rows
pg_createsubscriber: warning: pg_createsubscriber failed after the end
of recovery
pg_createsubscriber: hint: The target server cannot be used as a
physical replica anymore.
pg_createsubscriber: hint: You must recreate the physical replica
before continuing.

The problem with this failure is that standby has been promoted
already and we will have to re-create the physica replica again.

If you are not planning to have the checks for name length, this could
alternatively be fixed by including database id also while querying
pg_subscription like below in set_replication_progress function:
appendPQExpBuffer(str,
"SELECT oid FROM pg_catalog.pg_subscription \n"
"WHERE subname = '%s' AND subdbid = (SELECT oid FROM
pg_catalog.pg_database WHERE datname = '%s')",
dbinfo->subname,
dbinfo->dbname);

I have verified this fixes the issue.

Regards,
Vignesh

---

From: Peter Eisentraut <[email-masked]>
Date: Tue, 19 Mar 2024 17:19:00 +0100

> pg_catalog.pg_database WHERE datname = '%s')",

Yes, this is more correct anyway, because subscription names are 
per-database, not global.  So you should be able to make 
pg_createsubscriber use the same subscription name for each database.

---

From: Shubham Khanna <[email-masked]>
Date: Wed, 20 Mar 2024 15:46:13 +0530

> db2=# select subname, subdbid from pg_subscription;

The attached patch has the changes to handle the same.

Thanks and Regards,
Shubham Khanna.

---

From: "Euler Taveira" <[email-masked]>
Date: Thu, 21 Mar 2024 01:19:10 -0300

> fixups (mainly message wording and avoid fixed-size string buffers).

Thanks for your review. I'm attaching a new version (v32) that includes your
fixups, merges the v30-0002 into the main patch [1], addresses Hayato's review[2],
your reviews [3][4], and fixes the query for set_replication_progress() [5].

> implementation changes.  It could be a source code comment instead.

It uses the same structure as pg_rewind that also describes how it works
internally. I included a separate patch that completely removes the section.

> characters, like spaces, in database names etc.

get_base_conninfo() uses PQconninfoParse to parse the connection string. I
expect PQconnectdb to provide a suitable error message in this case. Even if it
builds keywords and values arrays, it is also susceptible to the same issue, no?

> literal quoting and/or escaping.

I completely forgot about this detail when I added the new options in v30. It is
fixed now. I also changed the tests to exercise it.

> whatever.  But I would rather just remove all this, it seems too

The main goal of the check_* functions are to minimize error during execution.
I removed the permission checks. The GUC checks were kept.

> avoid these kinds of battles.  It's also safer if we don't try to

That's a good point. I hadn't found an excuse to simplify this but you provided
one. :) There was a worry about ignoring some command-line options that changes
GUCs if the server was started. There was also an ugly case for dry run mode
that has to start the server (if it was running) at the end. Both cases are no
longer issues. The current code provides a suitable error if the target server
is running.

> testing, it always uses the compiled-in default port.

It works for me. See this snippet from the regression tests. The port (50945) is
used by pg_ctl.

# Running: pg_createsubscriber --verbose --verbose --pgdata /c/pg_createsubscriber/src/bin/pg_basebackup/tmp_check/t_040_pg_createsubscriber_node_s_data/pgdata --publisher-server port=50943 host=/tmp/qpngb0bPKo dbname='pg1' --socket-directory /tmp/qpngb0bPKo --subscriber-port 50945 --database pg1 --database pg2
pg_createsubscriber: validating connection string on publisher
.
.
pg_createsubscriber: pg_ctl command is: "/c/pg_createsubscriber/tmp_install/c/pg_createsubscriber/bin/pg_ctl" start -D "/c/pg_createsubscriber/src/bin/pg_basebackup/tmp_check/t_040_pg_createsubscriber_node_s_data/pgdata" -s -o "-p 50945" -o "-c listen_addresses='' -c unix_socket_permissions=0700 -c unix_socket_directories='/tmp/qpngb0bPKo'"
2024-03-20 18:15:24.517 -03 [105195] LOG:  starting PostgreSQL 17devel on x86_64-pc-linux-gnu, compiled by gcc (Debian 10.2.1-6) 10.2.1 20210110, 64-bit
2024-03-20 18:15:24.517 -03 [105195] LOG:  listening on Unix socket "/tmp/qpngb0bPKo/.s.PGSQL.50945"

> pg_upgrade-style directory of log files.  But it's ugly.

I removed the previous implementation that creates a new directory and stores
the log file there. I don't like the pg_upgrade-style directory because (a) it
stores part of the server log files in another place and (b) it is another
directory to ignore if your tool handles the data directory (like a backup
tool). My last test said it prints 35 server log lines. I expect that the user
redirects the output to a file so he/she can inspect it later if required.

[1] https://www.postgresql.org/message-id/34637e7f-0330-420d-8f45-1d022962d2fe%40app.fastmail.com
[2] https://www.postgresql.org/message-id/TYCPR01MB12077E2ACB82680CD7BAA22F9F52D2%40TYCPR01MB12077.jpnprd01.prod.outlook.com
[3] https://www.postgresql.org/message-id/085203c4-580d-4c50-90e3-e47249e14585%40eisentraut.org
[4] https://www.postgresql.org/message-id/7a970912-0b77-4942-84f7-2c9ca0bc05a5%40eisentraut.org
[5] https://www.postgresql.org/message-id/CALDaNm053p5Drh9%2B%3DVvkH0Zf_3xoLB%2Bxw8RsNZjEwhc5xc_W%3DQ%40mail.gmail.com

---

From: "Euler Taveira" <[email-masked]>
Date: Thu, 21 Mar 2024 01:48:30 -0300

> (The tool strongly recommends to use active voice, but I can ignore for now)

Thanks for another review. I posted a new patch (v32) that hopefully addresses
these points.

> logical replication setup."

Fixed.

> setup and pg_createsubscriber."

Not fixed.

> replication state is ready."

I avoided pg_createsubscriber at the beginning because it is already
used in the previous sentence. I kept the last part of the sentence
because it is similar to one in the logical replication [1].

> and start server several times, and b) only the serial copy works. Can you

Reading the sentence again, it is not clear. When I said "most of
execution time" I was referring to the actual logical replication setup.

> subscriber by promoting and creating subscriptions.

Isn't it better to leave these details to "How It Works"?

> "If these are not met, an error will be reported."

Fixed.

> "The given target data directory must have the same system identifier as the

Fixed.

> This line is not needed if bullet-style is not used. The line is just a supplement,

Fixed.

> "The source server must accept connections from the target server and not be in recovery."

Not fixed.

> Same as 08, this line is not needed if bullet-style is not used.

Fixed.

> settings during transformation. Hence, connections to the target server might fail."

Fixed.

> "During the recovery process,"

Not fixed. Our documentation uses "recovery process".

> "replicated, so an error would occur."

I didn't find this one. Maybe you checked a previous version.

> "It would avoid situations in which the target server might use WAL files from

Fixed.

> s/a/an

Fixed.

> s/of/of the/

Fixed.

> We can do double-quote for the word promote.

Why? It is referring to recovery_target_action. If you check this GUC,
you will notice that it also uses literal tag.

> "are also added so it avoids"

Fixed.

> "accepts read-write transactions"

Not fixed.

> (Sometimes we cannot identify that the implementation is intentinal or not.)

I don't know what are you referring to? If the new options are
--publication, --subscription and --replication-slot, they are
documented. Are you checking the latest patch?

> objects by pg_createsubscriber. How do other think?

I prefer explicit options. We can always expand it later if people think it is a
good idea to provide a format string.

> No one refers the define.

It was removed in v30.

> Declarations after the "{" are not needed, because we do not do typedef.

It is a leftover when I removed the typedef.

> changed but not the constant. Is it just missing or is there another rule?

It slipped my mind. Peter's fixups improves it.

> My idea is num_custom_pubs or something. Thought?

What does "custom" add to make the name clear? I added comments saying
so.

> main. So, how about makeing `CreateSubscriberOptions opt` to global one?

I avoided turning all the options global variables. Since the cleanup routine
required the target data directory to be a global variable, I just did it and
left the others alone.

> I think it should be specified here that subscriptions won't be removed with the

I rephrased this comment.

> Per comment 25, we can add a reference like "See comments atop the function"

It is a few lines above. I don't think you have to point it out. If you
are unsure about this decision, you should check the whole function.

> usage() was not updated based on recent changes.

Check v30.

> There is a memory-leak if multiple dbname are specified in the conninfo.

It is not a worrying or critical memory leak.

> No need to initialize the seed every time. Can you reuse pg_prng_state?

Sure.

> is specified. This follows the rule for CREATE SUBSCRIPTION.

Agreed.

> returning value (NULL or duplicated string)? We can use the condition (i == num_dbs-1)

It is not. This code path is not critical. You are suggesting to add
complexity here. Efficiency is a good goal but in this case it only adds
complexity with small return.

> Also, comments atop the function must be fixed.

I choose a short name. The comment seems ok to me.

> pg_strdup should be used here.

Fixed.

> options which do not have short notation are listed behind.

Fixed.

> Per other lines, pg_alloc() should be used.

I think you meant pg_malloc. Fixed.

> You said that the leak won't be concerned here. If so, why only 'p' has pg_free()?

Fixed.

> before the boot of the standby. Before that, the callback will be no-op.

The main reason is to catch future cases added *before* the point you
want to move this call that requires a cleanup. As you said it is a
no-op. My preference for atexit() calls is to add it as earlier as
possible to avoid leaving cases that it should trigger.

> separated into another funtion like is_standby_started()?

It is so small that I didn't bother adding a new function for it.

> with the controldata file, I think it can be located in controldata_util.c.

I added this comment here as a reference from where I extracted the
code. The referred function is from backend. Feel free to propose a
separate patch for it.

> > the command-line options are ignored during the execution.

I added a sentence for it. See "How It Works".

[1] https://www.postgresql.org/docs/current/logical-replication-architecture.html#LOGICAL-REPLICATION-SNAPSHOT

---

From: "Euler Taveira" <[email-masked]>
Date: Thu, 21 Mar 2024 01:56:11 -0300

> support everything, so documenting is ok.

Shlok, I'm not sure we should add a sentence about a pilot error. I added a
comment in check_subscriber that describes this situation. I think the comment
is sufficient to understand the limitation and, if it is possible in the future,
a check might be added for it. I didn't include v31-0004.

---

From: "Euler Taveira" <[email-masked]>
Date: Thu, 21 Mar 2024 01:59:06 -0300

> The attached patch has the changes to handle the same.

I included a different query that does the same. See v32.

---

From: Shlok Kyal <[email-masked]>
Date: Thu, 21 Mar 2024 15:19:16 +0530

Hi,

There is a compilation error while building postgres with the patch
due to a recent commit. I have attached a top-up patch v32-0003 to
resolve this compilation error.
I have not updated the version of the patch as I have not made any
change in v32-0001 and v32-0002 patch.

Thanks and regards,
Shlok Kyal

---

From: vignesh C <[email-masked]>
Date: Thu, 21 Mar 2024 17:05:00 +0530

> redirects the output to a file so he/she can inspect it later if required.

Here are a few suggestions:
1) I felt displaying the server log to the console is not a good idea,
I prefer this to be logged. There were similar comments from
Kuroda-san at [1], Peter at [2]. The number of lines will increase
based on the log level set. If you don't want to use pg_upgrade style,
how about exposing the log file option and logging it to the specified
log file.

2) Currently for publication, replication-slot and subscription, we
will have to specify these options based on the number of databases.
Here if we have 100 databases we will have to specify these options
100 times, it might not be user friendly. How about something like
what Tomas had proposed at [3] and  Amit proposed at [4]. It will be
better if the user has to just specify publication, replication slot
and subscription options only one time.
+       /* Number of object names must match number of databases */
+       if (num_pubs > 0 && num_pubs != num_dbs)
+       {
+               pg_log_error("wrong number of publication names");
+               pg_log_error_hint("Number of publication names (%d)
must match number of database names (%d).",
+                                                 num_pubs, num_dbs);
+               exit(1);
+       }

3) Can we have an option similar to dry-run which will display the
configurations required in the primary and standby node something
like:
pg_createsubscriber -D data_N2/ -P "port=5431 user=postgres"  -p 9999
-s /home/vignesh/postgres/inst/bin/ -U postgres -d db1 -d db2
--suggest-config
Suggested optimal configurations in the primary:
--------------------------------------
wallevel = logical
max_replication_slots = 3
max_wal_senders = 3
...
Suggested optimal configurations in the standby:
--------------------------------------
max_replication_slots = 3
max_wal_senders = 3
...

[1] - https://www.postgresql.org/message-id/TY3PR01MB9889593399165B9A04106741F5662%40TY3PR01MB9889.jpnprd01.prod.outlook.com
[2] - https://www.postgresql.org/message-id/7a970912-0b77-4942-84f7-2c9ca0bc05a5%40eisentraut.org
[3] - https://www.postgresql.org/message-id/6423dfeb-a729-45d3-b71e-7bf1b3adb0c9%40enterprisedb.com
[4] - https://www.postgresql.org/message-id/CAA4eK1%2BL_J4GYES6g19xqfpEVD3K2NPCAeu3tzrJfZgu_Fk9Tw%40mail.gmail.com

Regards,
Vignesh

---

From: Peter Eisentraut <[email-masked]>
Date: Thu, 21 Mar 2024 13:32:39 +0100

> how about exposing the log file option and logging it to the specified

Let's leave that for the next version.  We need to wrap things up for 
this release.

> and subscription options only one time.

Same.  Designing, implementing, discussing, and testing this cannot be 
done in the time remaining.

> Suggested optimal configurations in the primary:

How would this be different from what --dry-run does now?

---

From: vignesh C <[email-masked]>
Date: Thu, 21 Mar 2024 19:03:30 +0530

> Let's leave that for the next version.  We need to wrap things up for

Ok, This can be done as an improvement.

> done in the time remaining.

If we commit this we might not be able to change the way the option
behaves once the customers starts using it. How about removing these
options in the first version and adding it in the next version after
more discussion.

> How would this be different from what --dry-run does now?

Currently dry-run will do the check and might fail on identifying a
few failures like after checking subscriber configurations. Then the
user will have to correct the configuration and re-run then fix the
next set of failures. Whereas the suggest-config will display all the
optimal configuration for both the primary and the standby in a single
shot. This is not a must in the first version, it can be done as a
subsequent enhancement.

Regards,
Vignesh

---

From: "Euler Taveira" <[email-masked]>
Date: Thu, 21 Mar 2024 11:30:05 -0300

> options in the first version and adding it in the next version after

We don't need to redesign this one if we want to add a format string in a next
version. A long time ago, pg_dump started to accept pattern for tables without
breaking or deprecating the -t option. If you have 100 databases and you don't
want to specify the options or use a script to generate it for you, you also
have the option to let pg_createsubscriber generate the object names for you.
Per my experience, it will be a rare case.

> shot. This is not a must in the first version, it can be done as a

Do you meant publisher, right? Per order, check_subscriber is done before
check_publisher and it checks all settings on the subscriber before exiting. In
v30, I changed the way it provides the required settings. In a previous version,
it fails when it found a wrong setting; the current version, check all settings
from that server before providing a suitable error.

pg_createsubscriber: checking settings on publisher
pg_createsubscriber: primary has replication slot "physical_slot"
pg_createsubscriber: error: publisher requires wal_level >= logical
pg_createsubscriber: error: publisher requires 2 replication slots, but only 0 remain
pg_createsubscriber: hint: Consider increasing max_replication_slots to at least 3.
pg_createsubscriber: error: publisher requires 2 wal sender processes, but only 0 remain
pg_createsubscriber: hint: Consider increasing max_wal_senders to at least 3.

If you have such an error, you will fix them all and rerun using dry run mode
again to verify everything is ok. I don't have a strong preference about it. It
can be changed easily (unifying the check functions or providing a return for
each of the check functions).

---

From: "Euler Taveira" <[email-masked]>
Date: Fri, 22 Mar 2024 00:31:34 -0300

> change in v32-0001 and v32-0002 patch.

I'm attaching a new version (v33) to incorporate this fix (v32-0003) into the
main patch (v32-0001). This version also includes 2 new tests:

- refuse to run if the standby server is running
- refuse to run if the standby was promoted e.g. it is not in recovery

The first one exercises a recent change (standby should be stopped) and the
second one covers an important requirement.

Based on the discussion [1] about the check functions, Vignesh suggested that it
should check both server before exiting. v33-0003 implements it. I don't have a
strong preference; feel free to apply it.

[1] https://www.postgresql.org/message-id/CALDaNm1Dg5tDRmaabk%2BZND4WF17NrNq52WZxCE%2B90-PGz5trQQ%40mail.gmail.com

---

From: Amit Kapila <[email-masked]>
Date: Fri, 22 Mar 2024 09:24:10 +0530

> Per my experience, it will be a rare case.

But, why go with some UI in the first place which we don't think is a
good one, or at least don't have a broader agreement that it is a good
one? The problem with self-generated names for users could be that
they won't be able to make much out of it. If one has to always use
those internally then probably that would be acceptable. I would
prefer what Tomas proposed a few emails ago as compared to this one as
that has fewer options to be provided by users but still, they can
later identify objects. But surely, we should discuss if you or others
have better alternatives.

The user choosing to convert a physical standby to a subscriber would
in some cases be interested in converting it for all the databases
(say for the case of upgrade [1]) and giving options for each database
would be cumbersome for her.

> each of the check functions).

We can unify the checks but not sure if it is worth it. I am fine
either way. It would have been better if we provided a way for a user
to know the tool's requirement rather than letting him know via errors
but I think it should be okay to extend it later as well.

[1] - This tool can be used in an upgrade where the user first
converts physical standby to subscriber to get incremental changes and
then performs an online upgrade.

-- 
With Regards,
Amit Kapila.

---

From: Fabrízio de Royes Mello <[email-masked]>
Date: Fri, 22 Mar 2024 01:13:55 -0300

> then performs an online upgrade.

The first use case me and Euler discussed some time ago to upstream this
tool was exactly what Amit described so IMHO we should make it easier for
users to subscribe to all existing user databases.

Regards,

--
Fabrízio de Royes Mello

---

From: Amit Kapila <[email-masked]>
Date: Fri, 22 Mar 2024 09:55:48 +0530

On Fri, Mar 22, 2024 at 9:44 AM Fabrízio de Royes Mello
<[email-masked]> wrote:
>
> On Fri, Mar 22, 2024 at 12:54 AM Amit Kapila <[email-masked]> wrote:
> >
> >
> The first use case me and Euler discussed some time ago to upstream this tool was exactly what Amit described so IMHO we should make it easier for users to subscribe to all existing user databases.
>

I feel that will be a good use case for this tool especially now
(with this release) when we allow upgrade of subscriptions. In this
regard, I feel if the user doesn't specify any database it should have
subscriptions for all databases and the user should have a way to
provide publication/slot/subscription names.

-- 
With Regards,
Amit Kapila.

---

From: vignesh C <[email-masked]>
Date: Fri, 22 Mar 2024 19:42:00 +0530

> second one covers an important requirement.

Few comments:
1) In error case PQclear and PQfinish should be called:

2) Call fflush here before calling system command to get proper
ordered console output:
a) Call fflush:
+               int                     rc = system(cmd_str);
+
+               if (rc == 0)
+                       pg_log_info("subscriber successfully changed
the system identifier");
+               else
+                       pg_fatal("subscriber failed to change system
identifier: exit code: %d", rc);
+       }

b) similarly here:

c) similarly here:

3) Call PQClear in error cases:
a) Call PQClear
+       res = PQexec(conn, "SELECT system_identifier FROM
pg_catalog.pg_control_system()");
+       if (PQresultStatus(res) != PGRES_TUPLES_OK)
+       {
+               pg_log_error("could not get system identifier: %s",
+                                        PQresultErrorMessage(res));
+               disconnect_database(conn, true);
+       }

b) similarly here
+       if (PQntuples(res) != 1)
+       {
+               pg_log_error("could not get system identifier: got %d
rows, expected %d row",
+                                        PQntuples(res), 1);
+               disconnect_database(conn, true);
+       }
+

c) similarly here
+       res = PQexec(conn,
+                                "WHERE datname =
pg_catalog.current_database()");

d) There are few more places like this.

4) Since we are using a global variable, we might be able to remove
initializing many of them.
+       /* Default settings */
+       subscriber_dir = NULL;
+       opt.config_file = NULL;
+       opt.pub_conninfo_str = NULL;
+       opt.socket_dir = NULL;
+       opt.sub_port = DEFAULT_SUB_PORT;
+       opt.sub_username = NULL;
+       opt.database_names = (SimpleStringList){0};
+       opt.recovery_timeout = 0;

Regards,
Vignesh

---

From: Shubham Khanna <[email-masked]>
Date: Fri, 22 Mar 2024 19:45:35 +0530

> [1] https://www.postgresql.org/message-id/CALDaNm1Dg5tDRmaabk%2BZND4WF17NrNq52WZxCE%2B90-PGz5trQQ%40mail.gmail.com

I had run valgrind with pg_createsubscriber to see if there were any
issues. Valgrind reported the following issues:
==651272== LEAK SUMMARY:
==651272==    definitely lost: 1,319 bytes in 18 blocks
==651272==    indirectly lost: 1,280 bytes in 2 blocks
==651272==      possibly lost: 44 bytes in 3 blocks
==651272==    still reachable: 3,066 bytes in 22 blocks
==651272==         suppressed: 0 bytes in 0 blocks
==651272==
==651272== For lists of detected and suppressed errors, rerun with: -s
==651272== ERROR SUMMARY: 17 errors from 17 contexts (suppressed: 0 from 0)
The attached report has the details of the same.

Thanks and Regards:
Shubham Khanna.

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Mon, 25 Mar 2024 06:01:39 +0000

Dear Amit, Euler,

I also want to share my opinion, just in case.

> have better alternatives.

IIUC, added options were inspired by Tomas. And he said the limitation (pub/sub/slot
name cannot be specified) was acceptable for the first version. I agree with him.
(To be honest, I feel that options should be fewer for the first release)

> would be cumbersome for her.

+1 for the primary use case.

> but I think it should be okay to extend it later as well.

Both ways are OK, but I prefer to unify checks a bit. The number of working modes
in the same executables should be reduced as much as possible.

Also, I feel the current specification is acceptable. pg_upgrade checks one by
one and exits soon in bad cases. If both old and new clusters have issues, the
first run only reports the old one's issues. After DBA fixes and runs again,
issues on the new cluster are output.

[1]: https://www.postgresql.org/message-id/8d52c226-7e34-44f7-a919-759bf8d81541%40enterprisedb.com

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Mon, 25 Mar 2024 09:16:42 +0000

> (To be honest, I feel that options should be fewer for the first release)

Just to confirm - I don't think it is not completely needed. If we can agree a specification
in sometime, it's OK for me to add them. If you ask me, I still prefer Tomas's approach.

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Mon, 25 Mar 2024 11:48:05 +0000

Dear Shubham,

> issues.

Thanks for running the tool!

> Valgrind reported the following issues:
> ==651272== LEAK SUMMARY:

I read the report. I'm not sure all entries must be fixed. Based on other client
tools, old discussions [1], and current codes, I thought we could determine the below
rule:

* For global variables (and their attributes), no need to free the allocated memory.
* For local variables (and their attributes) in main(), no need to free the allocated memory.
* For local variables in other functions, they should be free'd at the end of the function.

Per above rule and your report, I made a top-up patch which adds pg_free() and
destroyPQExpBuffer() several places. How do you think?

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/

---

From: Peter Eisentraut <[email-masked]>
Date: Mon, 25 Mar 2024 12:55:39 +0100

> second one covers an important requirement.

I have committed your version v33.  I did another pass over the 
identifier and literal quoting.  I added quoting for replication slot 
names, for example, even though they can only contain a restricted set 
of characters, but it felt better to be defensive there.

I'm happy to entertain follow-up patches on some of the details like 
option naming that were still being discussed.  I just wanted to get the 
main functionality in in good time.  We can fine-tune the rest over the 
next few weeks.

> strong preference; feel free to apply it.

I haven't done anything about this.

---

From: Bharath Rupireddy <[email-masked]>
Date: Mon, 25 Mar 2024 18:06:03 +0530

On Mon, Mar 25, 2024 at 5:25 PM Peter Eisentraut <[email-masked]> wrote:
>
> I have committed your version v33.

Looks like BF animals aren't happy, please check -
https://buildfarm.postgresql.org/cgi-bin/show_failures.pl.

-- 
Bharath Rupireddy
PostgreSQL Contributors Team
RDS Open Source Databases
Amazon Web Services: https://aws.amazon.com

---

From: Peter Eisentraut <[email-masked]>
Date: Mon, 25 Mar 2024 13:55:24 +0100

> https://buildfarm.postgresql.org/cgi-bin/show_failures.pl.

Looks like sanitizer failures.  There were a few messages about that 
recently, but those were all just about freeing memory after use, which 
we don't necessarily require for client programs.  So maybe something else.

---

From: "Euler Taveira" <[email-masked]>
Date: Mon, 25 Mar 2024 10:02:01 -0300

> of characters, but it felt better to be defensive there.
Thanks.

> main functionality in in good time.  We can fine-tune the rest over the 
Agree. Let's continue the discussion about the details.

> I haven't done anything about this.
... including this one.

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Mon, 25 Mar 2024 16:06:46 +0000

Dear Bharath, Peter,

> we don't necessarily require for client programs.  So maybe something else.

It seems that there are several time of failures, [1] and [2].

## Analysis for failure 1

The failure caused by a time lag between walreceiver finishes and pg_is_in_recovery()
returns true.

According to the output [1], it seems that the tool failed at wait_for_end_recovery()
with the message "standby server disconnected from the primary". Also, lines
"redo done at..." and "terminating walreceiver process due to administrator command"
meant that walreceiver was requested to shut down by XLogShutdownWalRcv().

According to the source, we confirm that walreceiver is shut down in
StartupXLOG()->FinishWalRecovery()->XLogShutdownWalRcv(). Also, SharedRecoveryState
is changed to RECOVERY_STATE_DONE (this meant the pg_is_in_recovery() return true)
at the latter part of StartupXLOG().

So, if there is a delay between FinishWalRecovery() and change the state, the check
in wait_for_end_recovery() would be failed during the time. Since we allow to miss
the walreceiver 10 times and it is checked once per second, the failure occurs if
the time lag is longer than 10 seconds.

I do not have a good way to fix it. One approach is make NUM_CONN_ATTEMPTS larger,
but it's not a fundamental solution.

## Analysis for failure 2

According to [2], the physical replication slot which is specified as primary_slot_name
was not used by the walsender process. At that time walsender has not existed.

```
...
pg_createsubscriber: publisher: current wal senders: 0
pg_createsubscriber: command is: SELECT 1 FROM pg_catalog.pg_replication_slots WHERE active AND slot_name = 'physical_slot'
pg_createsubscriber: error: could not obtain replication slot information: got 0 rows, expected 1 row
...
```

Currently standby must be stopped before the command and current code does not
block the flow to ensure the replication is started. So there is a possibility
that the checking is run before walsender is launched.

One possible approach is to wait until the replication starts. Alternative one is
to ease the condition.

How do you think?

[1]: https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=serinus&dt=2024-03-25%2013%3A03%3A07
[2]: https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=culicidae&dt=2024-03-25%2013%3A53%3A58

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/

---

From: vignesh C <[email-masked]>
Date: Tue, 26 Mar 2024 06:26:44 +0530

> but it's not a fundamental solution.

I agree with your analysis, another way to fix could be to remove the
following check as increasing the count might still have the race
condition issue:
/*
* If it is still in recovery, make sure the target server is
* connected to the primary so it can receive the required WAL to
* finish the recovery process. If it is disconnected try
* NUM_CONN_ATTEMPTS in a row and bail out if not succeed.
*/
res = PQexec(conn,
"SELECT 1 FROM pg_catalog.pg_stat_wal_receiver");

I'm not sure whether we should worry about the condition where
recovery is not done and pg_stat_wal_receiver is exited as we have the
following sanity check in check_subscriber before we wait for recovery
to be finished:
/* The target server must be a standby */
if (!server_is_in_recovery(conn))
{
pg_log_error("target server must be a standby");
disconnect_database(conn, true);
}

Regards,
Vignesh

---

From: Amit Kapila <[email-masked]>
Date: Tue, 26 Mar 2024 08:03:26 +0530

> main functionality in in good time.  We can fine-tune the rest over the

I was looking at prior discussions on this topic to see if there are
any other open design points apart from this and noticed that the
points raised/discussed in the email [1] are also not addressed. IIRC,
the key point we discussed was that after promotion, the existing
replication objects should be removed (either optionally or always),
otherwise, it can lead to a new subscriber not being able to restart
or getting some unwarranted data.

[1] - https://www.postgresql.org/message-id/CAExHW5t4ew7ZrgcDdTv7YmuG7LVQT1ZaEny_EvtngHtEBNyjcQ%40mail.gmail.com

-- 
With Regards,
Amit Kapila.

---

From: "Euler Taveira" <[email-masked]>
Date: Mon, 25 Mar 2024 23:53:21 -0300

> but it's not a fundamental solution.

I was expecting that slow hosts might have issues in wait_for_end_recovery().
As you said it took a lot of steps between FinishWalRecovery() (where
walreceiver is shutdown -- XLogShutdownWalRcv) and SharedRecoveryState is set to
RECOVERY_STATE_DONE. If this window takes longer than NUM_CONN_ATTEMPTS *
WAIT_INTERVAL (10 seconds), it aborts the execution. That's a bad decision
because it already finished the promotion and it is just doing the final
preparation for the host to become a primary.

        /*   
         * If it is still in recovery, make sure the target server is
         * connected to the primary so it can receive the required WAL to
         * finish the recovery process. If it is disconnected try
         * NUM_CONN_ATTEMPTS in a row and bail out if not succeed.
         */
        res = PQexec(conn,
                     "SELECT 1 FROM pg_catalog.pg_stat_wal_receiver");
        if (PQntuples(res) == 0)
        {
            if (++count > NUM_CONN_ATTEMPTS)
            {
                stop_standby_server(subscriber_dir);
                pg_log_error("standby server disconnected from the primary");
                break;
            }    
        }
        else
            count = 0;          /* reset counter if it connects again */

This code was add to defend against the death/crash of the target server. There
are at least 3 options:

(1) increase NUM_CONN_ATTEMPTS * WAIT_INTERVAL seconds. We discussed this constant
and I decided to use 10 seconds because even in some slow hosts, this time
wasn't reached during my tests. It seems I forgot to test the combination of slow
host, asserts enabled, and ubsan. I didn't notice that pg_promote() uses 60
seconds as default wait. Maybe that's a reasonable value. I checked the
004_timeline_switch test and the last run took: 39.2s (serinus), 33.1s
(culicidae), 18.31s (calliphoridae) and 27.52s (olingo).

(2) check if the primary is not running when walreceiver is not available on the
target server. Increase the connection attempts iif the primary is not running.
Hence, the described case doesn't cause an increment on the count variable.

(3) set recovery_timeout default to != 0 and remove pg_stat_wal_receiver check
protection against the death/crash target server. I explained in a previous
message that timeout may occur in cases that WAL replay to reach consistent
state takes more than recovery-timeout seconds.

Option (1) is the easiest fix, however, we can have the same issue again if a
slow host decides to be even slower, hence, we have to adjust this value again.
Option (2) interprets the walreceiver absence as a recovery end and if the
primary server is running it can indicate that the target server is in the
imminence of the recovery end. Option (3) is not as resilient as the other
options.

The first patch implements a combination of (1) and (2).

> One possible approach is to wait until the replication starts. Alternative one is

That's my suggestion too. I reused NUM_CONN_ATTEMPTS (that was renamed to
NUM_ATTEMPTS in the first patch). See second patch.

--
Euler Taveira
EDB   https://www.enterprisedb.com/
From b2cfa13627bd844b28fe050ffb35e8b9696fdf2e Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email-masked]>
Date: Mon, 25 Mar 2024 22:01:52 -0300
Subject: [PATCH v1 1/2] Improve the code that checks if the recovery is
 finishing

The recovery process has a window between the walreceiver shutdown and
the pg_is_in_recovery function returns false. It means that the
pg_stat_wal_receiver checks can cause the server to finish the recovery
(even if it already reaches the recovery target). Since it checks the
pg_stat_wal_receiver to verify the primary is available, if it does not
return a row, PQping the primary server. If it is up and running, it
can indicate that the target server is finishing the recovery process,
hence, we shouldn't count it as an attempt. It avoids premature failures
on slow hosts.

While on it, increase the number of attempts (10 to 60). The wait time is
the same pg_promote function uses by default.
---
 src/bin/pg_basebackup/pg_createsubscriber.c | 30 +++++++++++++--------
 1 file changed, 19 insertions(+), 11 deletions(-)

diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index b8f8269340..cca93d8c25 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -30,6 +30,8 @@
 
 #define	DEFAULT_SUB_PORT	"50432"
 
+#define NUM_ATTEMPTS		60
+
 /* Command-line options */
 struct CreateSubscriberOptions
 {
@@ -93,7 +95,7 @@ static void pg_ctl_status(const char *pg_ctl_cmd, int rc);
 static void start_standby_server(const struct CreateSubscriberOptions *opt,
 								 bool restricted_access);
 static void stop_standby_server(const char *datadir);
-static void wait_for_end_recovery(const char *conninfo,
+static void wait_for_end_recovery(const struct LogicalRepInfo *dbinfo,
 								  const struct CreateSubscriberOptions *opt);
 static void create_publication(PGconn *conn, struct LogicalRepInfo *dbinfo);
 static void drop_publication(PGconn *conn, struct LogicalRepInfo *dbinfo);
@@ -1354,18 +1356,16 @@ stop_standby_server(const char *datadir)
  * the recovery process. By default, it waits forever.
  */
 static void
-wait_for_end_recovery(const char *conninfo, const struct CreateSubscriberOptions *opt)
+wait_for_end_recovery(const struct LogicalRepInfo *dbinfo, const struct CreateSubscriberOptions *opt)
 {
 	PGconn	   *conn;
 	int			status = POSTMASTER_STILL_STARTING;
 	int			timer = 0;
 	int			count = 0;		/* number of consecutive connection attempts */
 
-#define NUM_CONN_ATTEMPTS	10
-
 	pg_log_info("waiting for the target server to reach the consistent state");
 
-	conn = connect_database(conninfo, true);
+	conn = connect_database(dbinfo->subconninfo, true);
 
 	for (;;)
 	{
@@ -1384,16 +1384,24 @@ wait_for_end_recovery(const char *conninfo, const struct CreateSubscriberOptions
 		}
 
 		/*
-		 * If it is still in recovery, make sure the target server is
-		 * connected to the primary so it can receive the required WAL to
-		 * finish the recovery process. If it is disconnected try
-		 * NUM_CONN_ATTEMPTS in a row and bail out if not succeed.
+		 * If it is still in recovery, make sure the target server is connected
+		 * to the primary so it can receive the required WAL to finish the
+		 * recovery process. If the walreceiver process is not running it
+		 * should indicate that (i) the recovery is almost finished or (ii) the
+		 * primary is not running or is not accpeting connections. It should
+		 * count as attempts iif (ii) is true. In this case, try NUM_ATTEMPTS
+		 * in a row and bail out if not succeed.
 		 */
 		res = PQexec(conn,
 					 "SELECT 1 FROM pg_catalog.pg_stat_wal_receiver");
 		if (PQntuples(res) == 0)
 		{
-			if (++count > NUM_CONN_ATTEMPTS)
+			if (PQping(dbinfo->pubconninfo) != PQPING_OK)
+				count++;
+			else
+				count = 0;		/* reset counter if it connects again */
+
+			if (count > NUM_ATTEMPTS)
 			{
 				stop_standby_server(subscriber_dir);
 				pg_log_error("standby server disconnected from the primary");
@@ -2113,7 +2121,7 @@ main(int argc, char **argv)
 	start_standby_server(&opt, true);
 
 	/* Waiting the subscriber to be promoted */
-	wait_for_end_recovery(dbinfo[0].subconninfo, &opt);
+	wait_for_end_recovery(&dbinfo[0], &opt);
 
 	/*
 	 * Create the subscription for each database on subscriber. It does not
-- 
2.30.2

From 872510ed67b91cb7482dea3bce831549d3519e80 Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email-masked]>
Date: Mon, 25 Mar 2024 23:25:30 -0300
Subject: [PATCH v1 2/2] Improve the code that checks if the primary slot is
 available

The target server is started a few instructions before checking the
primary server and the replication slot might not be active. Instead of
failing the first time, try a few times.
---
 src/bin/pg_basebackup/pg_createsubscriber.c | 35 +++++++++++++++------
 1 file changed, 25 insertions(+), 10 deletions(-)

diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index cca93d8c25..f17e9bde9c 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -887,6 +887,8 @@ check_publisher(const struct LogicalRepInfo *dbinfo)
 	{
 		PQExpBuffer str = createPQExpBuffer();
 		char	   *psn_esc = PQescapeLiteral(conn, primary_slot_name, strlen(primary_slot_name));
+		int			ntuples;
+		int			count = 0;
 
 		appendPQExpBuffer(str,
 						  "SELECT 1 FROM pg_catalog.pg_replication_slots "
@@ -897,25 +899,38 @@ check_publisher(const struct LogicalRepInfo *dbinfo)

-		res = PQexec(conn, str->data);
-		if (PQresultStatus(res) != PGRES_TUPLES_OK)
+		/*
+		 * The replication slot might take some time to be active, try a few
+		 * times if necessary.
+		 */
+		do
 		{
-			pg_log_error("could not obtain replication slot information: %s",
-						 PQresultErrorMessage(res));
-			disconnect_database(conn, true);
-		}
+			res = PQexec(conn, str->data);
+			if (PQresultStatus(res) != PGRES_TUPLES_OK)
+			{
+				pg_log_error("could not obtain replication slot information: %s",
+							 PQresultErrorMessage(res));
+				disconnect_database(conn, true);
+			}
 
-		if (PQntuples(res) != 1)
+			ntuples = PQntuples(res);
+			PQclear(res);
+
+			if (ntuples == 1)	/* replication slot is already active */
+				break;
+			else
+				count++;
+		} while (count > NUM_ATTEMPTS);
+
+		if (ntuples != 1)
 		{
 			pg_log_error("could not obtain replication slot information: got %d rows, expected %d row",
-						 PQntuples(res), 1);
+						 ntuples, 1);
 			disconnect_database(conn, true);
 		}
 		else
 			pg_log_info("primary has replication slot \"%s\"",
 						primary_slot_name);
-
-		PQclear(res);
 	}
 
 	disconnect_database(conn, false);
-- 
2.30.2

---

From: "Euler Taveira" <[email-masked]>
Date: Mon, 25 Mar 2024 23:57:27 -0300

> or getting some unwarranted data.

See setup_subscriber.

        /*
         * Since the publication was created before the consistent LSN, it is
         * available on the subscriber when the physical replica is promoted.
         * Remove publications from the subscriber because it has no use.
         */
        drop_publication(conn, &dbinfo[i]);

---

From: Amit Kapila <[email-masked]>
Date: Tue, 26 Mar 2024 09:36:09 +0530

On Tue, Mar 26, 2024 at 8:27 AM Euler Taveira <[email-masked]> wrote:
>         drop_publication(conn, &dbinfo[I]);
>

This only drops the publications created by this tool, not the
pre-existing ones that we discussed in the link provided.

-- 
With Regards,
Amit Kapila.

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Tue, 26 Mar 2024 04:26:43 +0000

Dear Amit, Euler,

> pre-existing ones that we discussed in the link provided.

Another concern around here is the case which primary subscribes changes from others.
After the conversion, new subscriber also tries to connect to another publisher as
well - this may lead conflicts. This causes because both launcher/workers start
after recovery finishes. So, based on the Ashutosh's point, should we remove
such replication objects?

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/

---

From: Tomas Vondra <[email-masked]>
Date: Tue, 26 Mar 2024 20:12:59 +0100

> NUM_ATTEMPTS in the first patch). See second patch.

Perhaps I'm missing something, but why is NUM_CONN_ATTEMPTS even needed?
Why isn't recovery_timeout enough to decide if wait_for_end_recovery()
waited long enough?

IMHO the test should simply pass PG_TEST_DEFAULT_TIMEOUT when calling
pg_createsubscriber, and that should do the trick.

Increasing PG_TEST_DEFAULT_TIMEOUT is what buildfarm animals doing
things like ubsan/valgrind already use to deal with exactly this kind of
timeout problem.

Or is there a deeper problem with deciding if the system is in recovery?

regards

---

From: "Euler Taveira" <[email-masked]>
Date: Tue, 26 Mar 2024 17:17:15 -0300

> Why isn't recovery_timeout enough to decide if wait_for_end_recovery()

It was an attempt to decoupled a connection failure (that keeps streaming the
WAL) from recovery timeout. The NUM_CONN_ATTEMPTS guarantees that if the primary
is gone during the standby recovery process, there is a way to bail out. The
recovery-timeout is 0 (infinite) by default so you have an infinite wait without
this check. The idea behind this implementation is to avoid exiting in this
critical code path. If it times out here you might have to rebuild the standby
and start again. Amit suggested [1] that we use a value as recovery-timeout but
how high is a good value? I've already saw some long recovery process using
pglogical equivalent that timeout out after hundreds of minutes. Maybe I'm too
worried about a small percentage of cases and we should use 1h as default, for
example. It would reduce the complexity since the recovery process lacks some
progress indicators (LSN is not sufficient in this case and there isn't a
function to provide the current state -- stop applying WAL, reach target, new
timeline, etc).

If we remove the pg_stat_wal_receiver check, we should avoid infinite recovery
by default otherwise we will have some reports saying the tool is hanging when
in reality the primary has gone and WAL should be streamed.

> pg_createsubscriber, and that should do the trick.

That's a good idea. Tests are not exercising the recovery-timeout option.

> Or is there a deeper problem with deciding if the system is in recovery?

As I said with some recovery progress indicators it would be easier to make some
decisions like wait a few seconds because the WAL has already been applied and
it is creating a new timeline. The recovery timeout decision is a shot in the
dark because we might be aborting pg_createsubscriber when the target server is
about to set RECOVERY_STATE_DONE.

[1] https://www.postgresql.org/message-id/CAA4eK1JRgnhv_ySzuFjN7UaX9qxz5Hqcwew7Fk%3D%2BAbJbu0Kd9w%40mail.gmail.com

---

From: Tomas Vondra <[email-masked]>
Date: Tue, 26 Mar 2024 22:29:19 +0100

On 3/26/24 21:17, Euler Taveira wrote:
> and start again.

- This seems like something that should definitely be documented in the
comment before wait_for_end_recovery(). At the moment it only talks
about timeout, and nothing about NUM_CONN_ATTEMPTS.

- The NUM_CONN_ATTEMPTS name seems rather misleading, considering it
does not really count connection attempts, but number of times we have
not seen 1 in pg_catalog.pg_stat_wal_receiver.

- Not sure I follow the logic - it tries to avoid exiting by setting
infinite timeout, but it still exists based on NUM_CONN_ATTEMPTS. Isn't
that somewhat contradictory?

- Isn't the NUM_CONN_ATTEMPTS actually making it more fragile, i.e. more
likely to exit? For example, what if there's a short networking hiccup,
so that the standby can't connect to the primary.

- It seems a bit strange that even with the recovery timeout set, having
the limit of 10 "connection attempts" effectively establishes a separate
hard-coded limit of 10 seconds. Seems a bit surprising if I set recovery
limit to 1 minute, and it just dies after 10 seconds.

> Amit suggested [1] that we use a value as recovery-timeout but

I don't think there's a default timeout value that would work for
everyone. Either it's going to be too short for some cases, or it'll
take too long for some other cases.

I think there are two obvious default values for the timeout - infinity,
and 60 seconds, which is the default we use for other CLI tools (like
pg_ctl and so on). Considering the negative impact of exiting, I'd say
it's better to default to infinity. It's always possible to Ctrl-C or
terminate the process in some other way, if needed.

As for people complaining about infinite recovery - perhaps it'd be
sufficient to mention this in the messages printed by the tool, to make
it clearer. Or maybe even print something in the loop, because right now
it's entirely silent so it's easy to believe it's stuck. Perhaps not on
every loop, but at least in verbose mode it should print something.

> about to set RECOVERY_STATE_DONE.

Isn't it enough to check data in pg_stat_replication on the primary?

regards

---

From: Amit Kapila <[email-masked]>
Date: Mon, 29 Apr 2024 15:26:48 +0530

On Wed, Mar 27, 2024 at 1:47 AM Euler Taveira <[email-masked]> wrote:
> is gone during the standby recovery process, there is a way to bail out.
>

I think we don't need to check primary if the WAL corresponding to
consistent_lsn is already present on the standby. Shouldn't we first
check that? Once we ensure that the required WAL is copied, just
checking server_is_in_recovery() should be sufficient. I feel that
will be a direct way of ensuring what is required rather than
indirectly verifying the same (by checking pg_stat_wal_receiver) as we
are doing currently.

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Mon, 29 Apr 2024 16:28:17 +0530

> NUM_ATTEMPTS in the first patch). See second patch.

How can we guarantee that the slot can become active after these many
attempts? It still could be possible that on some slow machines it
didn't get activated even after NUM_ATTEMPTS. BTW, in the first place,
why do we need to ensure that the 'primary_slot_name' is active on the
primary?

-- 
With Regards,
Amit Kapila.

---

From: "Euler Taveira" <[email-masked]>
Date: Mon, 29 Apr 2024 08:53:19 -0300

> indirectly verifying the same (by checking pg_stat_wal_receiver) as we

How would you check it? WAL file? During recovery, you are not allowed to use
pg_current_wal_lsn.

Tomas suggested to me off-list that we should adopt a simple solution in
wait_for_end_recovery: wait for recovery_timeout without additional checks
(which means remove the pg_stat_wal_receiver logic).  When we have additional
information that we can reliably use in this function, we can add it. Hence, it
is also easy to adjust the PG_TEST_TIMEOUT_DEFAULT to have stable tests.

---

From: Amit Kapila <[email-masked]>
Date: Mon, 29 Apr 2024 17:28:10 +0530

> How would you check it? WAL file? During recovery, you are not allowed to use

How about pg_last_wal_receive_lsn()?

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Tue, 30 Apr 2024 12:04:17 +0530

> On Mon, Apr 29, 2024 at 5:23 PM Euler Taveira <[email-masked]> wrote:

I was trying to test this utility when 'sync_replication_slots' is on
and it gets in an ERROR loop [1] and never finishes. Please find the
postgresql.auto used on the standby attached. I think if the standby
has enabled sync_slots, you need to pass dbname in
GenerateRecoveryConfig(). I couldn't test it further but I wonder if
there are already synced slots on the standby (either due to
'sync_replication_slots' or users have used
pg_sync_replication_slots() before invoking pg_createsubscriber),
those would be retained as it is on new subscriber and lead to
unnecessary WAL retention and dead rows.

[1]
2024-04-30 11:50:43.239 IST [12536] LOG:  slot sync worker started
2024-04-30 11:50:43.247 IST [12536] ERROR:  slot synchronization
requires dbname to be specified in primary_conninfo

-- 
With Regards,
Amit Kapila.

---

From: Amit Kapila <[email-masked]>
Date: Tue, 30 Apr 2024 14:17:54 +0530

On Tue, Apr 30, 2024 at 12:04 PM Amit Kapila <[email-masked]> wrote:
> GenerateRecoveryConfig().

The other possibility is that when we start standby from
pg_createsubscriber, we specifically set 'sync_replication_slots' as
false.

>
> I couldn't test it further but I wonder if

This still needs some handling.

BTW, I don't see the use of following messages in --dry-run mode or
rather they could be misleading:
pg_createsubscriber: hint: If pg_createsubscriber fails after this
point, you must recreate the physical replica before continuing.
...
...
pg_createsubscriber: setting the replication progress (node name
"pg_0" ; LSN 0/0) on database "postgres"

Similarly, we should think if below messages are useful in --dry-run mode:
pg_createsubscriber: dropping publication
"pg_createsubscriber_5_887f7991" on database "postgres"
pg_createsubscriber: creating subscription
"pg_createsubscriber_5_887f7991" on database "postgres"
...
pg_createsubscriber: enabling subscription
"pg_createsubscriber_5_887f7991" on database "postgres"

-- 
With Regards,
Amit Kapila.

---

From: Thomas Munro <[email-masked]>
Date: Sun, 19 May 2024 10:55:11 +1200

040_pg_createsubscriber.pl seems to be failing occasionally on
culicidae near "--dry-run on node S".  I couldn't immediately see why.
That animal is using EXEC_BACKEND and I also saw a one-off failure a
bit like that on my own local Linux + EXEC_BACKEND test run
(sorry I didn't keep the details around).  Coincidence?

---

From: Amit Kapila <[email-masked]>
Date: Mon, 20 May 2024 15:58:50 +0530

> (sorry I didn't keep the details around).  Coincidence?

AFAICS, this is the same as one of the two BF failures being discussed
in this thread.

-- 
With Regards,
Amit Kapila.

---

From: Shlok Kyal <[email-masked]>
Date: Mon, 20 May 2024 16:29:50 +0530

Hi,

Hi,

I tested the scenario posted by Amit in [1], in which retaining synced
slots lead to unnecessary WAL retention and ERROR. This is raised as
the second open point in [2].
The steps to reproduce the issue:
(1) Setup physical replication with sync slot feature turned on by
setting sync_replication_slots = 'true' or using
pg_sync_replication_slots() on the standby node.
For physical replication setup, run pg_basebackup with -R  and -d option.
(2) Create a logical replication slot on primary node with failover
option as true. A corresponding slot is created on standby as part of
sync slot feature.
(3) Run pg_createsubscriber on standby node.
(4) On Checking for the replication slot on standby node, I noticed
that the logical slots created in step 2 are retained.
 I have attached the script to reproduce the issue.

I and Kuroda-san worked to resolve open points. Here are patches to
solve the second and third point in [2].
Patches proposed by Euler are also attached just in case, but they
were not modified.

v2-0001: not changed
v2-0002: not changed
v2-0003: ensures the slot sync is disabled during the conversion. This
resolves the second point.
v2-0004: drops sync slots which may be retained after running. This
resolves the second point.
v2-0005: removes misleading output messages in dry-run. This resolves
the third point.

[1]: https://www.postgresql.org/message-id/CAA4eK1KdCb%2B5sjYu6qCMXXdCX1y_ihr8kFzMozq0%3DP%3DauYxgog%40mail.gmail.com
[2]: https://www.postgresql.org/message-id/CAA4eK1J22UEfrqx222h5j9DQ7nxGrTbAa_BC%2B%3DmQXdXs-RCsew%40mail.gmail.com

Thanks and Regards,
Shlok Kyal

---

From: Amit Kapila <[email-masked]>
Date: Wed, 22 May 2024 16:49:52 +0530

> Patches proposed by Euler are also attached just in case, but they

Shouldn't we modify it as per the suggestion given in the email [1]? I
am wondering if we can entirely get rid of checking the primary
business and simply rely on recovery_timeout and keep checking
server_is_in_recovery(). If so, we can modify the test to use
non-default recovery_timeout (say 180s or something similar if we have
used it at any other place). As an additional check we can ensure that
constent_lsn is present on standby.

> v2-0002: not changed

We have added more tries to see if the primary_slot_name becomes
active but I think it is still fragile because it is possible on slow
machines that the required slot didn't become active even after more
retries. I have raised the same comment previously [2] and asked an
additional question but didn't get any response.

[1] - https://www.postgresql.org/message-id/CAA4eK1JJq_ER6Kq_H%3DjKHR75QPRd8y9_D%3DRtYw%3DaPYKMfqLi9A%40mail.gmail.com
[2] - https://www.postgresql.org/message-id/CAA4eK1LT3Z13Dg6p4Z%2B4adO_EY-ow5CmWfikEmBfL%3DeVrm8CPw%40mail.gmail.com

-- 
With Regards,
Amit Kapila.

---

From: "Euler Taveira" <[email-masked]>
Date: Wed, 22 May 2024 12:16:24 -0300

> constent_lsn is present on standby.

That's exactly what I want to propose as Tomas convinced me offlist that less is
better when we don't have a useful recovery progress reporting mechanism to make
sure it is still working on the recovery and we should wait.

> additional question but didn't get any response.

Following the same line that simplifies the code, we can: (a) add a loop in
check_subscriber() that waits until walreceiver is available on subscriber or
(b) use a timeout. The main advantage of (a) is that the primary slot is already
available but I'm afraid we need a escape mechanism for the loop (timeout?).

I'll summarize all issues as soon as I finish the review of sync slot support. I
think we should avoid new development if we judge that the item can be
documented as a limitation for this version. Nevertheless, I will share patches
so you can give your opinion on whether it is an open item or new development.

---

From: Amit Kapila <[email-masked]>
Date: Thu, 23 May 2024 14:24:25 +0530

> available but I'm afraid we need a escape mechanism for the loop (timeout?).

Sorry, it is not clear to me why we need any additional loop in
check_subscriber(), aren't we speaking about the problem in
check_publisher() function?

Why in the first place do we need to ensure that primary_slot_name is
active on the primary? You mentioned something related to WAL
retention but I don't know how that is related to this tool's
functionality. If at all, we are bothered about WAL retention on the
primary that should be the WAL corresponding to consistent_lsn
computed by setup_publisher() but this check doesn't seem to ensure
that.

-- 
With Regards,
Amit Kapila.

---

From: Shlok Kyal <[email-masked]>
Date: Fri, 24 May 2024 15:07:05 +0530

> constent_lsn is present on standby.

I have made changes as per your suggestion. I have used
pg_last_wal_receive_lsn to get lsn last wal received on standby and
check if consistent_lsn is already present on the standby.
I have only made changes in v3-0001. v3-0002, v3-0003, v3-0004 and
v3-0005 are not modified.

Thanks and Regards,
Shlok Kyal

---

From: "Euler Taveira" <[email-masked]>
Date: Fri, 07 Jun 2024 00:49:25 -0300

> so you can give your opinion on whether it is an open item or new development.

Here it is a patch series to fix the issues reported in recent discussions. The
patches 0001 and 0003 aim to fix the buildfarm issues. The patch 0002 removes
synchronized failover slots on subscriber since it has no use. I also included
an optional patch 0004 that improves the usability by checking both servers if
it already failed in any subscriber check.

As I said in this previous email I decided to remove the logic that reacts for
an issue on primary. We can reintroduce another code later if/when we have a
better way to check the recovery progress. It will rely on the recovery_timeout
and it adds recovery_timeout equals to PG_TEST_TIMEOUT_DEFAULT to let the
animals control how long it should wait for the recovery.

Since some animals reported some issues in the check_publisher routine that
checks if the primary_slot_name is in use on primary, this logic was removed
too (patch 0003). We could introduce a way to keep trying this test but the
conclusion is that it is not worth it and if the primary_slot_name does not
exist (due to a setup error), pg_createsubscriber will log an error message and
continue.

The 0002 removes any failover slots that remains on subscriber. Talking about
terminology, I noticed that slotsync.c uses "logical failover slots" and
"failover logical slots", I think the latter sounds better but I'm not a native
speaker. I also don't know if we use a short terminology like "failover slots"
"failover replication slots" or "failover logical replication slots". IMO we
can omit "logical" because "failover" infers it is a logical replication slot.
I'm also not sure about omitting "replication". It is not descriptive enough. I
prefer "failover replication slots".

Before sending this email I realized that I did nothing about physical
replication slots on the standby. I think we should also remove them too
unconditionally.

--
Euler Taveira
EDB   https://www.enterprisedb.com/
From 5d8b4781e6e9bcb00564f45c25e575a8abab6ae8 Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email-masked]>
Date: Tue, 21 May 2024 23:04:57 -0300
Subject: [PATCH 1/4] Only the recovery_timeout controls the end of recovery
 process

It used to check if the target server is connected to the primary server
(send required WAL) to rapidly react when the process won't succeed.
This code is not enough to guarantee that the recovery process will
complete. There is a window between the walreceiver shutdown and the
pg_is_in_recovery() returns false that can reach NUM_CONN_ATTEMPTS
attempts and fails.
Instead, rely only on recovery_timeout option to give up the process
after the specified number of seconds.
---
 doc/src/sgml/ref/pg_createsubscriber.sgml     |  7 -----
 src/bin/pg_basebackup/pg_createsubscriber.c   | 29 ++-----------------
 .../t/040_pg_createsubscriber.pl              |  2 ++
 3 files changed, 5 insertions(+), 33 deletions(-)

diff --git a/doc/src/sgml/ref/pg_createsubscriber.sgml b/doc/src/sgml/ref/pg_createsubscriber.sgml
index 142bffff02..a700697f88 100644
--- a/doc/src/sgml/ref/pg_createsubscriber.sgml
+++ b/doc/src/sgml/ref/pg_createsubscriber.sgml
@@ -325,13 +325,6 @@ PostgreSQL documentation
     connections to the target server should fail.
    </para>
 
-   <para>
-    During the recovery process, if the target server disconnects from the
-    source server, <application>pg_createsubscriber</application> will check a
-    few times if the connection has been reestablished to stream the required
-    WAL.  After a few attempts, it terminates with an error.
-   </para>
-
    <para>
     Since DDL commands are not replicated by logical replication, avoid
     executing DDL commands that change the database schema while running
diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index 90cc580811..f62f34b1a7 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -1360,6 +1360,9 @@ stop_standby_server(const char *datadir)
  *
  * If recovery_timeout option is set, terminate abnormally without finishing
  * the recovery process. By default, it waits forever.
+ *
+ * XXX Is the recovery process still in progress? When recovery process has a
+ * better progress reporting mechanism, it should be added here.
  */
 static void
 wait_for_end_recovery(const char *conninfo, const struct CreateSubscriberOptions *opt)
@@ -1367,9 +1370,6 @@ wait_for_end_recovery(const char *conninfo, const struct CreateSubscriberOptions
 	PGconn	   *conn;
 	int			status = POSTMASTER_STILL_STARTING;
 	int			timer = 0;
-	int			count = 0;		/* number of consecutive connection attempts */
-
-#define NUM_CONN_ATTEMPTS	10

@@ -1377,7 +1377,6 @@ wait_for_end_recovery(const char *conninfo, const struct CreateSubscriberOptions
 
 	for (;;)
 	{
-		PGresult   *res;
 		bool		in_recovery = server_is_in_recovery(conn);
 
 		/*
@@ -1391,28 +1390,6 @@ wait_for_end_recovery(const char *conninfo, const struct CreateSubscriberOptions
 			break;
 		}
 
-		/*
-		 * If it is still in recovery, make sure the target server is
-		 * connected to the primary so it can receive the required WAL to
-		 * finish the recovery process. If it is disconnected try
-		 * NUM_CONN_ATTEMPTS in a row and bail out if not succeed.
-		 */
-		res = PQexec(conn,
-					 "SELECT 1 FROM pg_catalog.pg_stat_wal_receiver");
-		if (PQntuples(res) == 0)
-		{
-			if (++count > NUM_CONN_ATTEMPTS)
-			{
-				stop_standby_server(subscriber_dir);
-				pg_log_error("standby server disconnected from the primary");
-				break;
-			}
-		}
-		else
-			count = 0;			/* reset counter if it connects again */
-
-		PQclear(res);
-
 		/* Bail out after recovery_timeout seconds if this option is set */
 		if (opt->recovery_timeout > 0 && timer >= opt->recovery_timeout)
 		{
diff --git a/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
index 2b883e6910..a677fefa94 100644
--- a/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
+++ b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
@@ -264,6 +264,7 @@ $node_p->restart;
 command_ok(
 	[
 		'pg_createsubscriber', '--verbose',
+		'--recovery-timeout', "$PostgreSQL::Test::Utils::timeout_default",
 		'--dry-run', '--pgdata',
 		$node_s->data_dir, '--publisher-server',
 		$node_p->connstr('pg1'), '--socket-directory',
@@ -301,6 +302,7 @@ command_ok(
 command_ok(
 	[
 		'pg_createsubscriber', '--verbose',
+		'--recovery-timeout', "$PostgreSQL::Test::Utils::timeout_default",
 		'--verbose', '--pgdata',
 		$node_s->data_dir, '--publisher-server',
 		$node_p->connstr('pg1'), '--socket-directory',
-- 
2.30.2

From a10edc03a6636d6fef9e013a507a8258096a6fd3 Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email-masked]>
Date: Wed, 22 May 2024 12:43:13 -0300
Subject: [PATCH 2/4] Remove failover replication slots on subscriber

After running pg_createsubscriber, these replication slots has no use on
subscriber.
---
 doc/src/sgml/ref/pg_createsubscriber.sgml     |  8 +++
 src/bin/pg_basebackup/pg_createsubscriber.c   | 49 ++++++++++++++++++-
 .../t/040_pg_createsubscriber.pl              | 20 +++++++-
 3 files changed, 75 insertions(+), 2 deletions(-)

diff --git a/doc/src/sgml/ref/pg_createsubscriber.sgml b/doc/src/sgml/ref/pg_createsubscriber.sgml
index a700697f88..ac56c0cce9 100644
--- a/doc/src/sgml/ref/pg_createsubscriber.sgml
+++ b/doc/src/sgml/ref/pg_createsubscriber.sgml
@@ -482,6 +482,14 @@ PostgreSQL documentation
      </para>
     </step>
 
+    <step>
+     <para>
+      If the standby server contains
+      <link linkend="logicaldecoding-replication-slots-synchronization">failover replication slots</link>,
+      they cannot be synchronized anymore so drop them.
+     </para>
+    </step>
+
     <step>
      <para>
       Update the system identifier on the target server. The
diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index f62f34b1a7..b42160c6c6 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -85,6 +85,7 @@ static void setup_recovery(const struct LogicalRepInfo *dbinfo, const char *data
 						   const char *lsn);
 static void drop_primary_replication_slot(struct LogicalRepInfo *dbinfo,
 										  const char *slotname);
+static void drop_failover_replication_slots(struct LogicalRepInfo *dbinfo);
 static char *create_logical_replication_slot(PGconn *conn,
 											 struct LogicalRepInfo *dbinfo);
 static void drop_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo,
@@ -1178,6 +1179,49 @@ drop_primary_replication_slot(struct LogicalRepInfo *dbinfo, const char *slotnam
 	}
 }
 
+/*
+ * Drop failover replication slots on subscriber. After the transformation, it
+ * has no use.
+ *
+ * XXX we might not fail here. Instead, we provide a warning so the user
+ * eventually drops this replication slot later.
+ */
+static void
+drop_failover_replication_slots(struct LogicalRepInfo *dbinfo)
+{
+	PGconn	   *conn;
+	PGresult   *res;
+
+	conn = connect_database(dbinfo[0].subconninfo, false);
+	if (conn != NULL)
+	{
+		/* Get failover replication slot names. */
+		res = PQexec(conn,
+					 "SELECT slot_name FROM pg_catalog.pg_replication_slots WHERE failover");
+
+		if (PQresultStatus(res) == PGRES_TUPLES_OK)
+		{
+			/* Remove failover replication slots from subscriber. */
+			for (int i = 0; i < PQntuples(res); i++)
+				drop_replication_slot(conn, &dbinfo[0], PQgetvalue(res, i, 0));
+		}
+		else
+		{
+			pg_log_warning("could not obtain failover replication slot information: %s",
+						   PQresultErrorMessage(res));
+			pg_log_warning_hint("Drop the failover replication slots on subscriber soon to avoid retention of WAL files.");
+		}
+
+		PQclear(res);
+		disconnect_database(conn, false);
+	}
+	else
+	{
+		pg_log_warning("could not drop failover replication slot");
+		pg_log_warning_hint("Drop the failover replication slots on subscriber soon to avoid retention of WAL files.");
+	}
+}
+
 /*
  * Create a logical replication slot and returns a LSN.
  *
@@ -1309,7 +1353,7 @@ start_standby_server(const struct CreateSubscriberOptions *opt, bool restricted_
 	PQExpBuffer pg_ctl_cmd = createPQExpBuffer();
 	int			rc;
 
-	appendPQExpBuffer(pg_ctl_cmd, "\"%s\" start -D \"%s\" -s",
+	appendPQExpBuffer(pg_ctl_cmd, "\"%s\" start -D \"%s\" -s -o \"-c sync_replication_slots=off\"",
 					  pg_ctl_path, subscriber_dir);
 	if (restricted_access)
 	{
@@ -2111,6 +2155,9 @@ main(int argc, char **argv)
 	/* Remove primary_slot_name if it exists on primary */
 	drop_primary_replication_slot(dbinfo, primary_slot_name);
 
+	/* Remove failover  replication slots if it exists on subscriber */
+	drop_failover_replication_slots(dbinfo);
+
 	/* Stop the subscriber */
 	pg_log_info("stopping the subscriber");
 	stop_standby_server(subscriber_dir);
diff --git a/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
index a677fefa94..0516d4e17e 100644
--- a/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
+++ b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
@@ -88,6 +88,7 @@ command_fails(
 
 # Set up node P as primary
 my $node_p = PostgreSQL::Test::Cluster->new('node_p');
+my $pconnstr = $node_p->connstr;
 $node_p->init(allows_streaming => 'logical');
 $node_p->start;
 
@@ -122,6 +123,8 @@ $node_s->init_from_backup($node_p, 'backup_1', has_streaming => 1);
 $node_s->append_conf(
 	'postgresql.conf', qq[
 primary_slot_name = '$slotname'
+primary_conninfo = '$pconnstr dbname=postgres'
+hot_standby_feedback = on
 ]);
 $node_s->set_standby_mode();
 $node_s->start;
@@ -260,6 +263,16 @@ max_worker_processes = 8
 # Restore default settings on both servers
 $node_p->restart;
 
+# Create failover slot to test its removal
+my $fslotname = 'failover_slot';
+$node_p->safe_psql('pg1',
+	"SELECT pg_create_logical_replication_slot('$fslotname', 'pgoutput', false, false, true)");
+$node_s->start;
+$node_s->safe_psql('postgres', "SELECT pg_sync_replication_slots()");
+my $result = $node_s->safe_psql('postgres', "SELECT slot_name FROM pg_replication_slots WHERE slot_name = '$fslotname' AND synced AND NOT temporary");
+is($result, 'failover_slot', 'failover slot is synced');
+$node_s->stop;
+
 # dry run mode on node S
 command_ok(
 	[
@@ -318,7 +331,7 @@ command_ok(
 	'run pg_createsubscriber on node S');
 
 # Confirm the physical replication slot has been removed
-my $result = $node_p->safe_psql('pg1',
+$result = $node_p->safe_psql('pg1',
 	"SELECT count(*) FROM pg_replication_slots WHERE slot_name = '$slotname'"
 );
 is($result, qq(0),
@@ -343,6 +356,11 @@ my @subnames = split("\n", $result);
 $node_s->wait_for_subscription_sync($node_p, $subnames[0]);
 $node_s->wait_for_subscription_sync($node_p, $subnames[1]);
 
+# Confirm the failover slot has been removed
+$result = $node_s->safe_psql('pg1',
+	"SELECT count(*) FROM pg_replication_slots WHERE slot_name = '$fslotname'");
+is($result, qq(0), 'failover slot was removed');
+
 # Check result on database pg1
 $result = $node_s->safe_psql('pg1', 'SELECT * FROM tbl1');
 is( $result, qq(first row
-- 
2.30.2

From f085fbef4ecd31aed470d858e57e9ea3b8014493 Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email-masked]>
Date: Fri, 24 May 2024 14:18:15 -0300
Subject: [PATCH 3/4] Remove replication slot check on primary

It used to check if the replication slot exists and is active on
primary. This check might fail on slow hosts because the replication
slot might not be active at the time of this check.

The current code obtains the replication slot name from the
primary_slot_name on standby and assumes the replication slot exists and
is active on primary. If it doesn't exist, this tool will log an error
and continue.
---
 src/bin/pg_basebackup/pg_createsubscriber.c | 53 ++-------------------
 1 file changed, 5 insertions(+), 48 deletions(-)

diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index b42160c6c6..6d5127c45e 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -880,47 +880,6 @@ check_publisher(const struct LogicalRepInfo *dbinfo)
 	pg_log_debug("publisher: max_wal_senders: %d", max_walsenders);
 	pg_log_debug("publisher: current wal senders: %d", cur_walsenders);
 
-	/*
-	 * If standby sets primary_slot_name, check if this replication slot is in
-	 * use on primary for WAL retention purposes. This replication slot has no
-	 * use after the transformation, hence, it will be removed at the end of
-	 * this process.
-	 */
-	if (primary_slot_name)
-	{
-		PQExpBuffer str = createPQExpBuffer();
-		char	   *psn_esc = PQescapeLiteral(conn, primary_slot_name, strlen(primary_slot_name));
-
-		appendPQExpBuffer(str,
-						  "SELECT 1 FROM pg_catalog.pg_replication_slots "
-						  "WHERE active AND slot_name = %s",
-						  psn_esc);
-
-		pg_free(psn_esc);
-
-		pg_log_debug("command is: %s", str->data);
-
-		res = PQexec(conn, str->data);
-		if (PQresultStatus(res) != PGRES_TUPLES_OK)
-		{
-			pg_log_error("could not obtain replication slot information: %s",
-						 PQresultErrorMessage(res));
-			disconnect_database(conn, true);
-		}
-
-		if (PQntuples(res) != 1)
-		{
-			pg_log_error("could not obtain replication slot information: got %d rows, expected %d row",
-						 PQntuples(res), 1);
-			disconnect_database(conn, true);
-		}
-		else
-			pg_log_info("primary has replication slot \"%s\"",
-						primary_slot_name);
-
-		PQclear(res);
-	}
-
 	disconnect_database(conn, false);
 
 	if (strcmp(wal_level, "logical") != 0)
@@ -2105,12 +2064,7 @@ main(int argc, char **argv)
 	/* Check if the standby server is ready for logical replication */
 	check_subscriber(dbinfo);
 
-	/*
-	 * Check if the primary server is ready for logical replication. This
-	 * routine checks if a replication slot is in use on primary so it relies
-	 * on check_subscriber() to obtain the primary_slot_name. That's why it is
-	 * called after it.
-	 */
+	/* Check if the primary server is ready for logical replication */
 	check_publisher(dbinfo);
 
 	/*
@@ -2152,7 +2106,10 @@ main(int argc, char **argv)
 	 */
 	setup_subscriber(dbinfo, consistent_lsn);
 
-	/* Remove primary_slot_name if it exists on primary */
+	/*
+	 * Remove primary_slot_name if it exists on primary.  This replication slot
+	 * has no use after the transformation.
+	 */
 	drop_primary_replication_slot(dbinfo, primary_slot_name);
 
 	/* Remove failover  replication slots if it exists on subscriber */
-- 
2.30.2

From a4f10c849f7584c4fb12005cec04b5d09701752f Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email-masked]>
Date: Fri, 24 May 2024 14:29:06 -0300
Subject: [PATCH 4/4] Check both servers before exiting

The current code provides multiple errors for each server. If any
subscriber condition is not met, it terminates without checking the
publisher conditions.
This change allows it to check both servers before terminating if any
condition is not met. It saves at least one dry run execution.
---
 src/bin/pg_basebackup/pg_createsubscriber.c | 24 ++++++++++++---------
 1 file changed, 14 insertions(+), 10 deletions(-)

diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index 6d5127c45e..b0233786d4 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -76,9 +76,9 @@ static uint64 get_standby_sysid(const char *datadir);
 static void modify_subscriber_sysid(const struct CreateSubscriberOptions *opt);
 static bool server_is_in_recovery(PGconn *conn);
 static char *generate_object_name(PGconn *conn);
-static void check_publisher(const struct LogicalRepInfo *dbinfo);
+static bool check_publisher(const struct LogicalRepInfo *dbinfo);
 static char *setup_publisher(struct LogicalRepInfo *dbinfo);
-static void check_subscriber(const struct LogicalRepInfo *dbinfo);
+static bool check_subscriber(const struct LogicalRepInfo *dbinfo);
 static void setup_subscriber(struct LogicalRepInfo *dbinfo,
 							 const char *consistent_lsn);
 static void setup_recovery(const struct LogicalRepInfo *dbinfo, const char *datadir,
@@ -803,7 +803,7 @@ server_is_in_recovery(PGconn *conn)
  *
  * XXX Does it not allow a synchronous replica?
  */
-static void
+static bool
 check_publisher(const struct LogicalRepInfo *dbinfo)
 {
 	PGconn	   *conn;
@@ -908,8 +908,7 @@ check_publisher(const struct LogicalRepInfo *dbinfo)
 
 	pg_free(wal_level);
 
-	if (failed)
-		exit(1);
+	return failed;
 }
 
 /*
@@ -923,7 +922,7 @@ check_publisher(const struct LogicalRepInfo *dbinfo)
  * there is not a reliable way to provide a suitable error saying the server C
  * will be broken at the end of this process (due to pg_resetwal).
  */
-static void
+static bool
 check_subscriber(const struct LogicalRepInfo *dbinfo)
 {
 	PGconn	   *conn;
@@ -1014,8 +1013,7 @@ check_subscriber(const struct LogicalRepInfo *dbinfo)
 		failed = true;
 	}
 
-	if (failed)
-		exit(1);
+	return failed;
 }
 
 /*
@@ -1771,6 +1769,9 @@ main(int argc, char **argv)
 
 	char		pidfile[MAXPGPATH];
 
+	bool		pub_failed = false;
+	bool		sub_failed = false;
+
 	pg_logging_init(argv[0]);
 	pg_logging_set_level(PG_LOG_WARNING);
 	progname = get_progname(argv[0]);
@@ -2062,10 +2063,13 @@ main(int argc, char **argv)
 	start_standby_server(&opt, true);
 
 	/* Check if the standby server is ready for logical replication */
-	check_subscriber(dbinfo);
+	sub_failed = check_subscriber(dbinfo);
 
 	/* Check if the primary server is ready for logical replication */
-	check_publisher(dbinfo);
+	pub_failed = check_publisher(dbinfo);
+
+	if (pub_failed || sub_failed)
+		exit(1);
 
 	/*
 	 * Stop the target server. The recovery process requires that the server
-- 
2.30.2

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Fri, 7 Jun 2024 09:17:59 +0000

Dear Euler,

Thanks for making the follow-up patch! I was looking forward to your updates.
I think this patch set is the solution for the found buildfarm error. However,
there are remained claims raised by others. You should reply what you think for
them. At least:

1) There are some misleading messages [1]. I think v3-0005 patch set can solve
   the issue.
2) pg_createsubscriber may fail If the primary has subscriptions [2]. IIUC
   possible approaches are A)"keep subscriptions disabled at the end",
   B)"by default drop the pre-existing subscriptions",
   C) "do nothing, just document the risk".

> replication slots on the standby. I think we should also remove them too

I also considered around here, but it might be difficult to predict the
expectation by users. Can we surely say that it's not intentional one? Regarding
the failover slot, it is OK because that's meaningful only on the standby,
but not sure other slots. I personally think we can keep current spec, but
how do other think? 

Below parts are comments for each patches.

0001
Basically LGTM. I was bit confused because the default timeout is not set, but
it seemed to follow the suggestion by Tomas [3].

0002
If you want to improve the commit message, please add that sync_replication_slots
is disabled during the conversion.

0003
Confirmed it followed the discussion [4].

0004
Basically LGTM.

Other minor comments are included by the attached diff file. It contains changes
to follow conventions and pgindent/pgperltidy.

[1]: https://www.postgresql.org/message-id/CAA4eK1J2fAvsJ2HihbWJ_GxETd6sdqSMrZdCVJEutRZRpm1MEQ%40mail.gmail.com
[2]: https://www.postgresql.org/message-id/CANhcyEWvimA1-f6hSrA%3D9qkfR5SonFb56b36M%2B%2BvT%3DLiFj%3D76g%40mail.gmail.com
[3]: https://www.postgresql.org/message-id/5d5dd4cd-6359-4109-88e8-c8e13035ae16%40enterprisedb.com
[4]: https://www.postgresql.org/message-id/CAA4eK1LZxYxcbeiOn3Q5hjXVtZKhJWj-fQtndAeTCvZrPev8BA%40mail.gmail.com

Best Regards,
Hayato Kuroda
FUJITSU LIMITED

---

From: Peter Eisentraut <[email-masked]>
Date: Mon, 17 Jun 2024 13:04:52 +0200

> it already failed in any subscriber check.

I have committed 0001, 0002, and 0003.  Let's keep an eye on the 
buildfarm to see if that stabilizes things.  So far it looks good.

For 0004, I suggest inverting the result values from check_publisher() 
and create_subscriber() so that it returns true if the check is ok.

---

From: Peter Eisentraut <[email-masked]>
Date: Mon, 17 Jun 2024 13:08:43 +0200

> to follow conventions and pgindent/pgperltidy.

I have included some of your changes in the patches from Euler that I 
committed today.  The 0004 patch might get rerolled by Euler, so he 
could include your relevant changes there.

I'm not sure about moving the sync_replication_slots setting around.  It 
seems to add a lot of work for no discernible gain?

The pgperltidy changes are probably better done separately, because they 
otherwise make quite a mess of the patch.  Maybe we'll just do that once 
more before we branch.

---

From: "Euler Taveira" <[email-masked]>
Date: Tue, 18 Jun 2024 04:23:52 -0300

Thanks.

> and create_subscriber() so that it returns true if the check is ok.

Yeah, the order doesn't matter after the commit b963913826. I thought about
changing the order but I didn't; I provided a minimal patch. Since you think
it is an improvement, I'm attaching another patch with this change.

--
Euler Taveira
EDB   https://www.enterprisedb.com/
From ec96586fa35be3f855fb8844a197f374ac82a622 Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email-masked]>
Date: Fri, 24 May 2024 14:29:06 -0300
Subject: [PATCH] Check both servers before exiting

The order it calls the check functions don't matter after commit
b96391382626306c301b67cbd2d28313d96741f3 (because there is no
primary_slot_name check anymore) so check the publisher first.
---
 src/bin/pg_basebackup/pg_createsubscriber.c | 28 ++++++++++++---------
 1 file changed, 16 insertions(+), 12 deletions(-)

index 5499e6d96a..26b4087ff7 100644
 
 	pg_free(wal_level);
 
-	if (failed)
-		exit(1);
+	return failed;
 }
 
 /*
 
-	if (failed)
-		exit(1);
+	return failed;
 }
 
 /*
 
 	char		pidfile[MAXPGPATH];
 
+	bool		pub_failed = false;
+	bool		sub_failed = false;
+
 	pg_logging_init(argv[0]);
 	pg_logging_set_level(PG_LOG_WARNING);
 	progname = get_progname(argv[0]);
@@ -2061,11 +2062,14 @@ main(int argc, char **argv)
 	pg_log_info("starting the standby with command-line options");
 	start_standby_server(&opt, true);
 
-	/* Check if the standby server is ready for logical replication */
 
 	/*

---

From: Noah Misch <[email-masked]>
Date: Sat, 22 Jun 2024 23:21:57 -0700

> commit d44032d

> +	appendPQExpBuffer(buf, " dbname=%s", dbname);

pg_createsubscriber fails on a dbname containing a space.  Use
appendConnStrVal() here and for other params in get_sub_conninfo().  See the
CVE-2016-5424 commits for more background.  For one way to test this scenario,
see generate_db() in the pg_upgrade test suite.

> +	char	   *slot_name_esc;
> +	slot_name_esc = PQescapeLiteral(conn, slot_name, strlen(slot_name));
> +
> +	appendPQExpBuffer(str,
> +					  "SELECT lsn FROM pg_catalog.pg_create_logical_replication_slot(%s, 'pgoutput', false, false, false)",

This is passing twophase=false, but the patch does not mention prepared
transactions.  Is the intent to not support workloads containing prepared
transactions?  If so, the documentation should say that, and the tool likely
should warn on startup if max_prepared_transactions != 0.

> +					  ipubname_esc);

This tool's documentation says it "guarantees that no transaction will be
lost."  I tried to determine whether achieving that will require something
like the fix from
https://postgr.es/m/flat/[email-masked].
(Not exactly the fix from that thread, since that thread has not discussed the
FOR ALL TABLES version of its race condition.)  I don't know.  On the one
hand, pg_createsubscriber benefits from creating a logical slot after creating
the publication.  That snapbuild.c process will wait for running XIDs.  On the
other hand, an INSERT/UPDATE/DELETE acquires its RowExclusiveLock and builds
its relcache entry before assigning an XID, so perhaps the snapbuild.c process
isn't enough to prevent that thread's race condition.  What do you think?

---

From: Amit Kapila <[email-masked]>
Date: Mon, 24 Jun 2024 12:08:59 +0530

> should warn on startup if max_prepared_transactions != 0.

The other point to note in this regard is that if we don't support
two_phase in the beginning during subscription/slot setup, users won't
be able to change it as we don't yet support changing it via alter
subscription (though the patch to alter two_pc is proposed for PG18).

-- 
With Regards,
Amit Kapila.

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Mon, 24 Jun 2024 06:47:54 +0000

Dear Noah,

> see generate_db() in the pg_upgrade test suite.

Thanks for pointing out. I made a fix patch. Test code was also modified accordingly.

> > +					  "SELECT lsn FROM
> pg_catalog.pg_create_logical_replication_slot(%s, 'pgoutput', false, false, false)",
> 

IIUC, We decided because it is a default behavior of logical replication. See [1].
+1 for improving a documentation, but not sure it is helpful for adding output.
I want to know opinions from others.

> isn't enough to prevent that thread's race condition.  What do you think?

IIUC, documentation just intended to say that a type of replication will be
switched from stream to logical one, at the certain point. Please give sometime
for analyzing.

[1]: https://www.postgresql.org/message-id/270ad9b8-9c46-40c3-b6c5-3d25b91d3a7d%40app.fastmail.com

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/

---

From: Amit Kapila <[email-masked]>
Date: Mon, 24 Jun 2024 17:20:21 +0530

> isn't enough to prevent that thread's race condition.  What do you think?

I am not able to imagine how the race condition discussed in the
thread you quoted can impact this patch. The problem discussed is
mainly the interaction when we are processing the changes in logical
decoding w.r.t concurrent DDL (Alter Publication ... Add Table). The
problem happens because we use the old cache state. I am missing your
point about the race condition mentioned in the thread you quoted with
snapbuild.c. Can you please elaborate a bit more?

-- 
With Regards,
Amit Kapila.

---

From: Noah Misch <[email-masked]>
Date: Mon, 24 Jun 2024 15:08:45 -0700

> > its relcache entry before assigning an XID, so perhaps the snapbuild.c process

Correction: it doesn't matter how the original INSERT/UPDATE/DELETE builds its
relcache entry, just how pgoutput of the change builds the relcache entry from
the historic snapshot.

> problem happens because we use the old cache state.

Right.  Taking the example from
http://postgr.es/m/[email-masked], LSNs
between what that mail calls 4) and 5) are not safely usable as start points.
pg_createsubscriber evades that thread's problem if the consistent_lsn it
passes to pg_replication_origin_advance() can't be in a bad-start-point LSN
span.  I cautiously bet the snapbuild.c process achieves that:

> I am missing your

When pg_createsubscriber calls pg_create_logical_replication_slot(), the key
part starts at:

	/*
	 * If caller needs us to determine the decoding start point, do so now.
	 * This might take a while.
	 */
	if (find_startpoint)
		DecodingContextFindStartpoint(ctx);

Two factors protect pg_createsubscriber.  First, (a) CREATE PUBLICATION
committed before pg_create_logical_replication_slot() started.  Second, (b)
DecodingContextFindStartpoint() waits for running XIDs to complete, via the
process described at the snapbuild.c "starting up in several stages" diagram.
Hence, the consistent_lsn is not in a bad-start-point LSN span.  It's fine
even if the original INSERT populated all caches before CREATE PUBLICATION
started and managed to assign an XID only after consistent_lsn.  From the
pgoutput perspective, that's indistinguishable from the transaction starting
at its first WAL record, after consistent_lsn.  The linked "long-standing data
loss bug in initial sync of logical replication" thread doesn't have (a),
hence its bug.  How close is that to accurate?

Thanks,
nm

---

From: "Euler Taveira" <[email-masked]>
Date: Tue, 25 Jun 2024 22:51:31 -0300

> Thanks for pointing out. I made a fix patch. Test code was also modified accordingly.

Patch looks good to me. I have one suggestion

-# Mostly Ported from 002_pg_upgrade.pl, but this returns a generated dbname.
+# Extracted from 002_pg_upgrade.pl.

and 2 small fixes:

-   'logical replication works on database $db1');
+   "logical replication works on database $db1");

-is($result, qq(row 1), 'logical replication works on database $db2');
+is($result, qq(row 1), "logical replication works on database $db2");

> I want to know opinions from others.

Documentation says

When two-phase commit is enabled, prepared transactions are sent to the
subscriber at the time of PREPARE TRANSACTION, and are processed as two-phase
transactions on the subscriber too. Otherwise, prepared transactions are sent to
the subscriber only when committed, and are then processed immediately by the
subscriber.

Hence, the replication should be working for prepared transactions even if it
created the slot with twophase = false. IIRC the user won't be able to change it
later. As Amit said in a previous email, once the command
ALTER SUBSCRIPTION ... SET (two_phase = on) is supported, users can change it
after running pg_createsubscriber. The other option is to add a command-line
option to enable or disable it.

---

From: Amit Kapila <[email-masked]>
Date: Wed, 26 Jun 2024 09:28:46 +0530

> option to enable or disable it.

Yeah, it is a good idea to add a new option for two_phase but that
should be done in the next version. For now, I suggest updating the
docs and probably raising a warning (if max_prepared_transactions !=
0) as suggested by Noah. This WARNING is useful because one could
expect that setting max_prepared_transactions != 0 means apply will
happen at prepare time after the subscriber is created by this tool.
The WARNING will be useful even if we support two_phase option as the
user may have set the non-zero value of max_prepared_transactions but
didn't use two_phase option.

-- 
With Regards,
Amit Kapila.

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Wed, 26 Jun 2024 06:04:55 +0000

Dear Amit, Euler,

Thanks for giving comment! 0001 was modified per suggestions.

> didn't use two_phase option.

I also think it should be tunable, in PG18+. 0002 adds a description in doc.
Also, this executable warns if the max_prepared_transactions != 0 on the
publisher. Subscriber-side is not checked now because I don't think it is helpful,
but it can be easily added. I did not add test for that, because current test code
does not check outputs.

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/

---

From: Amit Kapila <[email-masked]>
Date: Fri, 28 Jun 2024 10:41:42 +0530

On Wed, Jun 26, 2024 at 11:35 AM Hayato Kuroda (Fujitsu)
<[email-masked]> wrote:
> publisher.

I have slightly adjusted the doc update and warning message for the
0002 patch. Please see attached and let me know what do you think.

-- 
With Regards,
Amit Kapila.

---

From: Alexander Lakhin <[email-masked]>
Date: Sun, 30 Jun 2024 16:00:00 +0300

Hello Peter and Euler,

17.06.2024 14:04, Peter Eisentraut wrote:
> I have committed 0001, 0002, and 0003.  Let's keep an eye on the buildfarm to see if that stabilizes things.  So far 
> it looks good.
>
> For 0004, I suggest inverting the result values from check_publisher() and create_subscriber() so that it returns true 
> if the check is ok.

As a recent buildfarm failure [1] shows, that test addition introduced
new instability:
### Starting node "node_s"
# Running: pg_ctl -w -D 
/home/bf/bf-build/piculet/HEAD/pgsql.build/testrun/pg_basebackup/040_pg_createsubscriber/data/t_040_pg_createsubscriber_node_s_data/pgdata 
-l 
/home/bf/bf-build/piculet/HEAD/pgsql.build/testrun/pg_basebackup/040_pg_createsubscriber/log/040_pg_createsubscriber_node_s.log 
-o --cluster-name=node_s start
waiting for server to start.... done
server started
# Postmaster PID for node "node_s" is 416482
error running SQL: 'psql:<stdin>:1: ERROR:  skipping slot synchronization as the received slot sync LSN 0/30047F0 for 
slot "failover_slot" is ahead of the standby position 0/3004708'
while running 'psql -XAtq -d port=51506 host=/tmp/pqWohdD5Qj dbname='postgres' -f - -v ON_ERROR_STOP=1' with sql 'SELECT 
pg_sync_replication_slots()' at /home/bf/bf-build/piculet/HEAD/pgsql/src/test/perl/PostgreSQL/Test/Cluster.pm line 2126.

I could reproduce this failure with:
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -517,6 +517,7 @@ WalReceiverMain(char *startup_data, size_t startup_data_len)
                       * let the startup process and primary server know about
                       * them.
                       */
+pg_usleep(300000);
                      XLogWalRcvFlush(false, startpointTLI);

make -s check -C src/bin/pg_basebackup/ PROVE_TESTS="t/040*"

# +++ tap check in src/bin/pg_basebackup +++
t/040_pg_createsubscriber.pl .. 22/? # Tests were run but no plan was declared and done_testing() was not seen.
# Looks like your test exited with 29 just after 23.
t/040_pg_createsubscriber.pl .. Dubious, test returned 29 (wstat 7424, 0x1d00)
All 23 subtests passed

Test Summary Report
-------------------
t/040_pg_createsubscriber.pl (Wstat: 7424 Tests: 23 Failed: 0)
   Non-zero exit status: 29
   Parse errors: No plan found in TAP output
Files=1, Tests=23,  4 wallclock secs ( 0.01 usr  0.01 sys +  0.49 cusr  0.44 csys =  0.95 CPU)

Moreover, this test may suffer from autovacuum:
echo "
autovacuum_naptime = 1
autovacuum_analyze_threshold = 1
" > /tmp/temp.config
TEMP_CONFIG=/tmp/temp.config make -s check -C src/bin/pg_basebackup/ PROVE_TESTS="t/040*"

# +++ tap check in src/bin/pg_basebackup +++
t/040_pg_createsubscriber.pl .. 24/?
#   Failed test 'failover slot is synced'
#   at t/040_pg_createsubscriber.pl line 273.
#          got: ''
#     expected: 'failover_slot'
t/040_pg_createsubscriber.pl .. 28/? # Looks like you failed 1 test of 33.
t/040_pg_createsubscriber.pl .. Dubious, test returned 1 (wstat 256, 0x100)
Failed 1/33 subtests

[1] https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=piculet&dt=2024-06-28%2004%3A42%3A48

Best regards,
Alexander

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Wed, 3 Jul 2024 05:12:25 +0000

Dear Alexander and other hackers,

> /home/bf/bf-build/piculet/HEAD/pgsql.build/testrun/pg_basebackup/040_pg_c
> reatesubscriber/data/t_040_pg_createsubscriber_node_s_data/pgdata
> -l
> /home/bf/bf-build/piculet/HEAD/pgsql.build/testrun/pg_basebackup/040_pg_c
> reatesubscriber/log/040_pg_createsubscriber_node_s.log
> while running 'psql -XAtq -d port=51506 host=/tmp/pqWohdD5Qj
> dbname='postgres' -f - -v ON_ERROR_STOP=1' with sql 'SELECT
> pg_sync_replication_slots()' at
> /home/bf/bf-build/piculet/HEAD/pgsql/src/test/perl/PostgreSQL/Test/Cluster
> .pm line 2126.
> 

I thought it was not related with the feature of pg_createsubscriber. It was
related with the slotsync function. In the first place, the error message was
output with the below backtrace.

```
synchronize_one_slot
synchronize_slots
pg_sync_replication_slots
```

And I found the error could be reproduced by the attached script with the source
modification by Alexander [1]. According to my analysis, this could be caused when
1) a replication slot is created with failover = true and XLOG_RUNNING_XACT record is generated, and
2) pg_sync_replication_slots() is called *before* the record is flushed on the
standby. To stabilize the test, we can call wait_for_replay_catchup() after the slot
creation on the primary.

> t/040_pg_createsubscriber.pl .. Dubious, test returned 1 (wstat 256, 0x100)

Thanks for reporting. I could reproduce the failure, and some BF animals said NG.
According to them, the root cause seemed that slot synchronization was failed.

```
LOG:  statement: SELECT pg_sync_replication_slots()
LOG:  could not sync slot "failover_slot" as remote slot precedes local slot
DETAIL:  Remote slot has LSN 0/3006890 and catalog xmin 743, but local slot has LSN 0/3006890 and catalog xmin 744.
```

Based on that, I considered a scenario why the slot could not be synchronized.
I felt this was not caused by the pg_createsubscriber.

1. At initial stage, the xmin of the physical slot is 743, and nextXid of the
   primary is also 743.
2. Autovacuum worker starts a new transaction. nextXid is incremented to 744.
3. Tries to creates a logical replication slot with failover=true *before the
   transaction at step2 is replicated to the standby*.
4. While creating the slot, the catalog_xmin must be determined.
   The initial candidate is nextXid (= 744), but the oldest xmin of replication
   slots (=743) is used if it is older than nextXid. So 743 is chosen in this case.
   This operaion is done in CreateInitDecodingContext()->GetOldestSafeDecodingContext().
5. After that, the transaction at step2 is reached to the standby node and it
   updates the nextXid.
6. Finally runs pg pg_sync_replication_slots() on the standby. It finds a failover
   slot on the primary and tries to create on the standby. However, the
   catalog_xmin on the primary (743) is older than the nextXid of the standby (744)
   so that it skips to create a slot.

To avoid the issue, we can disable the autovacuuming while testing.

# Descriptions for attached files

An attached script can be used to reproduce the first failure without pg_createsubscriber.
It requires to modify the code like [1].

0001 patch can reduce the wait time of test. See [3]. I know the approach is bit hacky but
it worked.
0002 patch waits until the WAL record is replicated. This fixes a first failure.
0003 patch disables autovacuum for node_p and node_s. I think node_p is enough, but did
like that just in case. This fixes a second failure.

[1]: https://www.postgresql.org/message-id/0dffca12-bf17-4a7a-334d-225569de5e6e%40gmail.com
[2]: https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=adder&dt=2024-07-02%2008%3A45%3A39
[3]: https://www.postgresql.org/message-id/OSBPR01MB25521B15BF950D2523BBE143F5D32%40OSBPR01MB2552.jpnprd01.prod.outlook.com

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/

---

From: Amit Kapila <[email-masked]>
Date: Wed, 3 Jul 2024 11:50:40 +0530

> To avoid the issue, we can disable the autovacuuming while testing.

Your analysis looks correct to me. The test could fail due to
autovacuum. See the following comment in
040_standby_failover_slots_sync.

# Disable autovacuum to avoid generating xid during stats update as otherwise
# the new XID could then be replicated to standby at some random point making
# slots at primary lag behind standby during slot sync.
$publisher->append_conf('postgresql.conf', 'autovacuum = off');

> like that just in case. This fixes a second failure.

Disabling on the primary node should be sufficient. Let's do the
minimum required to stabilize this test.

-- 
With Regards,
Amit Kapila.

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Wed, 3 Jul 2024 07:12:38 +0000

Dear Amit,

> $publisher->append_conf('postgresql.conf', 'autovacuum = off');

Oh, I could not find the comment. I felt it should be added even in
040_pg_createsubscriber.pl. Done.

> minimum required to stabilize this test.

+1, removed.

PSA new version. 0001 has not been changed yet. A comment was added
in 0002 to clarify why we must wait. For 0003, a comment was added and
setting for standby was reverted.

Best Regards,
Hayato Kuroda
FUJITSU LIMITED
https://www.fujitsu.com/

---

From: Amit Kapila <[email-masked]>
Date: Wed, 3 Jul 2024 16:32:12 +0530

> setting for standby was reverted.

Pushed 0002 and 0003. Let's wait for a discussion on 0001.

-- 
With Regards,
Amit Kapila.

---

From: Alexander Lakhin <[email-masked]>
Date: Tue, 9 Jul 2024 14:00:00 +0300

Hello Amit and Kuroda-san,

> Pushed 0002 and 0003. Let's wait for a discussion on 0001.

Please look at another failure of the test [1]:
[13:28:05.647](2.460s) not ok 26 - failover slot is synced
[13:28:05.648](0.001s) #   Failed test 'failover slot is synced'
#   at /home/bf/bf-build/skink-master/HEAD/pgsql/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl line 307.
[13:28:05.648](0.000s) #          got: ''
#     expected: 'failover_slot'

with 040_pg_createsubscriber_node_s.log containing:
2024-07-08 13:28:05.369 UTC [3985464][client backend][0/2:0] LOG: statement: SELECT pg_sync_replication_slots()
2024-07-08 13:28:05.557 UTC [3985464][client backend][0/2:0] LOG: could not sync slot "failover_slot" as remote slot 
precedes local slot
2024-07-08 13:28:05.557 UTC [3985464][client backend][0/2:0] DETAIL:  Remote slot has LSN 0/30047B8 and catalog xmin 
743, but local slot has LSN 0/30047B8 and catalog xmin 744.

I could not reproduce it locally, but I've discovered that that subtest
somehow depends on pg_createsubscriber executed for the
'primary contains unmet conditions on node P' check. For example with this
test modification:
@@ -249,7 +249,7 @@ command_fails(
          $node_p->connstr($db1), '--socket-directory',
          $node_s->host, '--subscriber-port',
          $node_s->port, '--database',
-        $db1, '--database',
+        'XXX', '--database',
          $db2
      ],
      'primary contains unmet conditions on node P');

I see the same failure:
2024-07-09 10:19:43.284 UTC [938890] 040_pg_createsubscriber.pl LOG:  statement: SELECT pg_sync_replication_slots()
2024-07-09 10:19:43.292 UTC [938890] 040_pg_createsubscriber.pl LOG:  could not sync slot "failover_slot" as remote slot 
precedes local slot
2024-07-09 10:19:43.292 UTC [938890] 040_pg_createsubscriber.pl DETAIL:  Remote slot has LSN 0/3004780 and catalog xmin 
743, but local slot has LSN 0/3004780 and catalog xmin 744.

Thus maybe even a normal pg_createsubscriber run can affect the primary
server (it's catalog xmin) differently?

One difference I found in the logs, is that the skink failure's
regress_log_040_pg_createsubscriber contains:
pg_createsubscriber: error: publisher requires 2 wal sender processes, but only 1 remain

Though for a successful run I see locally (I can't find logs of
successful test runs on skink):
pg_createsubscriber: error: publisher requires 2 wal sender processes, but only 0 remain

[1] https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=skink&dt=2024-07-08%2013%3A16%3A35

Best regards,
Alexander

---

From: "Euler Taveira" <[email-masked]>
Date: Wed, 10 Jul 2024 08:20:50 -0300

> #     expected: 'failover_slot'

I'm wondering if the attached patch is sufficient to move the restart_lsn
forward. I experimented several lightweight ideas but none works. BTW the steps
to create the failover slot here is similar 040_standby_failover_slots_sync.pl.
I don't have a clue why it is failing for this one.

--
Euler Taveira
EDB   https://www.enterprisedb.com/
diff --git a/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
index 74b90d9a913..232dbbbc55e 100644
--- a/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
+++ b/src/bin/pg_basebackup/t/040_pg_createsubscriber.pl
@@ -297,6 +297,17 @@ $node_p->safe_psql($db1,
 	"SELECT pg_create_logical_replication_slot('$fslotname', 'pgoutput', false, false, true)"
 );
 $node_s->start;
+$node_p->safe_psql($db1, qq(
+	SELECT pg_logical_emit_message(true, 'a', '1');
+	CHECKPOINT;
+));
+$node_p->safe_psql($db1, qq(
+	SELECT *
+	FROM pg_logical_slot_get_binary_changes('$fslotname', NULL, NULL,
+		'proto_version', '1',
+		'publication_names', 'dummy',
+		'messages', 'true');
+));
 # Wait for the standby to catch up so that the standby is not lagging behind
 # the failover slot.
 $node_p->wait_for_replay_catchup($node_s);

---

From: Amit Kapila <[email-masked]>
Date: Thu, 11 Jul 2024 15:51:16 +0530

> server (it's catalog xmin) differently?

Yes, pg_createsubscriber can affect the primary server's catalog xmin
because it starts the standby server that can send HSFeedback (See
XLogWalRcvSendHSFeedback()), which can advance the physical slot's
xmin corresponding the following Insert in the test:

# Insert another row on node P and wait node S to catch up
$node_p->safe_psql($db1, "INSERT INTO tbl1 VALUES('second row')");
$node_p->wait_for_replay_catchup($node_s);

In the success case, pg_createsubscriber is able to send HSFeedback
and in the failure case, it won't. We can see the following logs in
040_pg_createsubscriber_node_p.log:

2024-07-08 13:28:00.872 UTC [3982331][walsender][:0] FATAL:  the
database system is starting up
2024-07-08 13:28:00.875 UTC [3982328][startup][:0] LOG:  database
system was shut down at 2024-07-08 13:28:00 UTC
2024-07-08 13:28:01.105 UTC [3981996][postmaster][:0] LOG:  database
system is ready to accept connections

This shows that when the test  'primary contains unmet conditions on
node P' starts the standby server the corresponding primary node was
not ready because we just restarted node_p before that test and didn't
ensure that the node_p is up and ready to accept connections before
starting the pg_createsubscriber test.

Even in the successful cases where the standby is able to connect to
primary for test 'primary contains unmet conditions on node P', there
is no guarantee that xmin of the physical slot will be updated at
least, we don't have anything in the test to ensure the same.

Now as before creating logical replication, we didn't ensure that the
physical slot's xmin has been caught up to the latest value, the test
can lead to failure like: "Remote slot has LSN 0/3004780 and catalog
xmin 743, but local slot has LSN 0/3004780 and catalog xmin 744".

The xmin on standby could have been advanced due to the following Insert:
# Insert another row on node P and wait node S to catch up
$node_p->safe_psql($db1, "INSERT INTO tbl1 VALUES('second row')");
$node_p->wait_for_replay_catchup($node_s);

We don't wait for the xmin to catch up corresponding to this insert
and I don't know if there is a way to do that. So, we should move this
Insert to after the call to pg_sync_replication_slots(). It won't
impact the general test of pg_createsubscriber.

Thanks to Hou-San for helping me in the analysis of this BF failure.

-- 
With Regards,
Amit Kapila.

---

From: Alexander Lakhin <[email-masked]>
Date: Thu, 11 Jul 2024 20:00:00 +0300

Hello Amit and Hou-San,

> Thanks to Hou-San for helping me in the analysis of this BF failure.

Thank you for investigating that issue!

May I ask you to look at another failure of the test occurred today [1]?
The failure log contains:
recovery_target_lsn = '0/30098D0'
pg_createsubscriber: starting the subscriber
...
pg_createsubscriber: server was started
pg_createsubscriber: waiting for the target server to reach the consistent state
...
2024-07-11 07:40:10.837 UTC [2948531][postmaster][:0] LOG:  received fast shutdown request

Though what I'm seeing after a successful run is:
recovery_target_lsn = '0/3009860'
pg_createsubscriber: starting the subscriber
...
pg_createsubscriber: server was started
pg_createsubscriber: waiting for the target server to reach the consistent state
...
2024-07-11 15:19:40.733 UTC [207517] 040_pg_createsubscriber.pl LOG:  statement: SELECT pg_catalog.pg_is_in_recovery()
2024-07-11 15:19:41.635 UTC [207514] LOG:  recovery stopping after WAL location (LSN) "0/3009860"
2024-07-11 15:19:41.635 UTC [207514] LOG:  redo done at 0/3009860 system usage: CPU: user: 0.00 s, system: 0.00 s, 
elapsed: 21.00 s

I've managed to reproduce the failure locally. With the bgwriter mod:
-#define LOG_SNAPSHOT_INTERVAL_MS 15000
+#define LOG_SNAPSHOT_INTERVAL_MS 150

and wal_debug=on, when running 5 test instances with parallel, I get the
failure with the following log:
recovery_target_lsn = '0/3009A20'
pg_createsubscriber: starting the subscriber

2024-07-11 14:40:04.551 UTC [205589:72][startup][33/0:0] LOG:  REDO @ 0/30099E8; LSN 0/3009A20: prev 0/30099B0; xid 0; 
len 24 - Standby/RUNNING_XACTS: nextXid 747 latestCompletedXid 746 oldestRunningXid 747
# ^^^ the last REDO record in the log
...
pg_createsubscriber: server was started
pg_createsubscriber: waiting for the target server to reach the consistent state
...
pg_createsubscriber: server was stopped
pg_createsubscriber: error: recovery timed out
...
[14:43:06.011](181.800s) not ok 30 - run pg_createsubscriber on node S
[14:43:06.012](0.000s)
[14:43:06.012](0.000s) #   Failed test 'run pg_createsubscriber on node S'
#   at t/040_pg_createsubscriber.pl line 356.

$ pg_waldump -p src/bin/pg_basebackup_1/tmp_check/t_040_pg_createsubscriber_node_s_data/pgdata/pg_wal/ 
000000010000000000000003 000000010000000000000003 | tail -2
rmgr: Standby     len (rec/tot):     50/    50, tx:          0, lsn: 0/030099B0, prev 0/03009948, desc: RUNNING_XACTS 
nextXid 747 latestCompletedXid 746 oldestRunningXid 747
rmgr: Standby     len (rec/tot):     50/    50, tx:          0, lsn: 0/030099E8, prev 0/030099B0, desc: RUNNING_XACTS 
nextXid 747 latestCompletedXid 746 oldestRunningXid 747

Whilst
$ pg_waldump -p src/bin/pg_basebackup_1/tmp_check/t_040_pg_createsubscriber_node_p_data/pgdata/pg_wal/ 
000000010000000000000003 000000010000000000000003 | tail
rmgr: Standby     len (rec/tot):     50/    50, tx:          0, lsn: 0/030099B0, prev 0/03009948, desc: RUNNING_XACTS 
nextXid 747 latestCompletedXid 746 oldestRunningXid 747
rmgr: Standby     len (rec/tot):     50/    50, tx:          0, lsn: 0/030099E8, prev 0/030099B0, desc: RUNNING_XACTS 
nextXid 747 latestCompletedXid 746 oldestRunningXid 747
rmgr: Heap2       len (rec/tot):     60/    60, tx:        747, lsn: 0/03009A20, prev 0/030099E8, desc: NEW_CID rel: 
1663/16384/6104, tid: 0/1, cmin: 4294967295, cmax: 0, combo: 4294967295
rmgr: Heap        len (rec/tot):     54/    54, tx:        747, lsn: 0/03009A60, prev 0/03009A20, desc: DELETE xmax: 
747, off: 1, infobits: [KEYS_UPDATED], flags: 0x00, blkref #0: rel 1663/16384/6104 blk 0
rmgr: Transaction len (rec/tot):     78/    78, tx:        747, lsn: 0/03009A98, prev 0/03009A60, desc: INVALIDATION ; 
inval msgs: catcache 49 catcache 46 relcache 0
rmgr: Transaction len (rec/tot):     98/    98, tx:        747, lsn: 0/03009AE8, prev 0/03009A98, desc: COMMIT 
2024-07-11 14:43:05.994561 UTC; relcache init file inval dbid 16384 tsid 1663; inval msgs: catcache 49 catcache 46 
relcache 0
rmgr: Heap2       len (rec/tot):     60/    60, tx:        748, lsn: 0/03009B50, prev 0/03009AE8, desc: NEW_CID rel: 
1663/16385/6104, tid: 0/1, cmin: 4294967295, cmax: 0, combo: 4294967295
rmgr: Heap        len (rec/tot):     54/    54, tx:        748, lsn: 0/03009B90, prev 0/03009B50, desc: DELETE xmax: 
748, off: 1, infobits: [KEYS_UPDATED], flags: 0x00, blkref #0: rel 1663/16385/6104 blk 0
rmgr: Transaction len (rec/tot):     78/    78, tx:        748, lsn: 0/03009BC8, prev 0/03009B90, desc: INVALIDATION ; 
inval msgs: catcache 49 catcache 46 relcache 0
rmgr: Transaction len (rec/tot):     98/    98, tx:        748, lsn: 0/03009C18, prev 0/03009BC8, desc: COMMIT 
2024-07-11 14:43:06.008619 UTC; relcache init file inval dbid 16385 tsid 1663; inval msgs: catcache 49 catcache 46 
relcache 0

[1] https://buildfarm.postgresql.org/cgi-bin/show_log.pl?nm=olingo&dt=2024-07-11%2007%3A25%3A12

Best regards,
Alexander

---

From: "Euler Taveira" <[email-masked]>
Date: Thu, 11 Jul 2024 20:24:36 -0300

> May I ask you to look at another failure of the test occurred today [1]?

Thanks for the report!

You are observing the same issue that Amit explained in [1]. The
pg_create_logical_replication_slot returns the EndRecPtr (see
slot->data.confirmed_flush in DecodingContextFindStartpoint()). EndRecPtr points
to the next record and it is a future position for an idle server. That's why
the recovery takes some time to finish because it is waiting for an activity to
increase the LSN position. Since you modified LOG_SNAPSHOT_INTERVAL_MS to create
additional WAL records soon, the EndRecPtr position is reached rapidly and the
recovery ends quickly.

Hayato proposes a patch [2] to create an additional WAL record that has the same
effect from you little hack: increase the LSN position to allow the recovery
finishes soon. I don't like the solution although it seems simple to implement.
As Amit said if we know the ReadRecPtr, we could use it as consistent LSN. The
problem is that it is used by logical decoding but it is not exposed. [reading
the code...] When the logical replication slot is created, restart_lsn points to
the lastReplayedEndRecPtr (see ReplicationSlotReserveWal()) that is the last
record replayed. Since the replication slots aren't in use, we could use the
restart_lsn from the last replication slot as a consistent LSN.

I'm attaching a patch that implements it.It runs in 6s instead of 26s.

[1] https://www.postgresql.org/message-id/CAA4eK1%2Bp%2B7Ag6nqdFRdqowK1EmJ6bG-MtZQ_54dnFBi%3D_OO5RQ%40mail.gmail.com
[2] https://www.postgresql.org/message-id/OSBPR01MB25521B15BF950D2523BBE143F5D32%40OSBPR01MB2552.jpnprd01.prod.outlook.com

--
Euler Taveira
EDB   https://www.enterprisedb.com/
From a3be65c9bea08d3c7ffd575342a879d37b28b9a8 Mon Sep 17 00:00:00 2001
From: Euler Taveira <[email-masked]>
Date: Thu, 11 Jul 2024 19:59:56 -0300
Subject: [PATCH] pg_createsubscriber: fix slow recovery

If the primary server is idle when you are running pg_createsubscriber,
it used to take some time during recovery. The reason is that it was
using the LSN returned by pg_create_logical_replication_slot as
recovery_target_lsn. This LSN points to "end + 1" record that might not
be available at WAL, hence, the recovery routine waits until some
activity from auxiliary processes write WAL records and once it reaches
the recovery_target_lsn position.

Instead, use restart_lsn from the last replication slot. It points to
the last WAL record that was replayed. Hence, the recovery finishes
soon.

create_logical_replication_slot() does not return LSN so change its
signature.

Discussion: https://www.postgresql.org/message-id/2377319.1719766794%40sss.pgh.pa.us
Discussion: https://www.postgresql.org/message-id/68de6498-0449-a113-dd03-e198dded0bac%40gmail.com
---
 src/bin/pg_basebackup/pg_createsubscriber.c | 56 ++++++++++++++++-----
 1 file changed, 43 insertions(+), 13 deletions(-)

diff --git a/src/bin/pg_basebackup/pg_createsubscriber.c b/src/bin/pg_basebackup/pg_createsubscriber.c
index 21dd50f8089..5ef3f751a5b 100644
--- a/src/bin/pg_basebackup/pg_createsubscriber.c
+++ b/src/bin/pg_basebackup/pg_createsubscriber.c
@@ -86,8 +86,8 @@ static void setup_recovery(const struct LogicalRepInfo *dbinfo, const char *data
 static void drop_primary_replication_slot(struct LogicalRepInfo *dbinfo,
 										  const char *slotname);
 static void drop_failover_replication_slots(struct LogicalRepInfo *dbinfo);
-static char *create_logical_replication_slot(PGconn *conn,
-											 struct LogicalRepInfo *dbinfo);
+static bool create_logical_replication_slot(PGconn *conn,
+											struct LogicalRepInfo *dbinfo);
 static void drop_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo,
 								  const char *slot_name);
 static void pg_ctl_status(const char *pg_ctl_cmd, int rc);
@@ -769,15 +769,47 @@ setup_publisher(struct LogicalRepInfo *dbinfo)
 		create_publication(conn, &dbinfo[i]);
 
 		/* Create replication slot on publisher */
-		if (lsn)
-			pg_free(lsn);
-		lsn = create_logical_replication_slot(conn, &dbinfo[i]);
-		if (lsn != NULL || dry_run)
+		if (create_logical_replication_slot(conn, &dbinfo[i]) || dry_run)
 			pg_log_info("create replication slot \"%s\" on publisher",
 						dbinfo[i].replslotname);
 		else
 			exit(1);
 
+		/*
+		 * Get restart_lsn from the last replication slot. It points to the
+		 * last record replayed. The LSN returned by
+		 * pg_create_logical_replication_slot() should not be used because it
+		 * points to a future position. In case the server is idle while
+		 * running this tool, the recovery will not finish soon because the
+		 * future position was not reached. Hence, obtain the restart_lsn that
+		 * points to a LSN already available at WAL. The recovery parameters
+		 * guarantee that this last replication slot will be applied.
+		 */
+		if ((i == num_dbs - 1) && !dry_run)
+		{
+			PQExpBuffer str = createPQExpBuffer();
+			PGresult   *res;
+
+			appendPQExpBuffer(str,
+							  "SELECT restart_lsn FROM pg_catalog.pg_replication_slots "
+							  "WHERE slot_name = '%s'",
+							  dbinfo[i].replslotname);
+			res = PQexec(conn, str->data);
+			if (PQresultStatus(res) != PGRES_TUPLES_OK)
+			{
+				pg_log_error("could not obtain consistent LSN: %s",
+							 PQresultErrorMessage(res));
+				disconnect_database(conn, true);
+			}
+			else
+			{
+				lsn = pg_strdup(PQgetvalue(res, 0, 0));
+			}
+
+			PQclear(res);
+			destroyPQExpBuffer(str);
+		}
+
 		disconnect_database(conn, false);
 	}
 
@@ -1283,19 +1315,18 @@ drop_failover_replication_slots(struct LogicalRepInfo *dbinfo)
 }
 
 /*
- * Create a logical replication slot and returns a LSN.
+ * Create a logical replication slot.
  *
  * CreateReplicationSlot() is not used because it does not provide the one-row
  * result set that contains the LSN.
  */
-static char *
+static bool
 create_logical_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo)
 {
 	PQExpBuffer str = createPQExpBuffer();
 	PGresult   *res = NULL;
 	const char *slot_name = dbinfo->replslotname;
 	char	   *slot_name_esc;
-	char	   *lsn = NULL;
 
 	Assert(conn != NULL);
 
@@ -1305,7 +1336,7 @@ create_logical_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo)
 	slot_name_esc = PQescapeLiteral(conn, slot_name, strlen(slot_name));
 
 	appendPQExpBuffer(str,
-					  "SELECT lsn FROM pg_catalog.pg_create_logical_replication_slot(%s, 'pgoutput', false, false, false)",
+					  "SELECT * FROM pg_catalog.pg_create_logical_replication_slot(%s, 'pgoutput', false, false, false)",
 					  slot_name_esc);
 
 	pg_free(slot_name_esc);
@@ -1322,10 +1353,9 @@ create_logical_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo)
 						 PQresultErrorMessage(res));
 			PQclear(res);
 			destroyPQExpBuffer(str);
-			return NULL;
+			return false;
 		}
 
-		lsn = pg_strdup(PQgetvalue(res, 0, 0));
 		PQclear(res);
 	}
 
@@ -1334,7 +1364,7 @@ create_logical_replication_slot(PGconn *conn, struct LogicalRepInfo *dbinfo)
 
 	destroyPQExpBuffer(str);
 
-	return lsn;
+	return true;
 }
 
 static void
-- 
2.30.2

---

From: Amit Kapila <[email-masked]>
Date: Mon, 15 Jul 2024 12:16:57 +0530

> additional WAL records soon, the EndRecPtr position is reached rapidly and the

If the recovery ends quickly (which is expected due to reduced
LOG_SNAPSHOT_INTERVAL_MS ) then why do we see "error: recovery timed
out"?

> record replayed.

The last 'lastReplayedEndRecPtr' should be the value of restart_lsn on
standby (when RecoveryInProgress is true) but here we are creating
slots on the publisher/primary, so shouldn't restart_lsn point to
"latest WAL insert pointer"?

-- 
With Regards,
Amit Kapila.

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Wed, 17 Jul 2024 07:53:22 +0000

Dear Alexander, Euler, Amit,

I also analyzed this failure, let me share it. Here, I think events in below
ordering were occurred.

1. Backend created a publication on $db2,
2. BGWriter generated RUNNING_XACT record, then
3. Backend created a replication slot on $db2.

In this case, the recovery_target_lsn is ahead of the RUNNING_XACT record generated
at step 3. Also, since both bgwriter and slot creation mark the record as
*UNIMPORTANT* one, the writer won't start again even after the
LOG_SNAPSHOT_INTERVAL_MS. The rule is written in BackgroundWriterMain():

```
			/*
			 * Only log if enough time has passed and interesting records have
			 * been inserted since the last snapshot.  Have to compare with <=
			 * instead of < because GetLastImportantRecPtr() points at the
			 * start of a record, whereas last_snapshot_lsn points just past
			 * the end of the record.
			 */
			if (now >= timeout &&
				last_snapshot_lsn <= GetLastImportantRecPtr())
			{
				last_snapshot_lsn = LogStandbySnapshot();
				last_snapshot_ts = now;
			}
```

Therefore, pg_createsubscriber waited until a new record was replicated, but no
activities were recorded, causing a timeout. Since this is a timing issue, Alexander
could reproduce the failure with shorter time duration and parallel running.

IIUC, the root cause is that pg_create_logical_replication_slot() returns a LSN
which is not generated yet. So, I think both mine [1] and Euler's approach [2]
can solve the issue. My proposal was to add an extra WAL record after the final
slot creation, and Euler's one was to use a restart_lsn as the recovery_target_lsn.
In case of primary server, restart_lsn is set to the latest WAL insert position and
then RUNNING_XACT record is generated later.

How do you think?

[1]: https://www.postgresql.org/message-id/[email-masked]
[2]: https://www.postgresql.org/message-id/b1f0f8c7-8f01-4950-af77-339df3dc4684%40app.fastmail.com

Best regards,
Hayato Kuroda
FUJITSU LIMITED

---

From: Amit Kapila <[email-masked]>
Date: Wed, 17 Jul 2024 15:40:08 +0530

> could reproduce the failure with shorter time duration and parallel running.

Your analysis sounds correct to me.

> slot creation, and Euler's one was to use a restart_lsn as the recovery_target_lsn.

I don't think it is correct to set restart_lsn as consistent_lsn point
because the same is used to set replication origin progress. Later
when we start the subscriber, the system will use that LSN as a
start_decoding_at point which is the point after which all the commits
will be replicated. So, we will end up incorrectly using restart_lsn
(LSN from where we start reading the WAL) as start_decoding_at point.
How could that be correct?

Now, even if we use restart_lsn as recovery_target_lsn and the LSN
returned by pg_create_logical_replication_slot() as consistent LSN to
set replication progress, that also could lead to data loss because
the subscriber may never get data between restart_lsn value and
consistent LSN value.

-- 
With Regards,
Amit Kapila.

---

From: "Hayato Kuroda (Fujitsu)" <[email-masked]>
Date: Wed, 17 Jul 2024 11:58:47 +0000

Dear Amit,

> Your analysis sounds correct to me.

Okay, so we could have a same picture...

> How could that be correct?

I didn't say we could use restart_lsn as consistent point of logical replication,
but I could agree the approach has issues.

> the subscriber may never get data between restart_lsn value and

You considered the case, e.g., tuples were inserted just after the restart_lsn
but before the RUNNING_XACT record? In this case, yes, the streaming replication
finishes before replicating tuples but logical replication will skip them.
Euler's approach cannot be used as-is.

Best regards,
Hayato Kuroda
FUJITSU LIMITED