From 065d4574940bf931c070f264c00267ecb58db734 Mon Sep 17 00:00:00 2001 From: "Chao Li (Evan)" Date: Wed, 22 Apr 2026 09:14:27 +0800 Subject: [PATCH v3] Allow altering subscription server/connection without resolving old conninfo When a subscription uses a foreign server, GetSubscription() resolves the publisher connection string immediately. That can fail if the subscription owner no longer has USAGE on the server or no longer has a valid user mapping. This becomes a problem for ALTER SUBSCRIPTION commands such as ... SERVER or ... CONNECTION, because they may need to update the subscription away from the broken server/conninfo, but currently fail while trying to resolve the old one first. Fix this by storing the subscription's server OID in the in-memory Subscription object and loading conninfo lazily only when it is actually needed. Add GetSubscriptionConninfo() to fetch the connection string on demand in the subscription memory context. This allows ALTER SUBSCRIPTION ... SERVER and ... CONNECTION to succeed without requiring access to the previous server or user mapping, while preserving the existing behavior for operations that really need to connect to the publisher. Add regression test coverage for both cases. Author: Chao Li Reviewed-by: Ajin Cherian Reviewed-by: Zsolt Parragi Discussion: https://postgr.es/m/D908370F-2695-4231-851D-17179A6A6F2A@gmail.com --- src/backend/catalog/pg_subscription.c | 67 ++++++++++++++----- src/backend/commands/subscriptioncmds.c | 9 ++- .../replication/logical/sequencesync.c | 2 +- src/backend/replication/logical/tablesync.c | 2 +- src/backend/replication/logical/worker.c | 9 +-- src/include/catalog/pg_subscription.h | 2 + src/test/regress/expected/subscription.out | 33 +++++++++ src/test/regress/regress.c | 3 + src/test/regress/sql/subscription.sql | 27 ++++++++ 9 files changed, 130 insertions(+), 24 deletions(-) diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c index 1f1fdc75af6..0f2594a702f 100644 --- a/src/backend/catalog/pg_subscription.c +++ b/src/backend/catalog/pg_subscription.c @@ -72,6 +72,18 @@ GetPublicationsStr(List *publications, StringInfo dest, bool quote_literal) /* * Fetch the subscription from the syscache. + * + * If missing_ok is false, throw an error if the subscription is not found. + * If true, return NULL in that case. + * + * If the subscription uses a foreign server and aclcheck is true, check + * whether the subscription owner has permission on that server and eagerly + * load the connection string into sub->conninfo. + * + * If the subscription uses a foreign server and aclcheck is false, + * sub->conninfo is left NULL until it is needed. Call + * GetSubscriptionConnInfo() to fetch it on demand. That accessor is also safe + * to call when sub->conninfo has already been populated. */ Subscription * GetSubscription(Oid subid, bool missing_ok, bool aclcheck) @@ -118,32 +130,32 @@ GetSubscription(Oid subid, bool missing_ok, bool aclcheck) sub->retaindeadtuples = subform->subretaindeadtuples; sub->maxretention = subform->submaxretention; sub->retentionactive = subform->subretentionactive; + sub->serverid = subform->subserver; + sub->conninfo = NULL; /* Get conninfo */ - if (OidIsValid(subform->subserver)) + if (OidIsValid(sub->serverid)) { - AclResult aclresult; - ForeignServer *server; - - server = GetForeignServer(subform->subserver); - /* recheck ACL if requested */ if (aclcheck) { + AclResult aclresult; + ForeignServer *server; + + server = GetForeignServer(sub->serverid); aclresult = object_aclcheck(ForeignServerRelationId, - subform->subserver, + sub->serverid, subform->subowner, ACL_USAGE); - if (aclresult != ACLCHECK_OK) ereport(ERROR, - (errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), - errmsg("subscription owner \"%s\" does not have permission on foreign server \"%s\"", - GetUserNameFromId(subform->subowner, false), - server->servername))); - } + errcode(ERRCODE_INSUFFICIENT_PRIVILEGE), + errmsg("subscription owner \"%s\" does not have permission on foreign server \"%s\"", + GetUserNameFromId(subform->subowner, false), + server->servername)); - sub->conninfo = ForeignServerConnectionString(subform->subowner, - server); + sub->conninfo = ForeignServerConnectionString(subform->subowner, + server); + } } else { @@ -197,6 +209,31 @@ GetSubscription(Oid subid, bool missing_ok, bool aclcheck) return sub; } +/* + * Return the subscription's connection string, loading it into the + * subscription memory context if necessary. + * + * GetSubscription must be called earlier to set sub->serverid, because ACL + * checks are performed there. + */ +char * +GetSubscriptionConnInfo(Subscription *sub) +{ + MemoryContext oldcxt; + + if (sub->conninfo) + return sub->conninfo; + + Assert(OidIsValid(sub->serverid)); + + oldcxt = MemoryContextSwitchTo(sub->cxt); + sub->conninfo = ForeignServerConnectionString(sub->owner, + GetForeignServer(sub->serverid)); + MemoryContextSwitchTo(oldcxt); + + return sub->conninfo; +} + /* * Return number of subscriptions defined in given database. * Used by dropdb() to check if database can indeed be dropped. diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c index 1e10d9d9a58..2b77f3e5592 100644 --- a/src/backend/commands/subscriptioncmds.c +++ b/src/backend/commands/subscriptioncmds.c @@ -1026,7 +1026,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, true, must_use_password, + wrconn = walrcv_connect(GetSubscriptionConnInfo(sub), true, true, + must_use_password, sub->name, &err); if (!wrconn) ereport(ERROR, @@ -1278,7 +1279,8 @@ AlterSubscription_refresh_seq(Subscription *sub) /* Try to connect to the publisher. */ must_use_password = sub->passwordrequired && !sub->ownersuperuser; - wrconn = walrcv_connect(sub->conninfo, true, true, must_use_password, + wrconn = walrcv_connect(GetSubscriptionConnInfo(sub), true, true, + must_use_password, sub->name, &err); if (!wrconn) ereport(ERROR, @@ -2129,7 +2131,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt, * available. */ must_use_password = sub->passwordrequired && !sub->ownersuperuser; - wrconn = walrcv_connect(stmt->conninfo ? stmt->conninfo : sub->conninfo, + wrconn = walrcv_connect(stmt->conninfo ? stmt->conninfo : + GetSubscriptionConnInfo(sub), true, true, must_use_password, sub->name, &err); if (!wrconn) diff --git a/src/backend/replication/logical/sequencesync.c b/src/backend/replication/logical/sequencesync.c index e2ff8d77b16..2ba521c5a43 100644 --- a/src/backend/replication/logical/sequencesync.c +++ b/src/backend/replication/logical/sequencesync.c @@ -711,7 +711,7 @@ LogicalRepSyncSequences(void) * Establish the connection to the publisher for sequence synchronization. */ LogRepWorkerWalRcvConn = - walrcv_connect(MySubscription->conninfo, true, true, + walrcv_connect(GetSubscriptionConnInfo(MySubscription), true, true, must_use_password, app_name.data, &err); if (LogRepWorkerWalRcvConn == NULL) diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c index eb718114297..d859b0d7311 100644 --- a/src/backend/replication/logical/tablesync.c +++ b/src/backend/replication/logical/tablesync.c @@ -1304,7 +1304,7 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos) * so that synchronous replication can distinguish them. */ LogRepWorkerWalRcvConn = - walrcv_connect(MySubscription->conninfo, true, true, + walrcv_connect(GetSubscriptionConnInfo(MySubscription), true, 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 dd6fc38a41e..e2855b0d077 100644 --- a/src/backend/replication/logical/worker.c +++ b/src/backend/replication/logical/worker.c @@ -5105,7 +5105,8 @@ maybe_reread_subscription(void) * 'parallel' to any other value or the server decides not to stream the * in-progress transaction. */ - if (strcmp(newsub->conninfo, MySubscription->conninfo) != 0 || + if (strcmp(GetSubscriptionConnInfo(newsub), + GetSubscriptionConnInfo(MySubscription)) != 0 || strcmp(newsub->name, MySubscription->name) != 0 || strcmp(newsub->slotname, MySubscription->slotname) != 0 || newsub->binary != MySubscription->binary || @@ -5703,8 +5704,8 @@ run_apply_worker(void) must_use_password = MySubscription->passwordrequired && !MySubscription->ownersuperuser; - LogRepWorkerWalRcvConn = walrcv_connect(MySubscription->conninfo, true, - true, must_use_password, + LogRepWorkerWalRcvConn = walrcv_connect(GetSubscriptionConnInfo(MySubscription), + true, true, must_use_password, MySubscription->name, &err); if (LogRepWorkerWalRcvConn == NULL) @@ -5972,7 +5973,7 @@ SetupApplyOrSyncWorker(int worker_slot) /* Connect to the origin and start the replication. */ elog(DEBUG1, "connecting to publisher using connection string \"%s\"", - MySubscription->conninfo); + GetSubscriptionConnInfo(MySubscription)); /* * Setup callback for syscache so that we know when something changes in diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h index a6a2ad1e49c..4dfc6dfce17 100644 --- a/src/include/catalog/pg_subscription.h +++ b/src/include/catalog/pg_subscription.h @@ -164,6 +164,7 @@ typedef struct Subscription * and the retention duration has not * exceeded max_retention_duration, when * defined */ + Oid serverid; /* Oid of the foreign server, if any */ char *conninfo; /* Connection string to the publisher */ char *slotname; /* Name of the replication slot */ char *synccommit; /* Synchronous commit setting for worker */ @@ -214,6 +215,7 @@ typedef struct Subscription extern Subscription *GetSubscription(Oid subid, bool missing_ok, bool aclcheck); +extern char *GetSubscriptionConnInfo(Subscription *sub); extern void DisableSubscription(Oid subid); extern int CountDBSubscriptions(Oid dbid); diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out index 7e3cabdb93f..c0db7636713 100644 --- a/src/test/regress/expected/subscription.out +++ b/src/test/regress/expected/subscription.out @@ -174,16 +174,49 @@ WARNING: subscription was created, but is not connected HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and alter the subscription to refresh publications. DROP USER MAPPING FOR regress_subscription_user3 SERVER test_server; RESET SESSION AUTHORIZATION; +CREATE SERVER test_server2 FOREIGN DATA WRAPPER test_fdw; +GRANT USAGE ON FOREIGN SERVER test_server2 TO regress_subscription_user3; +SET SESSION AUTHORIZATION regress_subscription_user3; +CREATE USER MAPPING FOR regress_subscription_user3 SERVER test_server2 OPTIONS(user 'bar', password 'secret'); +RESET SESSION AUTHORIZATION; REVOKE USAGE ON FOREIGN SERVER test_server FROM regress_subscription_user3; SET SESSION AUTHORIZATION regress_subscription_user3; -- fail, must connect but lacks USAGE on server, as well as user mapping DROP SUBSCRIPTION regress_testsub6; ERROR: could not connect to publisher when attempting to drop replication slot "dummy": subscription owner "regress_subscription_user3" does not have permission on foreign server "test_server" HINT: Use ALTER SUBSCRIPTION ... DISABLE to disable the subscription, and then use ALTER SUBSCRIPTION ... SET (slot_name = NONE) to disassociate it from the slot. +-- ok, should not need to resolve conninfo for the old broken server +ALTER SUBSCRIPTION regress_testsub6 SERVER test_server2; +SELECT fs.srvname + FROM pg_subscription s + JOIN pg_foreign_server fs ON fs.oid = s.subserver + WHERE s.subname = 'regress_testsub6'; + srvname +-------------- + test_server2 +(1 row) + +DROP USER MAPPING FOR regress_subscription_user3 SERVER test_server2; +RESET SESSION AUTHORIZATION; +REVOKE USAGE ON FOREIGN SERVER test_server2 FROM regress_subscription_user3; +SET SESSION AUTHORIZATION regress_subscription_user3; +-- ok, should not need to resolve conninfo for the current broken server +ALTER SUBSCRIPTION regress_testsub6 CONNECTION 'dbname=regress_doesnotexist2 password=secret'; +RESET SESSION AUTHORIZATION; +SELECT subserver = 0::oid AS uses_conninfo, subconninfo + FROM pg_subscription + WHERE subname = 'regress_testsub6'; + uses_conninfo | subconninfo +---------------+---------------------------------------------- + t | dbname=regress_doesnotexist2 password=secret +(1 row) + +SET SESSION AUTHORIZATION regress_subscription_user3; ALTER SUBSCRIPTION regress_testsub6 SET (slot_name = NONE); DROP SUBSCRIPTION regress_testsub6; SET SESSION AUTHORIZATION regress_subscription_user; REVOKE CREATE ON DATABASE REGRESSION FROM regress_subscription_user3; +DROP SERVER test_server2; DROP SERVER test_server; -- fail, FDW is dependent DROP FUNCTION test_fdw_connection(oid, oid, internal); diff --git a/src/test/regress/regress.c b/src/test/regress/regress.c index 2bcb5559a45..598635b6558 100644 --- a/src/test/regress/regress.c +++ b/src/test/regress/regress.c @@ -31,6 +31,7 @@ #include "executor/executor.h" #include "executor/functions.h" #include "executor/spi.h" +#include "foreign/foreign.h" #include "funcapi.h" #include "mb/pg_wchar.h" #include "miscadmin.h" @@ -735,6 +736,8 @@ PG_FUNCTION_INFO_V1(test_fdw_connection); Datum test_fdw_connection(PG_FUNCTION_ARGS) { + /* Ensure the test fails if no valid user mapping exists. */ + GetUserMapping(PG_GETARG_OID(0), PG_GETARG_OID(1)); PG_RETURN_TEXT_P(cstring_to_text("dbname=regress_doesnotexist user=doesnotexist password=secret")); } diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql index 6c3d9632e8a..73cb251e192 100644 --- a/src/test/regress/sql/subscription.sql +++ b/src/test/regress/sql/subscription.sql @@ -124,6 +124,12 @@ SET SESSION AUTHORIZATION regress_subscription_user3; CREATE SUBSCRIPTION regress_testsub6 SERVER test_server PUBLICATION testpub WITH (slot_name = 'dummy', connect = false); DROP USER MAPPING FOR regress_subscription_user3 SERVER test_server; +RESET SESSION AUTHORIZATION; +CREATE SERVER test_server2 FOREIGN DATA WRAPPER test_fdw; +GRANT USAGE ON FOREIGN SERVER test_server2 TO regress_subscription_user3; +SET SESSION AUTHORIZATION regress_subscription_user3; +CREATE USER MAPPING FOR regress_subscription_user3 SERVER test_server2 OPTIONS(user 'bar', password 'secret'); + RESET SESSION AUTHORIZATION; REVOKE USAGE ON FOREIGN SERVER test_server FROM regress_subscription_user3; SET SESSION AUTHORIZATION regress_subscription_user3; @@ -131,12 +137,33 @@ SET SESSION AUTHORIZATION regress_subscription_user3; -- fail, must connect but lacks USAGE on server, as well as user mapping DROP SUBSCRIPTION regress_testsub6; +-- ok, should not need to resolve conninfo for the old broken server +ALTER SUBSCRIPTION regress_testsub6 SERVER test_server2; +SELECT fs.srvname + FROM pg_subscription s + JOIN pg_foreign_server fs ON fs.oid = s.subserver + WHERE s.subname = 'regress_testsub6'; + +DROP USER MAPPING FOR regress_subscription_user3 SERVER test_server2; +RESET SESSION AUTHORIZATION; +REVOKE USAGE ON FOREIGN SERVER test_server2 FROM regress_subscription_user3; +SET SESSION AUTHORIZATION regress_subscription_user3; + +-- ok, should not need to resolve conninfo for the current broken server +ALTER SUBSCRIPTION regress_testsub6 CONNECTION 'dbname=regress_doesnotexist2 password=secret'; +RESET SESSION AUTHORIZATION; +SELECT subserver = 0::oid AS uses_conninfo, subconninfo + FROM pg_subscription + WHERE subname = 'regress_testsub6'; +SET SESSION AUTHORIZATION regress_subscription_user3; + ALTER SUBSCRIPTION regress_testsub6 SET (slot_name = NONE); DROP SUBSCRIPTION regress_testsub6; SET SESSION AUTHORIZATION regress_subscription_user; REVOKE CREATE ON DATABASE REGRESSION FROM regress_subscription_user3; +DROP SERVER test_server2; DROP SERVER test_server; -- fail, FDW is dependent -- 2.50.1 (Apple Git-155)