v61-0004-Non-replication-connection-and-app_name-change.patch

From 588687ead7ee064479155914878fe20f52a832b0 Mon Sep 17 00:00:00 2001
From: Shveta Malik <shveta.malik@gmail.com>
Date: Thu, 11 Jan 2024 14:17:29 +0530
Subject: [PATCH v61 4/4] Non replication connection and app_name change.

Changes in this patch:
1) Convert replication connection to non-replication one in slotsync worker.
2) Use app_name as {cluster_name}_slotsyncworker in the slotsync worker
connection.
---
 src/backend/commands/subscriptioncmds.c       | 12 +++--
 .../libpqwalreceiver/libpqwalreceiver.c       | 44 +++++++++++++------
 src/backend/replication/logical/slotsync.c    | 14 +++++-
 src/backend/replication/logical/tablesync.c   |  2 +-
 src/backend/replication/logical/worker.c      |  4 +-
 src/backend/replication/walreceiver.c         |  3 +-
 src/include/replication/walreceiver.h         |  5 ++-
 7 files changed, 60 insertions(+), 24 deletions(-)

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index f50be29d99..13da6b1466 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -753,7 +753,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
 
 		/* Try to connect to the publisher. */
 		must_use_password = !superuser_arg(owner) && opts.passwordrequired;
-		wrconn = walrcv_connect(conninfo, true, must_use_password,
+		wrconn = walrcv_connect(conninfo, true /* replication */ ,
+								true, must_use_password,
 								stmt->subname, &err);
 		if (!wrconn)
 			ereport(ERROR,
@@ -904,7 +905,8 @@ AlterSubscription_refresh(Subscription *sub, bool copy_data,
 
 	/* Try to connect to the publisher. */
 	must_use_password = sub->passwordrequired && !sub->ownersuperuser;
-	wrconn = walrcv_connect(sub->conninfo, true, must_use_password,
+	wrconn = walrcv_connect(sub->conninfo, true /* replication */ ,
+							true, must_use_password,
 							sub->name, &err);
 	if (!wrconn)
 		ereport(ERROR,
@@ -1530,7 +1532,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 
 		/* Try to connect to the publisher. */
 		must_use_password = sub->passwordrequired && !sub->ownersuperuser;
-		wrconn = walrcv_connect(sub->conninfo, true, must_use_password,
+		wrconn = walrcv_connect(sub->conninfo, true /* replication */ ,
+								true, must_use_password,
 								sub->name, &err);
 		if (!wrconn)
 			ereport(ERROR,
@@ -1781,7 +1784,8 @@ DropSubscription(DropSubscriptionStmt *stmt, bool isTopLevel)
 	 */
 	load_file("libpqwalreceiver", false);
 
-	wrconn = walrcv_connect(conninfo, true, must_use_password,
+	wrconn = walrcv_connect(conninfo, true /* replication */ ,
+							true, must_use_password,
 							subname, &err);
 	if (wrconn == NULL)
 	{
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index f910a3b103..8aa0103d8f 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -6,6 +6,9 @@
  * loaded as a dynamic module to avoid linking the main server binary with
  * libpq.
  *
+ * Apart from walreceiver, the libpq-specific routines here are now being used
+ * by logical replication workers and slotsync worker as well.
+
  * Portions Copyright (c) 2010-2024, PostgreSQL Global Development Group
  *
  *
@@ -50,7 +53,8 @@ struct WalReceiverConn
 
 /* Prototypes for interface functions */
 static WalReceiverConn *libpqrcv_connect(const char *conninfo,
-										 bool logical, bool must_use_password,
+										 bool replication, bool logical,
+										 bool must_use_password,
 										 const char *appname, char **err);
 static void libpqrcv_check_conninfo(const char *conninfo,
 									bool must_use_password);
@@ -125,7 +129,12 @@ _PG_init(void)
 }
 
 /*
- * Establish the connection to the primary server for XLOG streaming
+ * Establish the connection to the primary server.
+ *
+ * The connection established could be either a replication one or
+ * a non-replication one based on input argument 'replication'. And further
+ * if it is a replication connection, it could be either logical or physical
+ * based on input argument 'logical'.
  *
  * If an error occurs, this function will normally return NULL and set *err
  * to a palloc'ed error message. However, if must_use_password is true and
@@ -136,8 +145,8 @@ _PG_init(void)
  * case.
  */
 static WalReceiverConn *
-libpqrcv_connect(const char *conninfo, bool logical, bool must_use_password,
-				 const char *appname, char **err)
+libpqrcv_connect(const char *conninfo, bool replication, bool logical,
+				 bool must_use_password, const char *appname, char **err)
 {
 	WalReceiverConn *conn;
 	const char *keys[6];
@@ -150,17 +159,26 @@ libpqrcv_connect(const char *conninfo, bool logical, bool must_use_password,
 	 */
 	keys[i] = "dbname";
 	vals[i] = conninfo;
-	keys[++i] = "replication";
-	vals[i] = logical ? "database" : "true";
-	if (!logical)
+
+	/* We can not have logical without replication */
+	if (!replication)
+		Assert(!logical);
+	else
 	{
-		/*
-		 * The database name is ignored by the server in replication mode, but
-		 * specify "replication" for .pgpass lookup.
-		 */
-		keys[++i] = "dbname";
-		vals[i] = "replication";
+		keys[++i] = "replication";
+		vals[i] = logical ? "database" : "true";
+
+		if (!logical)
+		{
+			/*
+			 * The database name is ignored by the server in replication mode,
+			 * but specify "replication" for .pgpass lookup.
+			 */
+			keys[++i] = "dbname";
+			vals[i] = "replication";
+		}
 	}
+
 	keys[++i] = "fallback_application_name";
 	vals[i] = appname;
 	if (logical)
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index df13f16d8d..9be9c16c84 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -934,6 +934,7 @@ ReplSlotSyncWorkerMain(Datum main_arg)
 	char	   *dbname;
 	bool		am_cascading_standby;
 	char	   *err;
+	StringInfoData app_name;
 
 	ereport(LOG, errmsg("replication slot sync worker started"));
 
@@ -966,13 +967,22 @@ ReplSlotSyncWorkerMain(Datum main_arg)
 	 */
 	BackgroundWorkerInitializeConnection(dbname, NULL, 0);
 
+	initStringInfo(&app_name);
+	if (cluster_name[0])
+		appendStringInfo(&app_name, "%s_%s", cluster_name, "slotsyncworker");
+	else
+		appendStringInfo(&app_name, "%s", "slotsyncworker");
+
 	/*
 	 * Establish the connection to the primary server for slots
 	 * synchronization.
 	 */
-	wrconn = walrcv_connect(PrimaryConnInfo, true, false,
-							cluster_name[0] ? cluster_name : "slotsyncworker",
+	wrconn = walrcv_connect(PrimaryConnInfo, false /* replication */,
+							false, false,
+							app_name.data,
 							&err);
+	pfree(app_name.data);
+
 	if (!wrconn)
 		ereport(ERROR,
 				errcode(ERRCODE_CONNECTION_FAILURE),
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 5acab3f3e2..c5c6ac4bac 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -1329,7 +1329,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
 	 * so that synchronous replication can distinguish them.
 	 */
 	LogRepWorkerWalRcvConn =
-		walrcv_connect(MySubscription->conninfo, true,
+		walrcv_connect(MySubscription->conninfo, true /* replication */ , true,
 					   must_use_password,
 					   slotname, &err);
 	if (LogRepWorkerWalRcvConn == NULL)
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 3dea10f9b3..95e7324c8f 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -4518,7 +4518,9 @@ run_apply_worker()
 	must_use_password = MySubscription->passwordrequired &&
 		!MySubscription->ownersuperuser;
 
-	LogRepWorkerWalRcvConn = walrcv_connect(MySubscription->conninfo, true,
+	LogRepWorkerWalRcvConn = walrcv_connect(MySubscription->conninfo,
+											true /* replication */ ,
+											true,
 											must_use_password,
 											MySubscription->name, &err);
 
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index ffacd55e5c..cfe647ec58 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -296,7 +296,8 @@ WalReceiverMain(void)
 	sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
 
 	/* Establish the connection to the primary for XLOG streaming */
-	wrconn = walrcv_connect(conninfo, false, false,
+	wrconn = walrcv_connect(conninfo, true /* replication */ ,
+							false, false,
 							cluster_name[0] ? cluster_name : "walreceiver",
 							&err);
 	if (!wrconn)
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 5e942cb4fc..68ac074274 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -237,6 +237,7 @@ typedef struct WalRcvExecResult
  * returned with 'err' including the error generated.
  */
 typedef WalReceiverConn *(*walrcv_connect_fn) (const char *conninfo,
+											   bool replication,
 											   bool logical,
 											   bool must_use_password,
 											   const char *appname,
@@ -434,8 +435,8 @@ typedef struct WalReceiverFunctionsType
 
 extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
 
-#define walrcv_connect(conninfo, logical, must_use_password, appname, err) \
-	WalReceiverFunctions->walrcv_connect(conninfo, logical, must_use_password, appname, err)
+#define walrcv_connect(conninfo, replication, logical, must_use_password, appname, err) \
+	WalReceiverFunctions->walrcv_connect(conninfo, replication, logical, must_use_password, appname, err)
 #define walrcv_check_conninfo(conninfo, must_use_password) \
 	WalReceiverFunctions->walrcv_check_conninfo(conninfo, must_use_password)
 #define walrcv_get_conninfo(conn) \
-- 
2.34.1