Avoid errors during ALTER SUBSCRIPTION.

Previously, when retrieving the old Subscription object, constructing
the conninfo could encounter an error during
ForeignServerConnectionString(). ACL errors were handled properly, but
other errors could interfere with a user fixing the problem with ALTER
SUBSCRIPTION.

Reported-by: Chao Li <li.evan.chao@gmail.com>
Reviewed-by: Chao Li <li.evan.chao@gmail.com>
Discussion: https://postgr.es/m/D908370F-2695-4231-851D-17179A6A6F2A@gmail.com

diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index 45eff207746346ef7fda57bd20c3c812da3f612f..b5cb301db88a0593ee3e706fbab9cf8c602fb2cb 100644 (file)
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -78,9 +78,15 @@ GetPublicationsStr(List *publications, StringInfo dest, bool quote_literal)
 
 /*
  * Fetch the subscription from the syscache.
+ *
+ * If conninfo_needed is true, conninfo will be constructed, possibly
+ * encountering errors in ForeignServerConnectionString(). Callers not
+ * expecting such errors should pass false, in which case conninfo will be
+ * NULL.
  */
 Subscription *
-GetSubscription(Oid subid, bool missing_ok, bool aclcheck)
+GetSubscription(Oid subid, bool missing_ok, bool conninfo_needed,
+                               bool conninfo_aclcheck)
 {
        HeapTuple       tup;
        Subscription *sub;
@@ -90,6 +96,8 @@ GetSubscription(Oid subid, bool missing_ok, bool aclcheck)
        MemoryContext cxt;
        MemoryContext oldcxt;
 
+       Assert(conninfo_needed || !conninfo_aclcheck);
+
        tup = SearchSysCache1(SUBSCRIPTIONOID, ObjectIdGetDatum(subid));
 
        if (!HeapTupleIsValid(tup))
@@ -106,7 +114,7 @@ GetSubscription(Oid subid, bool missing_ok, bool aclcheck)
 
        subform = (Form_pg_subscription) GETSTRUCT(tup);
 
-       sub = palloc_object(Subscription);
+       sub = palloc0_object(Subscription);
        sub->cxt = cxt;
        sub->oid = subid;
        sub->dbid = subform->subdbid;
@@ -125,38 +133,40 @@ GetSubscription(Oid subid, bool missing_ok, bool aclcheck)
        sub->maxretention = subform->submaxretention;
        sub->retentionactive = subform->subretentionactive;
 
-       /* Get conninfo */
-       if (OidIsValid(subform->subserver))
+       if (conninfo_needed)
        {
-               AclResult       aclresult;
-               ForeignServer *server;
-
-               server = GetForeignServer(subform->subserver);
-
-               /* recheck ACL if requested */
-               if (aclcheck)
+               if (OidIsValid(subform->subserver))
                {
-                       aclresult = object_aclcheck(ForeignServerRelationId,
-                                                                               subform->subserver,
-                                                                               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)));
+                       AclResult       aclresult;
+                       ForeignServer *server;
+
+                       server = GetForeignServer(subform->subserver);
+
+                       if (conninfo_aclcheck)
+                       {
+                               /* recheck ACL if requested */
+                               aclresult = object_aclcheck(ForeignServerRelationId,
+                                                                                       subform->subserver,
+                                                                                       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)));
+                       }
+
+                       sub->conninfo = ForeignServerConnectionString(subform->subowner,
+                                                                                                                 server);
+               }
+               else
+               {
+                       datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
+                                                                                  tup,
+                                                                                  Anum_pg_subscription_subconninfo);
+                       sub->conninfo = TextDatumGetCString(datum);
                }
-
-               sub->conninfo = ForeignServerConnectionString(subform->subowner,
-                                                                                                         server);
-       }
-       else
-       {
-               datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
-                                                                          tup,
-                                                                          Anum_pg_subscription_subconninfo);
-               sub->conninfo = TextDatumGetCString(datum);
        }
 
        /* Get slotname */

diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index 070141bce7583eeeb4847d99e221cd7e8dcdd31b..c9faf68cbc5b4ac39850fdb8a6dfc980f3d2923d 100644 (file)
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -1451,6 +1451,7 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
        Datum           values[Natts_pg_subscription];
        HeapTuple       tup;
        Oid                     subid;
+       bool            orig_conninfo_needed = true;
        bool            update_tuple = false;
        bool            update_failover = false;
        bool            update_two_phase = false;
@@ -1485,14 +1486,89 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
                aclcheck_error(ACLCHECK_NOT_OWNER, OBJECT_SUBSCRIPTION,
                                           stmt->subname);
 
+       /* parse and check options */
+       switch (stmt->kind)
+       {
+               case ALTER_SUBSCRIPTION_OPTIONS:
+                       supported_opts = (SUBOPT_SLOT_NAME |
+                                                         SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
+                                                         SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
+                                                         SUBOPT_DISABLE_ON_ERR |
+                                                         SUBOPT_PASSWORD_REQUIRED |
+                                                         SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
+                                                         SUBOPT_RETAIN_DEAD_TUPLES |
+                                                         SUBOPT_MAX_RETENTION_DURATION |
+                                                         SUBOPT_WAL_RECEIVER_TIMEOUT |
+                                                         SUBOPT_ORIGIN);
+                       break;
+
+               case ALTER_SUBSCRIPTION_ENABLED:
+                       supported_opts = SUBOPT_ENABLED;
+                       break;
+
+               case ALTER_SUBSCRIPTION_SET_PUBLICATION:
+                       supported_opts = SUBOPT_COPY_DATA | SUBOPT_REFRESH;
+                       break;
+
+               case ALTER_SUBSCRIPTION_ADD_PUBLICATION:
+               case ALTER_SUBSCRIPTION_DROP_PUBLICATION:
+                       supported_opts = SUBOPT_REFRESH | SUBOPT_COPY_DATA;
+                       break;
+
+               case ALTER_SUBSCRIPTION_REFRESH_PUBLICATION:
+                       supported_opts = SUBOPT_COPY_DATA;
+                       break;
+
+               case ALTER_SUBSCRIPTION_SKIP:
+                       supported_opts = SUBOPT_LSN;
+                       break;
+
+               default:
+                       supported_opts = 0;
+                       break;
+       }
+
+       if (supported_opts > 0)
+               parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
+
+       /*
+        * Ensure that ALTER SUBSCRIPTION commands that could be used to fix a
+        * broken connection or prepare to drop a broken subscription don't
+        * attempt to construct the conninfo. Otherwise, we might encounter the
+        * error the user is trying to fix.
+        *
+        * Specifically, ALTER SUBSCRIPTION DISABLE, ALTER SUBSCRIPTION SERVER,
+        * ALTER SUBSCRIPTION CONNECTION, or ALTER SUBSCRIPTION SET
+        * (slot_name=NONE).
+        *
+        * NB: if the user specifies multiple SET options, then we may still need
+        * to construct conninfo even if slot_name is set to NONE.
+        */
+       if (stmt->kind == ALTER_SUBSCRIPTION_ENABLED)
+       {
+               if (opts.specified_opts == SUBOPT_ENABLED && !opts.enabled)
+                       orig_conninfo_needed = false;
+       }
+       else if (stmt->kind == ALTER_SUBSCRIPTION_SERVER ||
+                        stmt->kind == ALTER_SUBSCRIPTION_CONNECTION)
+       {
+               orig_conninfo_needed = false;
+       }
+       else if (stmt->kind == ALTER_SUBSCRIPTION_OPTIONS)
+       {
+               /* ... SET (slot_name = NONE) with no other options */
+               if (opts.specified_opts == SUBOPT_SLOT_NAME && !opts.slot_name)
+                       orig_conninfo_needed = false;
+       }
+
        /*
         * Skip ACL checks on the subscription's foreign server, if any. If
         * changing the server (or replacing it with a raw connection), then the
         * old one will be removed anyway. If changing something unrelated,
         * there's no need to do an additional ACL check here; that will be done
-        * by the subscription worker anyway.
+        * by the subscription worker.
         */
-       sub = GetSubscription(subid, false, false);
+       sub = GetSubscription(subid, false, orig_conninfo_needed, false);
 
        retain_dead_tuples = sub->retaindeadtuples;
        origin = sub->origin;
@@ -1523,20 +1599,6 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
        {
                case ALTER_SUBSCRIPTION_OPTIONS:
                        {
-                               supported_opts = (SUBOPT_SLOT_NAME |
-                                                                 SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
-                                                                 SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
-                                                                 SUBOPT_DISABLE_ON_ERR |
-                                                                 SUBOPT_PASSWORD_REQUIRED |
-                                                                 SUBOPT_RUN_AS_OWNER | SUBOPT_FAILOVER |
-                                                                 SUBOPT_RETAIN_DEAD_TUPLES |
-                                                                 SUBOPT_MAX_RETENTION_DURATION |
-                                                                 SUBOPT_WAL_RECEIVER_TIMEOUT |
-                                                                 SUBOPT_ORIGIN);
-
-                               parse_subscription_options(pstate, stmt->options,
-                                                                                  supported_opts, &opts);
-
                                if (IsSet(opts.specified_opts, SUBOPT_SLOT_NAME))
                                {
                                        /*
@@ -1802,8 +1864,6 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 
                case ALTER_SUBSCRIPTION_ENABLED:
                        {
-                               parse_subscription_options(pstate, stmt->options,
-                                                                                  SUBOPT_ENABLED, &opts);
                                Assert(IsSet(opts.specified_opts, SUBOPT_ENABLED));
 
                                if (!sub->slotname && opts.enabled)
@@ -1940,10 +2000,6 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 
                case ALTER_SUBSCRIPTION_SET_PUBLICATION:
                        {
-                               supported_opts = SUBOPT_COPY_DATA | SUBOPT_REFRESH;
-                               parse_subscription_options(pstate, stmt->options,
-                                                                                  supported_opts, &opts);
-
                                values[Anum_pg_subscription_subpublications - 1] =
                                        publicationListToArray(stmt->publication);
                                replaces[Anum_pg_subscription_subpublications - 1] = true;
@@ -1987,10 +2043,6 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
                                List       *publist;
                                bool            isadd = stmt->kind == ALTER_SUBSCRIPTION_ADD_PUBLICATION;
 
-                               supported_opts = SUBOPT_REFRESH | SUBOPT_COPY_DATA;
-                               parse_subscription_options(pstate, stmt->options,
-                                                                                  supported_opts, &opts);
-
                                publist = merge_publications(sub->publications, stmt->publication, isadd, stmt->subname);
                                values[Anum_pg_subscription_subpublications - 1] =
                                        publicationListToArray(publist);
@@ -2048,9 +2100,6 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
                                                         errmsg("%s is not allowed for disabled subscriptions",
                                                                        "ALTER SUBSCRIPTION ... REFRESH PUBLICATION")));
 
-                               parse_subscription_options(pstate, stmt->options,
-                                                                                  SUBOPT_COPY_DATA, &opts);
-
                                /*
                                 * The subscription option "two_phase" requires that
                                 * replication has passed the initial table synchronization
@@ -2096,8 +2145,6 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
 
                case ALTER_SUBSCRIPTION_SKIP:
                        {
-                               parse_subscription_options(pstate, stmt->options, SUBOPT_LSN, &opts);
-
                                /* ALTER SUBSCRIPTION ... SKIP supports only LSN option */
                                Assert(IsSet(opts.specified_opts, SUBOPT_LSN));
 
@@ -2163,6 +2210,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
                char       *err;
                WalReceiverConn *wrconn;
 
+               Assert(new_conninfo || orig_conninfo_needed);
+
                /* Load the library providing us libpq calls. */
                load_file("libpqwalreceiver", false);
 

diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index a3f2406ed83fd583f8081cc2c83af3bd6289ef9f..7799266c61409c984780b9134100f823ae09415b 100644 (file)
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -5074,7 +5074,7 @@ maybe_reread_subscription(void)
                started_tx = true;
        }
 
-       newsub = GetSubscription(MyLogicalRepWorker->subid, true, true);
+       newsub = GetSubscription(MyLogicalRepWorker->subid, true, true, true);
 
        if (newsub)
        {
@@ -5823,7 +5823,7 @@ InitializeLogRepWorker(void)
        LockSharedObject(SubscriptionRelationId, MyLogicalRepWorker->subid, 0,
                                         AccessShareLock);
 
-       MySubscription = GetSubscription(MyLogicalRepWorker->subid, true, true);
+       MySubscription = GetSubscription(MyLogicalRepWorker->subid, true, true, true);
 
        if (MySubscription)
        {

diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index a6a2ad1e49c24ad6641f2ec05718d94d748ad7ee..489442018892f1d056372df9a48d1308e546420c 100644 (file)
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -213,7 +213,8 @@ typedef struct Subscription
 #endif                                                 /* EXPOSE_TO_CLIENT_CODE */
 
 extern Subscription *GetSubscription(Oid subid, bool missing_ok,
-                                                                        bool aclcheck);
+                                                                        bool conninfo_needed,
+                                                                        bool conninfo_aclcheck);
 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 a1b3cc96d83a69acfc59f070b3c622aa3e24702a..61236672ce454c4165a17d2608ad76e2849ab412 100644 (file)
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -184,19 +184,40 @@ DETAIL:  Foreign data wrapper must be defined with CONNECTION specified.
 RESET SESSION AUTHORIZATION;
 ALTER FOREIGN DATA WRAPPER test_fdw CONNECTION test_fdw_connection;
 SET SESSION AUTHORIZATION regress_subscription_user3;
-CREATE SUBSCRIPTION regress_testsub6 SERVER test_server PUBLICATION testpub WITH (slot_name = 'dummy', connect = false);
+CREATE SUBSCRIPTION regress_testsub6 SERVER test_server
+  PUBLICATION testpub WITH (slot_name = 'dummy', connect = false);
 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;
 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
+-- ok, lacks USAGE on test_server, but replacing connection anyway
+BEGIN;
+ALTER SUBSCRIPTION regress_testsub6 CONNECTION 'dbname=regress_doesnotexist password=secret';
+ABORT;
+-- fails, cannot drop slot
 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.
+RESET SESSION AUTHORIZATION;
+GRANT USAGE ON FOREIGN SERVER test_server TO regress_subscription_user3;
+SET SESSION AUTHORIZATION regress_subscription_user3;
 ALTER SUBSCRIPTION regress_testsub6 SET (slot_name = NONE);
-DROP SUBSCRIPTION regress_testsub6;
+DROP SUBSCRIPTION regress_testsub6; --ok
+CREATE SUBSCRIPTION regress_testsub6 SERVER test_server
+  PUBLICATION testpub WITH (slot_name = 'dummy', connect = false);
+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;
+-- ok, test_server lacks user mapping, but replacing connection anyway
+BEGIN;
+ALTER SUBSCRIPTION regress_testsub6 CONNECTION 'dbname=regress_doesnotexist password=secret';
+ABORT;
+CREATE USER MAPPING FOR regress_subscription_user3 SERVER test_server OPTIONS(user 'foo', password 'secret');
+ALTER SUBSCRIPTION regress_testsub6 DISABLE;
+ALTER SUBSCRIPTION regress_testsub6 SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub6; --ok
+DROP USER MAPPING FOR regress_subscription_user3 SERVER test_server;
 SET SESSION AUTHORIZATION regress_subscription_user;
 REVOKE CREATE ON DATABASE REGRESSION FROM regress_subscription_user3;
 DROP SERVER test_server;

diff --git a/src/test/regress/regress.c b/src/test/regress/regress.c
index 4927f1ddcbfe246f47a291dd1f9a1f20c5dac15b..d5aafdf370c51baa37a25bba55f8f71d26054441 100644 (file)
--- 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"
@@ -736,6 +737,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 528a10b5481c65da376a199c0a1e91e5984457f1..665b510f1805ffc0c919afe05e23c21d7c1792e7 100644 (file)
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -132,18 +132,45 @@ RESET SESSION AUTHORIZATION;
 ALTER FOREIGN DATA WRAPPER test_fdw CONNECTION test_fdw_connection;
 SET SESSION AUTHORIZATION regress_subscription_user3;
 
-CREATE SUBSCRIPTION regress_testsub6 SERVER test_server PUBLICATION testpub WITH (slot_name = 'dummy', connect = false);
+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;
 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
+-- ok, lacks USAGE on test_server, but replacing connection anyway
+BEGIN;
+ALTER SUBSCRIPTION regress_testsub6 CONNECTION 'dbname=regress_doesnotexist password=secret';
+ABORT;
+
+-- fails, cannot drop slot
 DROP SUBSCRIPTION regress_testsub6;
 
+RESET SESSION AUTHORIZATION;
+GRANT USAGE ON FOREIGN SERVER test_server TO regress_subscription_user3;
+SET SESSION AUTHORIZATION regress_subscription_user3;
+
 ALTER SUBSCRIPTION regress_testsub6 SET (slot_name = NONE);
-DROP SUBSCRIPTION regress_testsub6;
+DROP SUBSCRIPTION regress_testsub6; --ok
+
+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;
+
+-- ok, test_server lacks user mapping, but replacing connection anyway
+BEGIN;
+ALTER SUBSCRIPTION regress_testsub6 CONNECTION 'dbname=regress_doesnotexist password=secret';
+ABORT;
+
+CREATE USER MAPPING FOR regress_subscription_user3 SERVER test_server OPTIONS(user 'foo', password 'secret');
+
+ALTER SUBSCRIPTION regress_testsub6 DISABLE;
+ALTER SUBSCRIPTION regress_testsub6 SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub6; --ok
+
+DROP USER MAPPING FOR regress_subscription_user3 SERVER test_server;
 
 SET SESSION AUTHORIZATION regress_subscription_user;
 REVOKE CREATE ON DATABASE REGRESSION FROM regress_subscription_user3;