public inbox for [email protected]
help / color / mirror / Atom feedFrom: Zhijie Hou (Fujitsu) <[email protected]>
To: shveta malik <[email protected]>
To: Drouvot, Bertrand <[email protected]>
Cc: Peter Smith <[email protected]>
Cc: Amit Kapila <[email protected]>
Cc: Hayato Kuroda (Fujitsu) <[email protected]>
Cc: Bharath Rupireddy <[email protected]>
Cc: Peter Eisentraut <[email protected]>
Cc: Bruce Momjian <[email protected]>
Cc: Ashutosh Sharma <[email protected]>
Cc: Andres Freund <[email protected]>
Cc: pgsql-hackers <[email protected]>
Cc: Ajin Cherian <[email protected]>
Cc: Alvaro Herrera <[email protected]>
Subject: RE: Synchronizing slots from primary to standby
Date: Mon, 27 Nov 2023 08:57:33 +0000
Message-ID: <OS0PR01MB57161938A348977127D56A6694BDA@OS0PR01MB5716.jpnprd01.prod.outlook.com> (raw)
In-Reply-To: <CAJpy0uDKPK537ZW8CAjDVmSjHtBKCzTUzq=it84SrA0PfoWFCg@mail.gmail.com>
References: <CAJpy0uC+2agRtF3H=n-hW5JkoPfaZkjPXJr==y3_PRE04dQvhw@mail.gmail.com>
<[email protected]>
<CAA4eK1+8k_yTsHmnExMQ3TWrAJ_KmPQwoze6DZ5OuQQhqK0wLw@mail.gmail.com>
<[email protected]>
<CAA4eK1LbnX5Lu5-Zphv0=x8T+WnvSVoaDZHTY5RNk04dsymctA@mail.gmail.com>
<[email protected]>
<OS0PR01MB5716CE0729CEB3B5994A954194B3A@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<[email protected]>
<CAJpy0uAT-oGsgmb4dC=dYYRNY7WXEAAWGWU2kL5phLz3Wcdifg@mail.gmail.com>
<CAA4eK1J=-kPHS1eHNBtzOQHZ64j6WSgSYQZ3fH=2vfiwy_48AA@mail.gmail.com>
<CAJpy0uB_y_1akXUWM9WnSdQ-NVWTnHTpmNu6gJQ5tnyuAMTUOg@mail.gmail.com>
<CAHut+Pv-yu71ogj_hRi6cCtmD55bsyw7XTxj1Nq8yVFKpY3NDQ@mail.gmail.com>
<OS0PR01MB571646B8186F6A06404BD50194BDA@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<OS0PR01MB57161C27BCFAF04C2B48B47394BDA@OS0PR01MB5716.jpnprd01.prod.outlook.com>
<[email protected]>
<CAJpy0uDKPK537ZW8CAjDVmSjHtBKCzTUzq=it84SrA0PfoWFCg@mail.gmail.com>
On Monday, November 27, 2023 4:51 PM shveta malik <[email protected]> wrote:
>
> On Mon, Nov 27, 2023 at 2:15 PM Drouvot, Bertrand
> <[email protected]> wrote:
> >
> > Hi,
> >
> > On 11/27/23 7:02 AM, Zhijie Hou (Fujitsu) wrote:
> > > On Monday, November 27, 2023 12:03 PM Zhijie Hou (Fujitsu)
> <[email protected]> wrote:
> > >>
> > >> Attach the V38 patch set which addressed all comments in [1][2]
> > >> except for the ones that mentioned above.
> > >>
> > >> [1]
> > >>
> https://www.postgresql.org/message-id/CAHut%2BPv-yu71ogj_hRi6cCtmD5
> > >> 5bsyw7XTxj1Nq8yVFKpY3NDQ%40mail.gmail.com
> > >> [2]
> > >>
> https://www.postgresql.org/message-id/CAHut%2BPuEGX5kr0xh06yv8ndoA
> > >> QvDNedoec1OqOq3GMxDN6p%3D9A%40mail.gmail.com
> > >
> > > I didn't increment the patch version, sorry for that. Attach the
> > > same patch set but increment the patch version to V39.
> >
> > Thanks!
> >
> > It looks like v39 does not contain (some / all?) the changes that have
> > been done in v38 [1].
> >
> > For example, slot_exists_in_list() still exists in v39 while it was
> > renamed to
> > validate_sync_slot() in v38.
> >
>
> Yes, I noticed that and informed Hou-san about this. New patches will be
> posted soon with the correction. Meanwhile, please review v38 instead if you
> intend to review patch002 right now. v39 is supposed to have changes in
> patch001 alone.
Here is the updated version(v39_2) which include all the changes made in 0002.
Please use for review, and sorry for the confusion.
Best Regards,
Hou zj
Attachments:
[application/octet-stream] v39_2-0003-Allow-slot-sync-worker-to-wait-for-the-cascading.patch (8.0K, ../OS0PR01MB57161938A348977127D56A6694BDA@OS0PR01MB5716.jpnprd01.prod.outlook.com/2-v39_2-0003-Allow-slot-sync-worker-to-wait-for-the-cascading.patch)
download | inline diff:
From 9c7f74df663c1d95e877dfd3e143e9619a724a9f Mon Sep 17 00:00:00 2001
From: Shveta Malik <[email protected]>
Date: Thu, 23 Nov 2023 12:49:05 +0530
Subject: [PATCH v39 3/3] Allow slot-sync worker to wait for the cascading
standbys.
The GUC standby_slot_names is needed to be set on first standby
in order to allow it to wait for confirmation for cascading
standbys before updating logical 'synced' slots in slot-sync worker.
The intent is that the logical slots (synced ones) should not go
ahead of cascading standbys.
For the user created slots on first standby, we already have this wait
logic in place in logical walsender and in pg_logical_slot_get_changes_guts(),
but for synced slots (which can not be consumed yet), we need to make
sure that they are not going ahead of cascading standbys and that is
acheived by introducing the wait in slot-sync worker before we actually
update the slots.
---
src/backend/replication/logical/slotsync.c | 86 ++++++++++++++++++++--
src/backend/replication/walsender.c | 11 +--
src/include/replication/walsender.h | 5 ++
3 files changed, 89 insertions(+), 13 deletions(-)
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
index 9aa3562500..1aac8ce648 100644
--- a/src/backend/replication/logical/slotsync.c
+++ b/src/backend/replication/logical/slotsync.c
@@ -83,6 +83,9 @@ typedef struct RemoteSlot
*/
#define WORKER_PRIMARY_CATCHUP_WAIT_ATTEMPTS 5
+static void ProcessSlotSyncInterrupts(WalReceiverConn **wrconn,
+ List **standby_slots);
+
/*
* Wait for remote slot to pass locally reserved position.
*
@@ -452,6 +455,52 @@ construct_slot_query(StringInfo s)
" WHERE failover and sync_state != 'i'");
}
+/*
+ * Wait for cascading physical standbys corresponding to physical slots
+ * specified in standby_slot_names GUC to confirm receiving given lsn.
+ */
+static void
+wait_for_standby_confirmation(XLogRecPtr wait_for_lsn,
+ WalReceiverConn *wrconn)
+{
+ List *standby_slots;
+
+ /* Nothing to be done */
+ if (strcmp(standby_slot_names, "") == 0)
+ return;
+
+ standby_slots = GetStandbySlotList(true);
+
+ for (;;)
+ {
+ int rc;
+
+ WalSndFilterStandbySlots(wait_for_lsn, &standby_slots);
+
+ /* Exit if done waiting for every slot. */
+ if (standby_slots == NIL)
+ break;
+
+ /*
+ * This will reload configuration and will refresh the standby_slots
+ * as well provided standby_slot_names GUC is changed by the user.
+ */
+ ProcessSlotSyncInterrupts(&wrconn, &standby_slots);
+
+ /*
+ * XXX: Is waiting for 5 second before retrying enough or more or
+ * less?
+ */
+ rc = WaitLatch(MyLatch,
+ WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ 5000L,
+ WAIT_EVENT_WAL_SENDER_WAIT_FOR_STANDBY_CONFIRMATION);
+
+ if (rc & WL_LATCH_SET)
+ ResetLatch(MyLatch);
+ }
+}
+
/*
* Synchronize single slot to given position.
*
@@ -723,6 +772,7 @@ synchronize_slots(WalReceiverConn *wrconn)
List *remote_slot_list = NIL;
MemoryContext oldctx = CurrentMemoryContext;
long naptime = WORKER_DEFAULT_NAPTIME_MS;
+ XLogRecPtr max_confirmed_lsn = 0;
ListCell *cell;
bool slot_updated = false;
TimestampTz now;
@@ -775,6 +825,9 @@ synchronize_slots(WalReceiverConn *wrconn)
remote_slot->plugin = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
Assert(!isnull);
+ if (remote_slot->confirmed_lsn > max_confirmed_lsn)
+ max_confirmed_lsn = remote_slot->confirmed_lsn;
+
/*
* It is possible to get null values for LSN and Xmin if slot is
* invalidated on the primary server, so handle accordingly.
@@ -818,6 +871,17 @@ synchronize_slots(WalReceiverConn *wrconn)
*/
drop_obsolete_slots(remote_slot_list);
+ /*
+ * If there are cascading standbys, wait for their confirmation before we
+ * update synced logical slots locally.
+ *
+ * Instead of waiting on confirmation for lsn of each slot, let us wait
+ * once for confirmation on max_confirmed_lsn. If that is confirmed by
+ * each cascading standby, we are good to update all the slots.
+ */
+ if (remote_slot_list)
+ wait_for_standby_confirmation(max_confirmed_lsn, wrconn);
+
/* Now sync the slots locally */
foreach(cell, remote_slot_list)
{
@@ -877,12 +941,18 @@ remote_connect(void)
* If primary_conninfo has changed, reconnect to primary.
*/
static void
-slotsync_reread_config(WalReceiverConn **wrconn)
+slotsync_reread_config(WalReceiverConn **wrconn, List **standby_slots)
{
char *conninfo = pstrdup(PrimaryConnInfo);
- ConfigReloadPending = false;
- ProcessConfigFile(PGC_SIGHUP);
+ /*
+ * Reload configs and recreate the standby_slot_names_list if GUC
+ * standby_slot_names changed.
+ */
+ if (standby_slots)
+ WalSndRereadConfigAndReInitSlotList(standby_slots);
+ else
+ ProcessConfigFile(PGC_SIGHUP);
/* Exit if GUC primary_conninfo got changed, let the launcher relaunch it */
if (strcmp(conninfo, PrimaryConnInfo) != 0)
@@ -905,7 +975,8 @@ slotsync_reread_config(WalReceiverConn **wrconn)
* Interrupt handler for main loop of slot sync worker.
*/
static void
-ProcessSlotSyncInterrupts(WalReceiverConn **wrconn)
+ProcessSlotSyncInterrupts(WalReceiverConn **wrconn,
+ List **standby_slots)
{
CHECK_FOR_INTERRUPTS();
@@ -921,7 +992,10 @@ ProcessSlotSyncInterrupts(WalReceiverConn **wrconn)
if (ConfigReloadPending)
- slotsync_reread_config(wrconn);
+ {
+ ConfigReloadPending = false;
+ slotsync_reread_config(wrconn, standby_slots);
+ }
}
/*
@@ -982,7 +1056,7 @@ ReplSlotSyncWorkerMain(Datum main_arg)
int rc;
long naptime;
- ProcessSlotSyncInterrupts(&wrconn);
+ ProcessSlotSyncInterrupts(&wrconn, NULL /* standby_slots */ );
/* Check if got promoted */
if (!RecoveryInProgress())
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index edb3656cd2..db5af0b598 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1608,7 +1608,7 @@ PhysicalWakeupLogicalWalSnd(void)
* Reload the config file and reinitialize the standby slot list if the GUC
* standby_slot_names has changed.
*/
-static void
+void
WalSndRereadConfigAndReInitSlotList(List **standby_slots)
{
char *pre_standby_slot_names = pstrdup(standby_slot_names);
@@ -1631,13 +1631,12 @@ WalSndRereadConfigAndReInitSlotList(List **standby_slots)
* This function updates the passed standby_slots list, removing any slots that
* have already caught up to or surpassed the given wait_for_lsn.
*/
-static void
+void
WalSndFilterStandbySlots(XLogRecPtr wait_for_lsn, List **standby_slots)
{
ListCell *lc;
- List *standby_slots_cpy = *standby_slots;
- foreach(lc, standby_slots_cpy)
+ foreach(lc, *standby_slots)
{
char *name = lfirst(lc);
XLogRecPtr restart_lsn = InvalidXLogRecPtr;
@@ -1704,10 +1703,8 @@ WalSndFilterStandbySlots(XLogRecPtr wait_for_lsn, List **standby_slots)
if (warningfmt)
ereport(WARNING, errmsg(warningfmt, name, "standby_slot_names"));
- standby_slots_cpy = foreach_delete_current(standby_slots_cpy, lc);
+ *standby_slots = foreach_delete_current(*standby_slots, lc);
}
-
- *standby_slots = standby_slots_cpy;
}
/*
diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h
index 1fcc22a127..c5c4714788 100644
--- a/src/include/replication/walsender.h
+++ b/src/include/replication/walsender.h
@@ -15,6 +15,7 @@
#include <signal.h>
#include "access/xlogdefs.h"
+#include "nodes/pg_list.h"
/*
* What to do with a snapshot in create replication slot command.
@@ -50,7 +51,11 @@ extern void WalSndWaitStopping(void);
extern void HandleWalSndInitStopping(void);
extern void WalSndRqstFileReload(void);
extern void PhysicalWakeupLogicalWalSnd(void);
+extern void WalSndFilterStandbySlots(XLogRecPtr wait_for_lsn,
+ List **standby_slots);
extern void WalSndWaitForStandbyConfirmation(XLogRecPtr wait_for_lsn);
+extern List *WalSndGetStandbySlots(void);
+extern void WalSndRereadConfigAndReInitSlotList(List **standby_slots);
/*
* Remember that we want to wakeup walsenders later
--
2.30.0.windows.2
[application/octet-stream] v39_2-0002-Add-logical-slot-sync-capability-to-the-physical.patch (103.1K, ../OS0PR01MB57161938A348977127D56A6694BDA@OS0PR01MB5716.jpnprd01.prod.outlook.com/3-v39_2-0002-Add-logical-slot-sync-capability-to-the-physical.patch)
download | inline diff:
From 8004acbaaceb64d4fcfdc2d429c232609b9a5efe Mon Sep 17 00:00:00 2001
From: Hou Zhijie <[email protected]>
Date: Mon, 27 Nov 2023 16:47:02 +0800
Subject: [PATCH v39 2/3] Add logical slot sync capability to the physical
standby
This patch implements synchronization of logical replication slots
from the primary server to the physical standby so that logical
replication can be resumed after failover. All the failover logical
replication slots on the primary (assuming configurations are
appropriate) are automatically created on the physical standbys and
are synced periodically. Slot-sync worker on the standby server
ping the primary server at regular intervals to get the necessary
failover logical slots information and create/update the slots locally.
GUC 'enable_syncslot' enables a physical standby to synchronize failover
logical replication slots from the primary server.
The replication launcher on the physical standby starts slot-sync worker
which is then responsible to keep on syncing the logical failover slots
from the primary server.
The nap time of worker is tuned according to the activity on the primary.
The worker starts with nap time of 10ms and if no activity is observed on
the primary for some time, then nap time is increased to 10sec. And if
activity is observed again, nap time is reduced back to 10ms.
The logical slots created by slot-sync worker on physical standbys are not
allowed to be dropped or consumed. Any attempt to perform logical decoding on
such slots will result in an error.
If a logical slot is invalidated on the primary, slot on the standby is also
invalidated. If a logical slot on the primary is valid but is invalidated
on the standby due to conflict (say required rows removed on the primary),
then that slot is dropped and recreated on the standby in next sync-cycle.
It is okay to recreate such slots as long as these are not consumable on the
standby (which is the case currently).
Slots synced on the standby can be identified using 'sync_state' column of
pg_replication_slots view. The values are:
'n': none for user slots,
'i': sync initiated for the slot but waiting for the remote slot on the
primary server to catch up.
'r': ready for periodic syncs.
---
doc/src/sgml/config.sgml | 35 +-
doc/src/sgml/logicaldecoding.sgml | 13 +
doc/src/sgml/system-views.sgml | 23 +
src/backend/access/transam/xlogrecovery.c | 11 +
src/backend/catalog/system_views.sql | 3 +-
src/backend/postmaster/bgworker.c | 5 +-
.../libpqwalreceiver/libpqwalreceiver.c | 44 +-
src/backend/replication/logical/Makefile | 1 +
.../replication/logical/applyparallelworker.c | 3 +-
src/backend/replication/logical/launcher.c | 659 +++++++++--
src/backend/replication/logical/logical.c | 22 +
src/backend/replication/logical/meson.build | 1 +
src/backend/replication/logical/slotsync.c | 1018 +++++++++++++++++
src/backend/replication/logical/tablesync.c | 5 +-
src/backend/replication/slot.c | 18 +-
src/backend/replication/slotfuncs.c | 33 +-
src/backend/replication/walsender.c | 2 +-
src/backend/storage/lmgr/lwlocknames.txt | 1 +
.../utils/activity/wait_event_names.txt | 2 +
src/backend/utils/misc/guc_tables.c | 12 +
src/backend/utils/misc/postgresql.conf.sample | 1 +
src/include/catalog/pg_proc.dat | 10 +-
src/include/commands/subscriptioncmds.h | 4 +
src/include/nodes/replnodes.h | 1 -
src/include/replication/logicallauncher.h | 9 +-
src/include/replication/logicalworker.h | 1 +
src/include/replication/slot.h | 19 +-
src/include/replication/walreceiver.h | 19 +
src/include/replication/worker_internal.h | 63 +-
src/test/recovery/t/050_verify_slot_order.pl | 127 ++
src/test/regress/expected/rules.out | 5 +-
src/test/regress/expected/sysviews.out | 3 +-
src/tools/pgindent/typedefs.list | 5 +
33 files changed, 2026 insertions(+), 152 deletions(-)
create mode 100644 src/backend/replication/logical/slotsync.c
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 6440378d5c..b2da901ad6 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -4357,6 +4357,12 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"' # Windows
meant to switch to a physical standby after the standby is promoted,
the physical replication slot for the standby should be listed here.
</para>
+ <para>
+ The standbys corresponding to the physical replication slots in
+ <varname>standby_slot_names</varname> must enable
+ <varname>enable_syncslot</varname> for the standbys to receive
+ failover logical slots changes from the primary.
+ </para>
</listitem>
</varlistentry>
@@ -4550,10 +4556,15 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
A password needs to be provided too, if the sender demands password
authentication. It can be provided in the
<varname>primary_conninfo</varname> string, or in a separate
- <filename>~/.pgpass</filename> file on the standby server (use
+ <filename>~/.pgpass</filename> file on the standby server. (use
<literal>replication</literal> as the database name).
- Do not specify a database name in the
- <varname>primary_conninfo</varname> string.
+ </para>
+ <para>
+ Specify <literal>dbname</literal> in
+ <varname>primary_conninfo</varname> string to allow synchronization
+ of slots from the primary server to the standby server.
+ This will only be used for slot synchronization. It is ignored
+ for streaming.
</para>
<para>
This parameter can only be set in the <filename>postgresql.conf</filename>
@@ -4878,6 +4889,24 @@ ANY <replaceable class="parameter">num_sync</replaceable> ( <replaceable class="
</listitem>
</varlistentry>
+ <varlistentry id="guc-enable-syncslot" xreflabel="enable_syncslot">
+ <term><varname>enable_syncslot</varname> (<type>boolean</type>)
+ <indexterm>
+ <primary><varname>enable_syncslot</varname> configuration parameter</primary>
+ </indexterm>
+ </term>
+ <listitem>
+ <para>
+ It enables a physical standby to synchronize logical failover slots
+ from the primary server so that logical subscribers are not blocked
+ after failover.
+ </para>
+ <para>
+ It is disabled by default. This parameter can only be set in the
+ <filename>postgresql.conf</filename> file or on the server command line.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist>
</sect2>
diff --git a/doc/src/sgml/logicaldecoding.sgml b/doc/src/sgml/logicaldecoding.sgml
index cd152d4ced..90bec06f36 100644
--- a/doc/src/sgml/logicaldecoding.sgml
+++ b/doc/src/sgml/logicaldecoding.sgml
@@ -346,6 +346,19 @@ postgres=# select * from pg_logical_slot_get_changes('regression_slot', NULL, NU
<function>pg_log_standby_snapshot</function> function on the primary.
</para>
+ <para>
+ A logical replication slot on the primary can be synchronized to the hot
+ standby by enabling the failover option during slot creation and set
+ <varname>enable_syncslot</varname> on the standby. The physical replication
+ slot for the standby should be listed in
+ <varname>standby_slot_names</varname> on the primary to prevent the
+ subscriber from consuming changes faster than the hot standby.
+ Additionally, similar to creating a logical replication slot on the hot
+ standby, <varname>hot_standby_feedback</varname> should be set on the
+ standby and a physical slot between the primary and the standby should be
+ used.
+ </para>
+
<caution>
<para>
Replication slots persist across crashes and know nothing about the state
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index 1dc695fd3a..f08beb83e1 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2543,6 +2543,29 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
after failover. Always false for physical slots.
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>sync_state</structfield> <type>char</type>
+ </para>
+ <para>
+ Defines slot synchronization state. This is meaningful on the physical
+ standby which has enabled slots synchronization.
+ </para>
+ <para>
+ State code:
+ <literal>n</literal> = none for user created slots,
+ <literal>i</literal> = sync initiated for the slot but slot is not ready
+ yet for periodic syncs,
+ <literal>r</literal> = ready for periodic syncs
+ </para>
+ <para>
+ The primary server will have sync_state as 'n' for all the slots.
+ But if the standby is promoted to become the new primary server,
+ sync_state can be seen 'r' as well. On this new primary server, slots
+ with sync_state as 'r' and 'n' behaves the same.
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/src/backend/access/transam/xlogrecovery.c b/src/backend/access/transam/xlogrecovery.c
index c61566666a..8ea6dc799a 100644
--- a/src/backend/access/transam/xlogrecovery.c
+++ b/src/backend/access/transam/xlogrecovery.c
@@ -49,7 +49,9 @@
#include "postmaster/bgwriter.h"
#include "postmaster/startup.h"
#include "replication/slot.h"
+#include "replication/logicallauncher.h"
#include "replication/walreceiver.h"
+#include "replication/worker_internal.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/latch.h"
@@ -1435,6 +1437,15 @@ FinishWalRecovery(void)
*/
XLogShutdownWalRcv();
+ /*
+ * Shutdown the slot sync workers to prevent potential conflicts between
+ * user processes and slotsync workers after a promotion. Additionally,
+ * drop any slots that have initiated but not yet completed the sync
+ * process.
+ */
+ ShutDownSlotSync();
+ slotsync_drop_initiated_slots();
+
/*
* We are now done reading the xlog from stream. Turn off streaming
* recovery to force fetching the files (which would be required at end of
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 63038f87f7..c4b3e8a807 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1024,7 +1024,8 @@ CREATE VIEW pg_replication_slots AS
L.safe_wal_size,
L.two_phase,
L.conflicting,
- L.failover
+ L.failover,
+ L.sync_state
FROM pg_get_replication_slots() AS L
LEFT JOIN pg_database D ON (L.datoid = D.oid);
diff --git a/src/backend/postmaster/bgworker.c b/src/backend/postmaster/bgworker.c
index 48a9924527..0e039c786f 100644
--- a/src/backend/postmaster/bgworker.c
+++ b/src/backend/postmaster/bgworker.c
@@ -125,11 +125,14 @@ static const struct
"ParallelWorkerMain", ParallelWorkerMain
},
{
- "ApplyLauncherMain", ApplyLauncherMain
+ "LauncherMain", LauncherMain
},
{
"ApplyWorkerMain", ApplyWorkerMain
},
+ {
+ "ReplSlotSyncWorkerMain", ReplSlotSyncWorkerMain
+ },
{
"ParallelApplyWorkerMain", ParallelApplyWorkerMain
},
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 336c2bec99..5f82d01e6d 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -34,6 +34,7 @@
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/tuplestore.h"
+#include "utils/varlena.h"
PG_MODULE_MAGIC;
@@ -58,6 +59,7 @@ static void libpqrcv_get_senderinfo(WalReceiverConn *conn,
char **sender_host, int *sender_port);
static char *libpqrcv_identify_system(WalReceiverConn *conn,
TimeLineID *primary_tli);
+static char *libpqrcv_get_dbname_from_conninfo(const char *conninfo);
static int libpqrcv_server_version(WalReceiverConn *conn);
static void libpqrcv_readtimelinehistoryfile(WalReceiverConn *conn,
TimeLineID tli, char **filename,
@@ -100,6 +102,7 @@ static WalReceiverFunctionsType PQWalReceiverFunctions = {
.walrcv_send = libpqrcv_send,
.walrcv_create_slot = libpqrcv_create_slot,
.walrcv_alter_slot = libpqrcv_alter_slot,
+ .walrcv_get_dbname_from_conninfo = libpqrcv_get_dbname_from_conninfo,
.walrcv_get_backend_pid = libpqrcv_get_backend_pid,
.walrcv_exec = libpqrcv_exec,
.walrcv_disconnect = libpqrcv_disconnect
@@ -413,6 +416,45 @@ libpqrcv_server_version(WalReceiverConn *conn)
return PQserverVersion(conn->streamConn);
}
+/*
+ * Get database name from the primary server's conninfo.
+ *
+ * If dbname is not found in connInfo, return NULL value.
+ */
+static char *
+libpqrcv_get_dbname_from_conninfo(const char *connInfo)
+{
+ PQconninfoOption *opts;
+ PQconninfoOption *opt;
+ char *dbname = NULL;
+ char *err = NULL;
+
+ opts = PQconninfoParse(connInfo, &err);
+ if (opts == NULL)
+ {
+ /* The error string is malloc'd, so we must free it explicitly */
+ char *errcopy = err ? pstrdup(err) : "out of memory";
+
+ PQfreemem(err);
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("invalid connection string syntax: %s", errcopy)));
+ }
+
+ for (opt = opts; opt->keyword != NULL; ++opt)
+ {
+ /*
+ * If multiple dbnames are specified, then the last one will be
+ * returned
+ */
+ if (strcmp(opt->keyword, "dbname") == 0 && opt->val &&
+ opt->val[0] != '\0')
+ dbname = pstrdup(opt->val);
+ }
+
+ return dbname;
+}
+
/*
* Start streaming WAL data from given streaming options.
*
@@ -998,7 +1040,7 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
*/
static void
libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
- bool failover)
+ bool failover)
{
StringInfoData cmd;
PGresult *res;
diff --git a/src/backend/replication/logical/Makefile b/src/backend/replication/logical/Makefile
index 2dc25e37bb..ba03eeff1c 100644
--- a/src/backend/replication/logical/Makefile
+++ b/src/backend/replication/logical/Makefile
@@ -25,6 +25,7 @@ OBJS = \
proto.o \
relation.o \
reorderbuffer.o \
+ slotsync.o \
snapbuild.o \
tablesync.o \
worker.o
diff --git a/src/backend/replication/logical/applyparallelworker.c b/src/backend/replication/logical/applyparallelworker.c
index 9b37736f8e..192c9e1860 100644
--- a/src/backend/replication/logical/applyparallelworker.c
+++ b/src/backend/replication/logical/applyparallelworker.c
@@ -922,7 +922,8 @@ ParallelApplyWorkerMain(Datum main_arg)
before_shmem_exit(pa_shutdown, PointerGetDatum(seg));
SpinLockAcquire(&MyParallelShared->mutex);
- MyParallelShared->logicalrep_worker_generation = MyLogicalRepWorker->generation;
+ MyParallelShared->logicalrep_worker_generation =
+ MyLogicalRepWorker->hdr.generation;
MyParallelShared->logicalrep_worker_slot_no = worker_slot;
SpinLockRelease(&MyParallelShared->mutex);
diff --git a/src/backend/replication/logical/launcher.c b/src/backend/replication/logical/launcher.c
index 501910b445..acafb54276 100644
--- a/src/backend/replication/logical/launcher.c
+++ b/src/backend/replication/logical/launcher.c
@@ -8,20 +8,27 @@
* src/backend/replication/logical/launcher.c
*
* NOTES
- * This module contains the logical replication worker launcher which
- * uses the background worker infrastructure to start the logical
- * replication workers for every enabled subscription.
+ * This module contains the replication worker launcher which
+ * uses the background worker infrastructure to:
+ * a) start the logical replication workers for every enabled subscription
+ * when not in standby_mode.
+ * b) start the slot sync worker for logical failover slots synchronization
+ * from the primary server when in standby_mode.
*
*-------------------------------------------------------------------------
*/
#include "postgres.h"
+#include "access/genam.h"
#include "access/heapam.h"
#include "access/htup.h"
#include "access/htup_details.h"
#include "access/tableam.h"
#include "access/xact.h"
+#include "access/xlogrecovery.h"
+#include "catalog/pg_authid.h"
+#include "catalog/pg_database.h"
#include "catalog/pg_subscription.h"
#include "catalog/pg_subscription_rel.h"
#include "funcapi.h"
@@ -44,6 +51,7 @@
#include "storage/procsignal.h"
#include "tcop/tcopprot.h"
#include "utils/builtins.h"
+#include "utils/fmgroids.h"
#include "utils/memutils.h"
#include "utils/pg_lsn.h"
#include "utils/ps_status.h"
@@ -57,6 +65,9 @@
int max_logical_replication_workers = 4;
int max_sync_workers_per_subscription = 2;
int max_parallel_apply_workers_per_subscription = 2;
+bool enable_syncslot = false;
+
+SlotSyncWorkerInfo *SlotSyncWorker = NULL;
LogicalRepWorker *MyLogicalRepWorker = NULL;
@@ -70,6 +81,7 @@ typedef struct LogicalRepCtxStruct
dshash_table_handle last_start_dsh;
/* Background workers. */
+ SlotSyncWorkerInfo ss_worker; /* slot sync worker */
LogicalRepWorker workers[FLEXIBLE_ARRAY_MEMBER];
} LogicalRepCtxStruct;
@@ -102,6 +114,7 @@ static void logicalrep_launcher_onexit(int code, Datum arg);
static void logicalrep_worker_onexit(int code, Datum arg);
static void logicalrep_worker_detach(void);
static void logicalrep_worker_cleanup(LogicalRepWorker *worker);
+static void slotsync_worker_cleanup(SlotSyncWorkerInfo *worker);
static int logicalrep_pa_worker_count(Oid subid);
static void logicalrep_launcher_attach_dshmem(void);
static void ApplyLauncherSetWorkerStartTime(Oid subid, TimestampTz start_time);
@@ -178,6 +191,8 @@ get_subscription_list(void)
}
/*
+ * This is common code for logical workers and slot sync worker.
+ *
* Wait for a background worker to start up and attach to the shmem context.
*
* This is only needed for cleaning up the shared memory in case the worker
@@ -186,12 +201,14 @@ get_subscription_list(void)
* Returns whether the attach was successful.
*/
static bool
-WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
+WaitForReplicationWorkerAttach(LogicalWorkerHeader *worker,
uint16 generation,
- BackgroundWorkerHandle *handle)
+ BackgroundWorkerHandle *handle,
+ LWLock *lock)
{
BgwHandleStatus status;
int rc;
+ bool is_slotsync_worker = (lock == SlotSyncWorkerLock) ? true : false;
for (;;)
{
@@ -199,27 +216,32 @@ WaitForReplicationWorkerAttach(LogicalRepWorker *worker,
CHECK_FOR_INTERRUPTS();
- LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+ LWLockAcquire(lock, LW_SHARED);
/* Worker either died or has started. Return false if died. */
if (!worker->in_use || worker->proc)
{
- LWLockRelease(LogicalRepWorkerLock);
+ LWLockRelease(lock);
return worker->in_use;
}
- LWLockRelease(LogicalRepWorkerLock);
+ LWLockRelease(lock);
/* Check if worker has died before attaching, and clean up after it. */
status = GetBackgroundWorkerPid(handle, &pid);
if (status == BGWH_STOPPED)
{
- LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
+ LWLockAcquire(lock, LW_EXCLUSIVE);
/* Ensure that this was indeed the worker we waited for. */
if (generation == worker->generation)
- logicalrep_worker_cleanup(worker);
- LWLockRelease(LogicalRepWorkerLock);
+ {
+ if (is_slotsync_worker)
+ slotsync_worker_cleanup((SlotSyncWorkerInfo *) worker);
+ else
+ logicalrep_worker_cleanup((LogicalRepWorker *) worker);
+ }
+ LWLockRelease(lock);
return false;
}
@@ -262,8 +284,8 @@ logicalrep_worker_find(Oid subid, Oid relid, bool only_running)
if (isParallelApplyWorker(w))
continue;
- if (w->in_use && w->subid == subid && w->relid == relid &&
- (!only_running || w->proc))
+ if (w->hdr.in_use && w->subid == subid && w->relid == relid &&
+ (!only_running || w->hdr.proc))
{
res = w;
break;
@@ -290,7 +312,8 @@ logicalrep_workers_find(Oid subid, bool only_running)
{
LogicalRepWorker *w = &LogicalRepCtx->workers[i];
- if (w->in_use && w->subid == subid && (!only_running || w->proc))
+ if (w->hdr.in_use && w->subid == subid &&
+ (!only_running || w->hdr.proc))
res = lappend(res, w);
}
@@ -351,7 +374,7 @@ retry:
{
LogicalRepWorker *w = &LogicalRepCtx->workers[i];
- if (!w->in_use)
+ if (!w->hdr.in_use)
{
worker = w;
slot = i;
@@ -380,7 +403,7 @@ retry:
* If the worker was marked in use but didn't manage to attach in
* time, clean it up.
*/
- if (w->in_use && !w->proc &&
+ if (w->hdr.in_use && !w->hdr.proc &&
TimestampDifferenceExceeds(w->launch_time, now,
wal_receiver_timeout))
{
@@ -438,9 +461,9 @@ retry:
/* Prepare the worker slot. */
worker->type = wtype;
worker->launch_time = now;
- worker->in_use = true;
- worker->generation++;
- worker->proc = NULL;
+ worker->hdr.in_use = true;
+ worker->hdr.generation++;
+ worker->hdr.proc = NULL;
worker->dbid = dbid;
worker->userid = userid;
worker->subid = subid;
@@ -457,7 +480,7 @@ retry:
TIMESTAMP_NOBEGIN(worker->reply_time);
/* Before releasing lock, remember generation for future identification. */
- generation = worker->generation;
+ generation = worker->hdr.generation;
LWLockRelease(LogicalRepWorkerLock);
@@ -510,7 +533,7 @@ retry:
{
/* Failed to start worker, so clean up the worker slot. */
LWLockAcquire(LogicalRepWorkerLock, LW_EXCLUSIVE);
- Assert(generation == worker->generation);
+ Assert(generation == worker->hdr.generation);
logicalrep_worker_cleanup(worker);
LWLockRelease(LogicalRepWorkerLock);
@@ -522,19 +545,23 @@ retry:
}
/* Now wait until it attaches. */
- return WaitForReplicationWorkerAttach(worker, generation, bgw_handle);
+ return WaitForReplicationWorkerAttach((LogicalWorkerHeader *) worker,
+ generation,
+ bgw_handle,
+ LogicalRepWorkerLock);
}
/*
* Internal function to stop the worker and wait until it detaches from the
- * slot.
+ * slot. It is used for both logical workers and slot sync worker.
*/
static void
-logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo)
+logicalrep_worker_stop_internal(LogicalWorkerHeader *worker, int signo,
+ LWLock *lock)
{
uint16 generation;
- Assert(LWLockHeldByMeInMode(LogicalRepWorkerLock, LW_SHARED));
+ Assert(LWLockHeldByMeInMode(lock, LW_SHARED));
/*
* Remember which generation was our worker so we can check if what we see
@@ -550,7 +577,7 @@ logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo)
{
int rc;
- LWLockRelease(LogicalRepWorkerLock);
+ LWLockRelease(lock);
/* Wait a bit --- we don't expect to have to wait long. */
rc = WaitLatch(MyLatch,
@@ -564,7 +591,7 @@ logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo)
}
/* Recheck worker status. */
- LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+ LWLockAcquire(lock, LW_SHARED);
/*
* Check whether the worker slot is no longer used, which would mean
@@ -591,7 +618,7 @@ logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo)
if (!worker->proc || worker->generation != generation)
break;
- LWLockRelease(LogicalRepWorkerLock);
+ LWLockRelease(lock);
/* Wait a bit --- we don't expect to have to wait long. */
rc = WaitLatch(MyLatch,
@@ -604,7 +631,7 @@ logicalrep_worker_stop_internal(LogicalRepWorker *worker, int signo)
CHECK_FOR_INTERRUPTS();
}
- LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+ LWLockAcquire(lock, LW_SHARED);
}
}
@@ -623,7 +650,9 @@ logicalrep_worker_stop(Oid subid, Oid relid)
if (worker)
{
Assert(!isParallelApplyWorker(worker));
- logicalrep_worker_stop_internal(worker, SIGTERM);
+ logicalrep_worker_stop_internal((LogicalWorkerHeader *) worker,
+ SIGTERM,
+ LogicalRepWorkerLock);
}
LWLockRelease(LogicalRepWorkerLock);
@@ -669,8 +698,10 @@ logicalrep_pa_worker_stop(ParallelApplyWorkerInfo *winfo)
/*
* Only stop the worker if the generation matches and the worker is alive.
*/
- if (worker->generation == generation && worker->proc)
- logicalrep_worker_stop_internal(worker, SIGINT);
+ if (worker->hdr.generation == generation && worker->hdr.proc)
+ logicalrep_worker_stop_internal((LogicalWorkerHeader *) worker,
+ SIGINT,
+ LogicalRepWorkerLock);
LWLockRelease(LogicalRepWorkerLock);
}
@@ -696,14 +727,14 @@ logicalrep_worker_wakeup(Oid subid, Oid relid)
/*
* Wake up (using latch) the specified logical replication worker.
*
- * Caller must hold lock, else worker->proc could change under us.
+ * Caller must hold lock, else worker->hdr.proc could change under us.
*/
void
logicalrep_worker_wakeup_ptr(LogicalRepWorker *worker)
{
Assert(LWLockHeldByMe(LogicalRepWorkerLock));
- SetLatch(&worker->proc->procLatch);
+ SetLatch(&worker->hdr.proc->procLatch);
}
/*
@@ -718,7 +749,7 @@ logicalrep_worker_attach(int slot)
Assert(slot >= 0 && slot < max_logical_replication_workers);
MyLogicalRepWorker = &LogicalRepCtx->workers[slot];
- if (!MyLogicalRepWorker->in_use)
+ if (!MyLogicalRepWorker->hdr.in_use)
{
LWLockRelease(LogicalRepWorkerLock);
ereport(ERROR,
@@ -727,7 +758,7 @@ logicalrep_worker_attach(int slot)
slot)));
}
- if (MyLogicalRepWorker->proc)
+ if (MyLogicalRepWorker->hdr.proc)
{
LWLockRelease(LogicalRepWorkerLock);
ereport(ERROR,
@@ -736,7 +767,7 @@ logicalrep_worker_attach(int slot)
"another worker, cannot attach", slot)));
}
- MyLogicalRepWorker->proc = MyProc;
+ MyLogicalRepWorker->hdr.proc = MyProc;
before_shmem_exit(logicalrep_worker_onexit, (Datum) 0);
LWLockRelease(LogicalRepWorkerLock);
@@ -771,7 +802,9 @@ logicalrep_worker_detach(void)
LogicalRepWorker *w = (LogicalRepWorker *) lfirst(lc);
if (isParallelApplyWorker(w))
- logicalrep_worker_stop_internal(w, SIGTERM);
+ logicalrep_worker_stop_internal((LogicalWorkerHeader *) w,
+ SIGTERM,
+ LogicalRepWorkerLock);
}
LWLockRelease(LogicalRepWorkerLock);
@@ -794,10 +827,10 @@ logicalrep_worker_cleanup(LogicalRepWorker *worker)
Assert(LWLockHeldByMeInMode(LogicalRepWorkerLock, LW_EXCLUSIVE));
worker->type = WORKERTYPE_UNKNOWN;
- worker->in_use = false;
- worker->proc = NULL;
- worker->dbid = InvalidOid;
+ worker->hdr.in_use = false;
+ worker->hdr.proc = NULL;
worker->userid = InvalidOid;
+ worker->dbid = InvalidOid;
worker->subid = InvalidOid;
worker->relid = InvalidOid;
worker->leader_pid = InvalidPid;
@@ -931,9 +964,18 @@ ApplyLauncherRegister(void)
memset(&bgw, 0, sizeof(bgw));
bgw.bgw_flags = BGWORKER_SHMEM_ACCESS |
BGWORKER_BACKEND_DATABASE_CONNECTION;
- bgw.bgw_start_time = BgWorkerStart_RecoveryFinished;
+
+ /*
+ * The launcher now takes care of launching both logical apply workers and
+ * logical slot sync worker. Thus to cater to the requirements of both,
+ * start it as soon as a consistent state is reached. This will help
+ * slot sync worker to start timely on a physical standby while on a
+ * non-standby server, it holds same meaning as that of
+ * BgWorkerStart_RecoveryFinished.
+ */
+ bgw.bgw_start_time = BgWorkerStart_ConsistentState;
snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres");
- snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ApplyLauncherMain");
+ snprintf(bgw.bgw_function_name, BGW_MAXLEN, "LauncherMain");
snprintf(bgw.bgw_name, BGW_MAXLEN,
"logical replication launcher");
snprintf(bgw.bgw_type, BGW_MAXLEN,
@@ -1115,13 +1157,447 @@ ApplyLauncherWakeup(void)
}
/*
- * Main loop for the apply launcher process.
+ * Clean up slot sync worker info.
+ */
+static void
+slotsync_worker_cleanup(SlotSyncWorkerInfo *worker)
+{
+ Assert(LWLockHeldByMeInMode(SlotSyncWorkerLock, LW_EXCLUSIVE));
+
+ worker->hdr.in_use = false;
+ worker->hdr.proc = NULL;
+ worker->last_update_time = 0;
+}
+
+/*
+ * Attach slot sync worker to SlotSyncWorkerInfo assigned by the launcher.
*/
void
-ApplyLauncherMain(Datum main_arg)
+slotsync_worker_attach()
{
- ereport(DEBUG1,
- (errmsg_internal("logical replication launcher started")));
+ /* Block concurrent access. */
+ LWLockAcquire(SlotSyncWorkerLock, LW_EXCLUSIVE);
+
+ SlotSyncWorker = &LogicalRepCtx->ss_worker;
+
+ if (!SlotSyncWorker->hdr.in_use)
+ {
+ LWLockRelease(SlotSyncWorkerLock);
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("replication slot sync worker not initialized, cannot attach")));
+ }
+
+ if (SlotSyncWorker->hdr.proc)
+ {
+ LWLockRelease(SlotSyncWorkerLock);
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("replication slot sync worker is already running, cannot attach")));
+ }
+
+ SlotSyncWorker->hdr.proc = MyProc;
+
+ before_shmem_exit(slotsync_worker_detach, (Datum) 0);
+
+ LWLockRelease(SlotSyncWorkerLock);
+}
+
+/*
+ * Detach the slot sync worker (cleans up the worker info).
+ */
+void
+slotsync_worker_detach(int code, Datum arg)
+{
+ LWLockAcquire(SlotSyncWorkerLock, LW_EXCLUSIVE);
+ slotsync_worker_cleanup(SlotSyncWorker);
+ LWLockRelease(SlotSyncWorkerLock);
+}
+
+/*
+ * Start slot sync background worker.
+ *
+ * Returns true on success, false on failure.
+ */
+static bool
+slotsync_worker_launch()
+{
+ BackgroundWorker bgw;
+ BackgroundWorkerHandle *bgw_handle;
+ SlotSyncWorkerInfo *worker;
+ bool attach;
+
+ /* The shared memory must only be modified under lock. */
+ LWLockAcquire(SlotSyncWorkerLock, LW_EXCLUSIVE);
+
+ SlotSyncWorker = &LogicalRepCtx->ss_worker;
+
+ worker = SlotSyncWorker;
+
+ /* Prepare the new worker. */
+ worker->hdr.in_use = true;
+
+ /*
+ * 'proc' will be assigned in ReplSlotSyncWorkerMain when the worker
+ * attaches to SlotSyncWorkerInfo.
+ */
+ worker->hdr.proc = NULL;
+
+ LWLockRelease(SlotSyncWorkerLock);
+
+ /* Register the new dynamic worker. */
+ memset(&bgw, 0, sizeof(bgw));
+ bgw.bgw_flags = BGWORKER_SHMEM_ACCESS |
+ BGWORKER_BACKEND_DATABASE_CONNECTION;
+ bgw.bgw_start_time = BgWorkerStart_ConsistentState;
+ snprintf(bgw.bgw_library_name, MAXPGPATH, "postgres");
+
+ snprintf(bgw.bgw_function_name, BGW_MAXLEN, "ReplSlotSyncWorkerMain");
+
+ snprintf(bgw.bgw_name, BGW_MAXLEN, "replication slot sync worker");
+
+ snprintf(bgw.bgw_type, BGW_MAXLEN, "slot sync worker");
+
+ bgw.bgw_restart_time = BGW_NEVER_RESTART;
+ bgw.bgw_notify_pid = MyProcPid;
+
+ if (!RegisterDynamicBackgroundWorker(&bgw, &bgw_handle))
+ {
+ /* Failed to start worker, so clean up the worker slot. */
+ LWLockAcquire(SlotSyncWorkerLock, LW_EXCLUSIVE);
+ slotsync_worker_cleanup(worker);
+ LWLockRelease(SlotSyncWorkerLock);
+
+ ereport(WARNING,
+ (errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
+ errmsg("out of background worker slots"),
+ errhint("You might need to increase %s.",
+ "max_worker_processes")));
+ return false;
+ }
+
+ /* Now wait until it attaches. */
+ attach = WaitForReplicationWorkerAttach((LogicalWorkerHeader *) worker,
+ worker->hdr.generation,
+ bgw_handle,
+ SlotSyncWorkerLock);
+
+ if (!attach)
+ ereport(WARNING,
+ (errmsg("replication slot sync worker failed to attach")));
+
+ return attach;
+}
+
+/*
+ * Internal function to stop the slot sync worker and cleanup afterwards.
+ */
+static void
+slotsync_worker_stop_internal(SlotSyncWorkerInfo *worker)
+{
+ LWLockAcquire(SlotSyncWorkerLock, LW_SHARED);
+ ereport(LOG,
+ (errmsg("stopping replication slot sync worker with pid: %d",
+ worker->hdr.proc->pid)));
+ logicalrep_worker_stop_internal((LogicalWorkerHeader *) worker,
+ SIGINT,
+ SlotSyncWorkerLock);
+ LWLockRelease(SlotSyncWorkerLock);
+}
+
+/*
+ * Validate if db with given dbname exists.
+ *
+ * Can't use existing functions like 'get_database_oid' from dbcommands.c for
+ * validity purpose as they need db connection.
+ */
+static bool
+validate_dbname(const char *dbname)
+{
+ HeapTuple tuple;
+ Relation relation;
+ SysScanDesc scan;
+ ScanKeyData key[1];
+ bool valid;
+
+ /* Start a transaction so we can access pg_database */
+ StartTransactionCommand();
+
+ /* Form a scan key */
+ ScanKeyInit(&key[0], Anum_pg_database_datname, BTEqualStrategyNumber,
+ F_NAMEEQ, CStringGetDatum(dbname));
+
+ /* No db connection, force heap scan */
+ relation = table_open(DatabaseRelationId, AccessShareLock);
+ scan = systable_beginscan(relation, DatabaseNameIndexId, false,
+ NULL, 1, key);
+
+ tuple = systable_getnext(scan);
+
+ if (HeapTupleIsValid(tuple))
+ valid = true;
+ else
+ valid = false;
+
+ /* all done */
+ systable_endscan(scan);
+ table_close(relation, AccessShareLock);
+
+ CommitTransactionCommand();
+ return valid;
+}
+
+/*
+ * Checks if GUC are set appropriately before starting slot sync worker
+ */
+static bool
+slotsync_checks(long *wait_time, bool *retry)
+{
+ char *dbname;
+
+ *retry = false;
+
+ if (!enable_syncslot)
+ return false;
+
+ /*
+ * Since the above GUC is set, check that other GUC settings
+ * (primary_slot_name, hot_standby_feedback, primary_conninfo, wal_level)
+ * are compatible with slot synchronization. If not, issue warnings.
+ */
+
+ /*
+ * A physical replication slot(primary_slot_name) is required on the
+ * primary to ensure that the rows needed by the standby are not removed
+ * after restarting, so that the synchronized slot on the standby will not
+ * be invalidated.
+ */
+ if (!WalRcv || WalRcv->slotname[0] == '\0')
+ {
+ ereport(WARNING,
+ errmsg("skipping slots synchronization as primary_slot_name is not set"));
+
+ /*
+ * It's possible that the Walreceiver has not been started yet, adjust
+ * the wait_time to retry sooner in the next synchronization cycle.
+ */
+ *wait_time = wal_retrieve_retry_interval;
+
+ /*
+ * Tell caller to retry the connection for the case where
+ * primary_slot_name is set but Walreceiver is not yet started.
+ */
+ if (PrimarySlotName && strcmp(PrimarySlotName, "") != 0)
+ *retry = true;
+
+ return false;
+ }
+
+ /*
+ * Hot_standby_feedback must be enabled to cooperate with the physical
+ * replication slot, which allows informing the primary about the xmin and
+ * catalog_xmin values on the standby.
+ */
+ if (!hot_standby_feedback)
+ {
+ ereport(WARNING,
+ errmsg("skipping slots synchronization as hot_standby_feedback is off"));
+ return false;
+ }
+
+ /*
+ * Logical decoding requires wal_level >= logical and we currently only
+ * synchronize logical slots.
+ */
+ if (wal_level < WAL_LEVEL_LOGICAL)
+ {
+ ereport(WARNING,
+ errmsg("skipping slots synchronisation as it requires wal_level >= logical"));
+ return false;
+ }
+
+ /*
+ * The slot sync worker needs a database connection for walrcv_exec to
+ * work.
+ */
+ dbname = walrcv_get_dbname_from_conninfo(PrimaryConnInfo);
+ if (dbname == NULL)
+ {
+ ereport(WARNING,
+ errmsg("skipping slots synchronization as dbname is not specified in primary_conninfo"));
+ return false;
+ }
+
+ if (!validate_dbname(dbname))
+ {
+ ereport(WARNING,
+ errmsg("skipping slots synchronization as dbname specified in primary_conninfo is not a valid one"));
+ return false;
+ }
+
+ return true;
+}
+
+/*
+ * Shut down the slot sync worker.
+ */
+void
+ShutDownSlotSync(void)
+{
+ if (LogicalRepCtx->ss_worker.hdr.in_use)
+ slotsync_worker_stop_internal(&LogicalRepCtx->ss_worker);
+}
+
+/*
+ * Re-read the config file.
+ *
+ * If one of the slot sync options has changed, stop the slot sync worker
+ * and set ss_recheck flag to enable the caller to recheck slot sync GUCs
+ * before restarting the worker
+ */
+static void
+LauncherRereadConfig(bool *ss_recheck)
+{
+ char *conninfo = pstrdup(PrimaryConnInfo);
+ char *slotname = pstrdup(PrimarySlotName);
+ bool syncslot = enable_syncslot;
+ bool standbyfeedback = hot_standby_feedback;
+
+ ConfigReloadPending = false;
+ ProcessConfigFile(PGC_SIGHUP);
+
+ /*
+ * If any of the related GUCs changed, stop the slot sync worker. The
+ * worker will be relaunched in next sync-cycle using the new GUCs.
+ */
+ if ((strcmp(conninfo, PrimaryConnInfo) != 0) ||
+ (strcmp(slotname, PrimarySlotName) != 0) ||
+ (syncslot != enable_syncslot) ||
+ (standbyfeedback != hot_standby_feedback))
+ {
+ ShutDownSlotSync();
+
+ /* Retry slot sync with new GUCs */
+ *ss_recheck = true;
+ }
+
+ pfree(conninfo);
+ pfree(slotname);
+}
+
+/*
+ * Launch slot sync background worker.
+ */
+static void
+LaunchSlotSyncWorker(long *wait_time)
+{
+ LWLockAcquire(SlotSyncWorkerLock, LW_SHARED);
+
+ /* The worker is running already */
+ if (SlotSyncWorker && SlotSyncWorker->hdr.in_use &&
+ SlotSyncWorker->hdr.proc)
+ {
+ LWLockRelease(SlotSyncWorkerLock);
+ return;
+ }
+
+ LWLockRelease(SlotSyncWorkerLock);
+
+ /*
+ * If launch failed, adjust the wait_time to retry in the next sync-cycle
+ * sooner.
+ */
+ if (!slotsync_worker_launch())
+ {
+ *wait_time = Min(*wait_time, wal_retrieve_retry_interval);
+ }
+}
+
+/*
+ * Launch logical replication apply workers for enabled subscriptions.
+ */
+static void
+LaunchSubscriptionApplyWorker(long *wait_time)
+{
+ List *sublist;
+ ListCell *lc;
+ MemoryContext subctx;
+ MemoryContext oldctx;
+
+ /* Use temporary context to avoid leaking memory across cycles. */
+ subctx = AllocSetContextCreate(TopMemoryContext,
+ "Logical Replication Launcher sublist",
+ ALLOCSET_DEFAULT_SIZES);
+ oldctx = MemoryContextSwitchTo(subctx);
+
+ /* Start any missing workers for enabled subscriptions. */
+ sublist = get_subscription_list();
+ foreach(lc, sublist)
+ {
+ Subscription *sub = (Subscription *) lfirst(lc);
+ LogicalRepWorker *w;
+ TimestampTz last_start;
+ TimestampTz now;
+ long elapsed;
+
+ if (!sub->enabled)
+ continue;
+
+ LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
+ w = logicalrep_worker_find(sub->oid, InvalidOid, false);
+ LWLockRelease(LogicalRepWorkerLock);
+
+ if (w != NULL)
+ continue; /* worker is running already */
+
+ /*
+ * If the worker is eligible to start now, launch it. Otherwise,
+ * adjust wait_time so that we'll wake up as soon as it can be
+ * started.
+ *
+ * Each subscription's apply worker can only be restarted once per
+ * wal_retrieve_retry_interval, so that errors do not cause us to
+ * repeatedly restart the worker as fast as possible. In cases where
+ * a restart is expected (e.g., subscription parameter changes),
+ * another process should remove the last-start entry for the
+ * subscription so that the worker can be restarted without waiting
+ * for wal_retrieve_retry_interval to elapse.
+ */
+ last_start = ApplyLauncherGetWorkerStartTime(sub->oid);
+ now = GetCurrentTimestamp();
+ if (last_start == 0 ||
+ (elapsed = TimestampDifferenceMilliseconds(last_start, now)) >=
+ wal_retrieve_retry_interval)
+ {
+ ApplyLauncherSetWorkerStartTime(sub->oid, now);
+ logicalrep_worker_launch(WORKERTYPE_APPLY,
+ sub->dbid, sub->oid, sub->name,
+ sub->owner, InvalidOid,
+ DSM_HANDLE_INVALID);
+ }
+ else
+ {
+ *wait_time = Min(*wait_time,
+ wal_retrieve_retry_interval - elapsed);
+ }
+ }
+
+ /* Switch back to original memory context. */
+ MemoryContextSwitchTo(oldctx);
+ /* Clean the temporary memory. */
+ MemoryContextDelete(subctx);
+}
+
+/*
+ * Main loop for the launcher process.
+ */
+void
+LauncherMain(Datum main_arg)
+{
+ bool start_slotsync = false;
+ bool recheck_slotsync = true;
+
+ elog(DEBUG1, "logical replication launcher started");
before_shmem_exit(logicalrep_launcher_onexit, (Datum) 0);
@@ -1139,79 +1615,32 @@ ApplyLauncherMain(Datum main_arg)
*/
BackgroundWorkerInitializeConnection(NULL, NULL, 0);
+ load_file("libpqwalreceiver", false);
+
/* Enter main loop */
for (;;)
{
int rc;
- List *sublist;
- ListCell *lc;
- MemoryContext subctx;
- MemoryContext oldctx;
long wait_time = DEFAULT_NAPTIME_PER_CYCLE;
CHECK_FOR_INTERRUPTS();
- /* Use temporary context to avoid leaking memory across cycles. */
- subctx = AllocSetContextCreate(TopMemoryContext,
- "Logical Replication Launcher sublist",
- ALLOCSET_DEFAULT_SIZES);
- oldctx = MemoryContextSwitchTo(subctx);
-
- /* Start any missing workers for enabled subscriptions. */
- sublist = get_subscription_list();
- foreach(lc, sublist)
+ /*
+ * If it is Hot standby, then try to launch slot sync worker else
+ * launch apply workers.
+ */
+ if (RecoveryInProgress() && !PromoteIsTriggered())
{
- Subscription *sub = (Subscription *) lfirst(lc);
- LogicalRepWorker *w;
- TimestampTz last_start;
- TimestampTz now;
- long elapsed;
-
- if (!sub->enabled)
- continue;
-
- LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
- w = logicalrep_worker_find(sub->oid, InvalidOid, false);
- LWLockRelease(LogicalRepWorkerLock);
+ /* Make validation checks first */
+ if (recheck_slotsync)
+ start_slotsync = slotsync_checks(&wait_time, &recheck_slotsync);
- if (w != NULL)
- continue; /* worker is running already */
-
- /*
- * If the worker is eligible to start now, launch it. Otherwise,
- * adjust wait_time so that we'll wake up as soon as it can be
- * started.
- *
- * Each subscription's apply worker can only be restarted once per
- * wal_retrieve_retry_interval, so that errors do not cause us to
- * repeatedly restart the worker as fast as possible. In cases
- * where a restart is expected (e.g., subscription parameter
- * changes), another process should remove the last-start entry
- * for the subscription so that the worker can be restarted
- * without waiting for wal_retrieve_retry_interval to elapse.
- */
- last_start = ApplyLauncherGetWorkerStartTime(sub->oid);
- now = GetCurrentTimestamp();
- if (last_start == 0 ||
- (elapsed = TimestampDifferenceMilliseconds(last_start, now)) >= wal_retrieve_retry_interval)
- {
- ApplyLauncherSetWorkerStartTime(sub->oid, now);
- logicalrep_worker_launch(WORKERTYPE_APPLY,
- sub->dbid, sub->oid, sub->name,
- sub->owner, InvalidOid,
- DSM_HANDLE_INVALID);
- }
- else
- {
- wait_time = Min(wait_time,
- wal_retrieve_retry_interval - elapsed);
- }
+ /* Start slot sync workers if checks passed */
+ if (start_slotsync)
+ LaunchSlotSyncWorker(&wait_time);
}
-
- /* Switch back to original memory context. */
- MemoryContextSwitchTo(oldctx);
- /* Clean the temporary memory. */
- MemoryContextDelete(subctx);
+ else
+ LaunchSubscriptionApplyWorker(&wait_time);
/* Wait for more work. */
rc = WaitLatch(MyLatch,
@@ -1226,10 +1655,7 @@ ApplyLauncherMain(Datum main_arg)
}
if (ConfigReloadPending)
- {
- ConfigReloadPending = false;
- ProcessConfigFile(PGC_SIGHUP);
- }
+ LauncherRereadConfig(&recheck_slotsync);
}
/* Not reachable */
@@ -1260,7 +1686,8 @@ GetLeaderApplyWorkerPid(pid_t pid)
{
LogicalRepWorker *w = &LogicalRepCtx->workers[i];
- if (isParallelApplyWorker(w) && w->proc && pid == w->proc->pid)
+ if (isParallelApplyWorker(w) && w->hdr.proc &&
+ pid == w->hdr.proc->pid)
{
leader_pid = w->leader_pid;
break;
@@ -1298,13 +1725,13 @@ pg_stat_get_subscription(PG_FUNCTION_ARGS)
memcpy(&worker, &LogicalRepCtx->workers[i],
sizeof(LogicalRepWorker));
- if (!worker.proc || !IsBackendPid(worker.proc->pid))
+ if (!worker.hdr.proc || !IsBackendPid(worker.hdr.proc->pid))
continue;
if (OidIsValid(subid) && worker.subid != subid)
continue;
- worker_pid = worker.proc->pid;
+ worker_pid = worker.hdr.proc->pid;
values[0] = ObjectIdGetDatum(worker.subid);
if (isTablesyncWorker(&worker))
diff --git a/src/backend/replication/logical/logical.c b/src/backend/replication/logical/logical.c
index 8288da5277..5bef0138b4 100644
--- a/src/backend/replication/logical/logical.c
+++ b/src/backend/replication/logical/logical.c
@@ -524,6 +524,28 @@ CreateDecodingContext(XLogRecPtr start_lsn,
errmsg("replication slot \"%s\" was not created in this database",
NameStr(slot->data.name))));
+ /*
+ * Slots in state SYNCSLOT_STATE_INITIATED should have been dropped on
+ * promotion.
+ */
+ if (!RecoveryInProgress() && slot->data.sync_state == SYNCSLOT_STATE_INITIATED)
+ elog(ERROR, "replication slot \"%s\" was not synced completely from the primary server",
+ NameStr(slot->data.name));
+
+ /*
+ * Do not allow consumption of a "synchronized" slot until the standby
+ * gets promoted. Also do not allow consumption of slots with sync_state
+ * as SYNCSLOT_STATE_INITIATED as they are not synced completely to be
+ * used.
+ */
+ if (RecoveryInProgress() && slot->data.sync_state != SYNCSLOT_STATE_NONE)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot use replication slot \"%s\" for logical decoding",
+ NameStr(slot->data.name)),
+ errdetail("This slot is being synced from the primary server."),
+ errhint("Specify another replication slot.")));
+
/*
* Check if slot has been invalidated due to max_slot_wal_keep_size. Avoid
* "cannot get changes" wording in this errmsg because that'd be
diff --git a/src/backend/replication/logical/meson.build b/src/backend/replication/logical/meson.build
index d48cd4c590..9e52ec421f 100644
--- a/src/backend/replication/logical/meson.build
+++ b/src/backend/replication/logical/meson.build
@@ -11,6 +11,7 @@ backend_sources += files(
'proto.c',
'relation.c',
'reorderbuffer.c',
+ 'slotsync.c',
'snapbuild.c',
'tablesync.c',
'worker.c',
diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c
new file mode 100644
index 0000000000..9aa3562500
--- /dev/null
+++ b/src/backend/replication/logical/slotsync.c
@@ -0,0 +1,1018 @@
+/*-------------------------------------------------------------------------
+ * slotsync.c
+ * PostgreSQL worker for synchronizing slots to a standby server from the
+ * primary server.
+ *
+ * Copyright (c) 2023, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * src/backend/replication/logical/slotsync.c
+ *
+ * This file contains the code for slot sync worker on a physical standby
+ * to fetch logical failover slots information from the primary server,
+ * create the slots on the standby and synchronize them periodically.
+ *
+ * It also takes care of dropping the slots which were created by it and are
+ * currently not needed to be synchronized.
+ *
+ * It takes a nap of WORKER_DEFAULT_NAPTIME_MS before every next
+ * synchronization. If there is no activity observed on the primary server for
+ * some time, the nap time is increased to WORKER_INACTIVITY_NAPTIME_MS, but if
+ * any activity is observed, the nap time reverts to the default value.
+ *---------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "access/xlogrecovery.h"
+#include "commands/dbcommands.h"
+#include "pgstat.h"
+#include "postmaster/bgworker.h"
+#include "postmaster/interrupt.h"
+#include "replication/logical.h"
+#include "replication/logicallauncher.h"
+#include "replication/logicalworker.h"
+#include "replication/walreceiver.h"
+#include "replication/worker_internal.h"
+#include "storage/ipc.h"
+#include "storage/procarray.h"
+#include "tcop/tcopprot.h"
+#include "utils/builtins.h"
+#include "utils/guc_hooks.h"
+#include "utils/pg_lsn.h"
+#include "utils/varlena.h"
+
+/*
+ * Structure to hold information fetched from the primary server about a logical
+ * replication slot.
+ */
+typedef struct RemoteSlot
+{
+ char *name;
+ char *plugin;
+ char *database;
+ bool two_phase;
+ bool failover;
+ XLogRecPtr restart_lsn;
+ XLogRecPtr confirmed_lsn;
+ TransactionId catalog_xmin;
+
+ /* RS_INVAL_NONE if valid, or the reason of invalidation */
+ ReplicationSlotInvalidationCause invalidated;
+} RemoteSlot;
+
+/* Worker's nap time in case of regular activity on the primary server */
+#define WORKER_DEFAULT_NAPTIME_MS 10L /* 10 ms */
+
+/* Worker's nap time in case of no-activity on the primary server */
+#define WORKER_INACTIVITY_NAPTIME_MS 10000L /* 10 sec */
+
+/*
+ * Inactivity Threshold in ms before increasing nap time of worker.
+ *
+ * If the lsn of slot being monitored did not change for this threshold time,
+ * then increase nap time of current worker from WORKER_DEFAULT_NAPTIME_MS to
+ * WORKER_INACTIVITY_NAPTIME_MS.
+ */
+#define WORKER_INACTIVITY_THRESHOLD_MS 10000L /* 10 sec */
+
+/*
+ * Number of attempts for wait_for_primary_slot_catchup() after
+ * which it aborts the wait and the slot sync worker then moves
+ * to the next slot creation/sync.
+ */
+#define WORKER_PRIMARY_CATCHUP_WAIT_ATTEMPTS 5
+
+/*
+ * Wait for remote slot to pass locally reserved position.
+ *
+ * Ping and wait for the primary server for
+ * WORKER_PRIMARY_CATCHUP_WAIT_ATTEMPTS during a slot creation, if it still
+ * does not catch up, abort the wait. The ones for which wait is aborted will
+ * attempt the wait and sync in the next sync-cycle.
+ *
+ * *persist will be set to false if the slot has disappeared or was invalidated
+ * on the primary; otherwise, it will be set to true.
+ */
+static bool
+wait_for_primary_slot_catchup(WalReceiverConn *wrconn, RemoteSlot *remote_slot,
+ bool *persist)
+{
+#define WAIT_OUTPUT_COLUMN_COUNT 4
+ StringInfoData cmd;
+ int wait_count = 0;
+
+ if (persist)
+ *persist = true;
+
+ ereport(LOG,
+ errmsg("waiting for remote slot \"%s\" LSN (%X/%X) and catalog xmin"
+ " (%u) to pass local slot LSN (%X/%X) and catalog xmin (%u)",
+ remote_slot->name,
+ LSN_FORMAT_ARGS(remote_slot->restart_lsn),
+ remote_slot->catalog_xmin,
+ LSN_FORMAT_ARGS(MyReplicationSlot->data.restart_lsn),
+ MyReplicationSlot->data.catalog_xmin));
+
+ initStringInfo(&cmd);
+ appendStringInfo(&cmd,
+ "SELECT conflicting, restart_lsn, confirmed_flush_lsn,"
+ " catalog_xmin FROM pg_catalog.pg_replication_slots"
+ " WHERE slot_name = %s",
+ quote_literal_cstr(remote_slot->name));
+
+ for (;;)
+ {
+ XLogRecPtr new_invalidated;
+ XLogRecPtr new_restart_lsn;
+ XLogRecPtr new_confirmed_lsn;
+ TransactionId new_catalog_xmin;
+ WalRcvExecResult *res;
+ TupleTableSlot *slot;
+ int rc;
+ bool isnull;
+ Oid slotRow[WAIT_OUTPUT_COLUMN_COUNT] = {BOOLOID, LSNOID, LSNOID,
+ XIDOID};
+
+ CHECK_FOR_INTERRUPTS();
+
+ Assert(RecoveryInProgress());
+
+ res = walrcv_exec(wrconn, cmd.data, WAIT_OUTPUT_COLUMN_COUNT, slotRow);
+
+ if (res->status != WALRCV_OK_TUPLES)
+ ereport(ERROR,
+ (errmsg("could not fetch slot info for slot \"%s\" from the"
+ " primary server: %s",
+ remote_slot->name, res->err)));
+
+ slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+ if (!tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+ {
+ ereport(WARNING,
+ (errmsg("slot \"%s\" disappeared from the primary server,"
+ " slot creation aborted", remote_slot->name)));
+ pfree(cmd.data);
+ walrcv_clear_result(res);
+
+ /*
+ * The slot being created will be dropped when it is released (see
+ * ReplicationSlotRelease).
+ */
+ if (persist)
+ *persist = false;
+
+ return false;
+ }
+
+
+ /*
+ * It is possible to get null values for LSN and Xmin if slot is
+ * invalidated on the primary server, so handle accordingly.
+ */
+ new_invalidated = DatumGetBool(slot_getattr(slot, 1, &isnull));
+ Assert(!isnull);
+
+ new_restart_lsn = DatumGetLSN(slot_getattr(slot, 2, &isnull));
+ if (new_invalidated || isnull)
+ {
+ ereport(WARNING,
+ (errmsg("slot \"%s\" invalidated on the primary server,"
+ " slot creation aborted", remote_slot->name)));
+ pfree(cmd.data);
+ ExecClearTuple(slot);
+ walrcv_clear_result(res);
+
+ /*
+ * The slot being created will be dropped when it is released (see
+ * ReplicationSlotRelease).
+ */
+ if (persist)
+ *persist = false;
+
+ return false;
+ }
+
+ /*
+ * Once we got valid restart_lsn, then confirmed_lsn and catalog_xmin
+ * are expected to be valid/non-null.
+ */
+ new_confirmed_lsn = DatumGetLSN(slot_getattr(slot, 3, &isnull));
+ Assert(!isnull);
+
+ new_catalog_xmin = DatumGetTransactionId(slot_getattr(slot,
+ 4, &isnull));
+ Assert(!isnull);
+
+ ExecClearTuple(slot);
+ walrcv_clear_result(res);
+
+ if (new_restart_lsn >= MyReplicationSlot->data.restart_lsn &&
+ TransactionIdFollowsOrEquals(new_catalog_xmin,
+ MyReplicationSlot->data.catalog_xmin))
+ {
+ /* Update new values in remote_slot */
+ remote_slot->restart_lsn = new_restart_lsn;
+ remote_slot->confirmed_lsn = new_confirmed_lsn;
+ remote_slot->catalog_xmin = new_catalog_xmin;
+
+ ereport(LOG,
+ errmsg("wait over for remote slot \"%s\" as its LSN (%X/%X)"
+ " and catalog xmin (%u) has now passed local slot LSN"
+ " (%X/%X) and catalog xmin (%u)",
+ remote_slot->name,
+ LSN_FORMAT_ARGS(new_restart_lsn),
+ new_catalog_xmin,
+ LSN_FORMAT_ARGS(MyReplicationSlot->data.restart_lsn),
+ MyReplicationSlot->data.catalog_xmin));
+ pfree(cmd.data);
+
+ return true;
+ }
+
+ if (++wait_count >= WORKER_PRIMARY_CATCHUP_WAIT_ATTEMPTS)
+ {
+ ereport(LOG,
+ errmsg("aborting the wait for remote slot \"%s\" and moving"
+ " to the next slot, will attempt creating it again",
+ remote_slot->name));
+ pfree(cmd.data);
+
+ return false;
+ }
+
+ /*
+ * XXX: Is waiting for 2 seconds before retrying enough or more or
+ * less?
+ */
+ rc = WaitLatch(MyLatch,
+ WL_LATCH_SET | WL_TIMEOUT | WL_POSTMASTER_DEATH,
+ 2000L,
+ WAIT_EVENT_REPL_SLOTSYNC_PRIMARY_CATCHUP);
+
+ ResetLatch(MyLatch);
+
+ /* Emergency bailout if postmaster has died */
+ if (rc & WL_POSTMASTER_DEATH)
+ proc_exit(1);
+ }
+}
+
+/*
+ * Update local slot metadata as per remote_slot's positions
+ */
+static void
+local_slot_update(RemoteSlot *remote_slot)
+{
+ Assert(MyReplicationSlot->data.invalidated == RS_INVAL_NONE);
+
+ LogicalConfirmReceivedLocation(remote_slot->confirmed_lsn);
+ LogicalIncreaseXminForSlot(remote_slot->confirmed_lsn,
+ remote_slot->catalog_xmin);
+ LogicalIncreaseRestartDecodingForSlot(remote_slot->confirmed_lsn,
+ remote_slot->restart_lsn);
+
+ SpinLockAcquire(&MyReplicationSlot->mutex);
+ MyReplicationSlot->data.invalidated = remote_slot->invalidated;
+ SpinLockRelease(&MyReplicationSlot->mutex);
+
+ ReplicationSlotMarkDirty();
+}
+
+/*
+ * Drop the slots for which sync is initiated but not yet completed
+ * i.e. they are still waiting for the primary server to catch up.
+ */
+void
+slotsync_drop_initiated_slots(void)
+{
+ List *slots = NIL;
+ ListCell *lc;
+
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+
+ for (int i = 0; i < max_replication_slots; i++)
+ {
+ ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
+
+ if (s->in_use && s->data.sync_state == SYNCSLOT_STATE_INITIATED)
+ slots = lappend(slots, s);
+ }
+
+ LWLockRelease(ReplicationSlotControlLock);
+
+ foreach(lc, slots)
+ {
+ ReplicationSlot *s = (ReplicationSlot *) lfirst(lc);
+
+ ReplicationSlotDrop(NameStr(s->data.name), true, false);
+ ereport(LOG,
+ (errmsg("dropped replication slot \"%s\" of dbid %d as it "
+ "was not sync-ready", NameStr(s->data.name),
+ s->data.database)));
+ }
+
+ list_free(slots);
+}
+
+/*
+ * Get list of local logical slot names which are synchronized from
+ * the primary server.
+ */
+static List *
+get_local_synced_slot_names(void)
+{
+ List *localSyncedSlots = NIL;
+
+ LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
+
+ for (int i = 0; i < max_replication_slots; i++)
+ {
+ ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
+
+ /* Check if it is logical synchronized slot */
+ if (s->in_use && SlotIsLogical(s) &&
+ (s->data.sync_state != SYNCSLOT_STATE_NONE))
+ {
+ localSyncedSlots = lappend(localSyncedSlots, s);
+ }
+ }
+
+ LWLockRelease(ReplicationSlotControlLock);
+
+ return localSyncedSlots;
+}
+
+/*
+ * Helper function to check if local_slot is present in remote_slots list.
+ *
+ * It also checks if logical slot is locally invalidated i.e. invalidated on
+ * the standby but valid on the primary server. If found so, it sets
+ * locally_invalidated to true.
+ */
+static bool
+validate_sync_slot(ReplicationSlot *local_slot, List *remote_slots,
+ bool *locally_invalidated)
+{
+ ListCell *cell;
+
+ foreach(cell, remote_slots)
+ {
+ RemoteSlot *remote_slot = (RemoteSlot *) lfirst(cell);
+
+ if (strcmp(remote_slot->name, NameStr(local_slot->data.name)) == 0)
+ {
+ /*
+ * If remote slot is not invalidated but local slot is marked as
+ * invalidated, then set the bool.
+ */
+ *locally_invalidated =
+ (remote_slot->invalidated == RS_INVAL_NONE) &&
+ (local_slot->data.invalidated != RS_INVAL_NONE);
+
+ return true;
+ }
+ }
+
+ return false;
+}
+
+/*
+ * Drop obsolete slots
+ *
+ * Drop the slots that no longer need to be synced i.e. these either do not
+ * exist on the primary or are no longer enabled for failover.
+ *
+ * Also drop the slots that are valid on the primary that got invalidated
+ * on the standby due to conflict (say required rows removed on the primary).
+ * The assumption is, that these will get recreated in next sync-cycle and
+ * it is okay to drop and recreate such slots as long as these are not
+ * consumable on the standby (which is the case currently).
+ */
+static void
+drop_obsolete_slots(List *remote_slot_list)
+{
+ List *local_slot_list = NIL;
+ ListCell *lc_slot;
+
+ /*
+ * Get the list of local 'synced' slot so that those not on remote could
+ * be dropped.
+ */
+ local_slot_list = get_local_synced_slot_names();
+
+ foreach(lc_slot, local_slot_list)
+ {
+ ReplicationSlot *local_slot = (ReplicationSlot *) lfirst(lc_slot);
+ bool local_exists = false;
+ bool locally_invalidated = false;
+
+ local_exists = validate_sync_slot(local_slot, remote_slot_list,
+ &locally_invalidated);
+
+ /*
+ * Drop the local slot either if it is not in the remote slots list or
+ * is invalidated while remote slot is still valid.
+ */
+ if (!local_exists || locally_invalidated)
+ {
+ ReplicationSlotDrop(NameStr(local_slot->data.name), true, false);
+
+ ereport(LOG,
+ (errmsg("dropped replication slot \"%s\" of dbid %d",
+ NameStr(local_slot->data.name),
+ local_slot->data.database)));
+ }
+ }
+}
+
+/*
+ * Constructs the query in order to get failover logical slots
+ * information from the primary server.
+ */
+static void
+construct_slot_query(StringInfo s)
+{
+ /*
+ * Fetch slots with failover enabled.
+ *
+ * If we are on cascading standby, we should fetch only those slots from
+ * the first standby which have sync_state as either 'n' or 'r'. Slots
+ * with sync_state as 'i' are not sync ready yet. And when we are on the
+ * first standby, the primary server is supposed to have slots with
+ * sync_state as 'n' only (or it may have all 'n', 'r' and 'i' if standby
+ * is promoted as primary). Thus in all the cases, filter sync_state !='i'
+ * is appropriate one.
+ */
+ appendStringInfo(s,
+ "SELECT slot_name, plugin, confirmed_flush_lsn,"
+ " restart_lsn, catalog_xmin, two_phase, failover,"
+ " database, pg_get_slot_invalidation_cause(slot_name)"
+ " FROM pg_catalog.pg_replication_slots"
+ " WHERE failover and sync_state != 'i'");
+}
+
+/*
+ * Synchronize single slot to given position.
+ *
+ * This creates a new slot if there is no existing one and updates the
+ * metadata of the slot as per the data received from the primary server.
+ *
+ * The 'sync_state' in slot.data is set to SYNCSLOT_STATE_INITIATED
+ * immediately after creation. It stays in same state until the
+ * initialization is complete. The initialization is considered to
+ * be completed once the remote_slot catches up with locally reserved
+ * position and local slot is updated. The sync_state is then changed
+ * to SYNCSLOT_STATE_READY.
+ */
+static void
+synchronize_one_slot(WalReceiverConn *wrconn, RemoteSlot *remote_slot,
+ bool *slot_updated)
+{
+ ReplicationSlot *s;
+ char sync_state = 0;
+
+ /*
+ * Make sure that concerned WAL is received before syncing slot to target
+ * lsn received from the primary server.
+ *
+ * This check should never pass as on the primary server, we have waited
+ * for the standby's confirmation before updating the logical slot.
+ */
+ if (remote_slot->confirmed_lsn > WalRcv->latestWalEnd)
+ {
+ elog(ERROR, "skipping sync of slot \"%s\" as the received slot sync "
+ "LSN %X/%X is ahead of the standby position %X/%X",
+ remote_slot->name,
+ LSN_FORMAT_ARGS(remote_slot->confirmed_lsn),
+ LSN_FORMAT_ARGS(WalRcv->latestWalEnd));
+
+ return;
+ }
+
+ /* Search for the named slot */
+ if ((s = SearchNamedReplicationSlot(remote_slot->name, true)))
+ {
+ SpinLockAcquire(&s->mutex);
+ sync_state = s->data.sync_state;
+ SpinLockRelease(&s->mutex);
+ }
+
+ StartTransactionCommand();
+
+ /*
+ * Already existing slot (created by slot sync worker) and ready for sync,
+ * acquire and sync it.
+ */
+ if (sync_state == SYNCSLOT_STATE_READY)
+ {
+ ReplicationSlotAcquire(remote_slot->name, true);
+
+ /*
+ * Copy the invalidation cause from remote only if local slot is not
+ * invalidated locally, we don't want to overwrite existing one.
+ */
+ if (MyReplicationSlot->data.invalidated == RS_INVAL_NONE)
+ {
+ SpinLockAcquire(&MyReplicationSlot->mutex);
+ MyReplicationSlot->data.invalidated = remote_slot->invalidated;
+ SpinLockRelease(&MyReplicationSlot->mutex);
+ }
+
+ /* Skip the sync if slot has been invalidated locally. */
+ if (MyReplicationSlot->data.invalidated != RS_INVAL_NONE)
+ goto cleanup;
+
+ /*
+ * With hot_standby_feedback enabled and invalidations handled
+ * apropriately as above, this should never happen.
+ */
+ if (remote_slot->restart_lsn < MyReplicationSlot->data.restart_lsn)
+ {
+ ereport(ERROR,
+ errmsg("not synchronizing local slot \"%s\" LSN(%X/%X)"
+ " to remote slot's LSN(%X/%X) as synchronization "
+ " would move it backwards", remote_slot->name,
+ LSN_FORMAT_ARGS(MyReplicationSlot->data.restart_lsn),
+ LSN_FORMAT_ARGS(remote_slot->restart_lsn)));
+
+ goto cleanup;
+ }
+
+ if (remote_slot->confirmed_lsn != MyReplicationSlot->data.confirmed_flush ||
+ remote_slot->restart_lsn != MyReplicationSlot->data.restart_lsn ||
+ remote_slot->catalog_xmin != MyReplicationSlot->data.catalog_xmin)
+ {
+ /* Update LSN of slot to remote slot's current position */
+ local_slot_update(remote_slot);
+ ReplicationSlotSave();
+ *slot_updated = true;
+ }
+ }
+
+ /*
+ * Already existing slot but not ready (i.e. waiting for the primary
+ * server to catch-up), lets attempt to make it sync-ready now.
+ */
+ else if (sync_state == SYNCSLOT_STATE_INITIATED)
+ {
+ ReplicationSlotAcquire(remote_slot->name, true);
+
+ /* Skip the sync if slot has been invalidated locally. */
+ if (MyReplicationSlot->data.invalidated != RS_INVAL_NONE)
+ goto cleanup;
+
+ /*
+ * Refer the slot creation part (last 'else' block) for more details
+ * on this wait.
+ */
+ if (remote_slot->restart_lsn < MyReplicationSlot->data.restart_lsn ||
+ TransactionIdPrecedes(remote_slot->catalog_xmin,
+ MyReplicationSlot->data.catalog_xmin))
+ {
+ if (!wait_for_primary_slot_catchup(wrconn, remote_slot, NULL))
+ {
+ goto cleanup;
+ }
+ }
+
+ /*
+ * Wait for primary is over, update the lsns and mark the slot as
+ * READY for further syncs.
+ */
+ local_slot_update(remote_slot);
+ SpinLockAcquire(&MyReplicationSlot->mutex);
+ MyReplicationSlot->data.sync_state = SYNCSLOT_STATE_READY;
+ SpinLockRelease(&MyReplicationSlot->mutex);
+
+ /* Save the changes */
+ ReplicationSlotMarkDirty();
+ ReplicationSlotSave();
+
+ *slot_updated = true;
+
+ ereport(LOG, errmsg("newly locally created slot \"%s\" is sync-ready now",
+ remote_slot->name));
+ }
+ /* User created slot with the same name exists, raise ERROR. */
+ else if (sync_state == SYNCSLOT_STATE_NONE)
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("skipping sync of slot \"%s\" as it is a user created"
+ " slot", remote_slot->name),
+ errdetail("This slot has failover enabled on the primary and"
+ " thus is sync candidate but user created slot with"
+ " the same name already exists on the standby")));
+ }
+ /* Otherwise create the slot first. */
+ else
+ {
+ TransactionId xmin_horizon = InvalidTransactionId;
+ ReplicationSlot *slot;
+
+ ReplicationSlotCreate(remote_slot->name, true, RS_EPHEMERAL,
+ remote_slot->two_phase,
+ remote_slot->failover);
+ slot = MyReplicationSlot;
+
+ SpinLockAcquire(&slot->mutex);
+ slot->data.database = get_database_oid(remote_slot->database, false);
+
+ namestrcpy(&slot->data.plugin, remote_slot->plugin);
+ SpinLockRelease(&slot->mutex);
+
+ ReplicationSlotReserveWal();
+
+ LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
+ xmin_horizon = GetOldestSafeDecodingTransactionId(true);
+ SpinLockAcquire(&slot->mutex);
+ slot->effective_catalog_xmin = xmin_horizon;
+ slot->data.catalog_xmin = xmin_horizon;
+ SpinLockRelease(&slot->mutex);
+ ReplicationSlotsComputeRequiredXmin(true);
+ LWLockRelease(ProcArrayLock);
+
+ /*
+ * If the local restart_lsn and/or local catalog_xmin is ahead of
+ * those on the remote then we cannot create the local slot in sync
+ * with the primary server because that would mean moving the local
+ * slot backwards and we might not have WALs retained for old LSN. In
+ * this case we will wait for the primary server's restart_lsn and
+ * catalog_xmin to catch up with the local one before attempting the
+ * sync.
+ */
+ if (remote_slot->restart_lsn < MyReplicationSlot->data.restart_lsn ||
+ TransactionIdPrecedes(remote_slot->catalog_xmin,
+ MyReplicationSlot->data.catalog_xmin))
+ {
+ bool persist;
+
+ if (!wait_for_primary_slot_catchup(wrconn, remote_slot, &persist))
+ {
+ /*
+ * The remote slot didn't catch up to locally reserved
+ * position.
+ *
+ * We do not drop the slot because the restart_lsn can be
+ * ahead of the current location when recreating the slot in
+ * the next cycle. It may take more time to create such a
+ * slot. Therefore, we persist it (provided remote-slot is
+ * still valid) and attempt the wait and synchronization in
+ * the next cycle.
+ */
+ if (persist)
+ {
+ /* Mark it as sync initiated */
+ SpinLockAcquire(&MyReplicationSlot->mutex);
+ slot->data.sync_state = SYNCSLOT_STATE_INITIATED;
+ SpinLockRelease(&MyReplicationSlot->mutex);
+
+ ReplicationSlotPersist();
+ *slot_updated = true;
+ }
+
+ goto cleanup;
+ }
+ }
+
+
+ /*
+ * Wait for primary is either not needed or is over. Update the lsns
+ * and mark the slot as READY for further syncs.
+ */
+ local_slot_update(remote_slot);
+ SpinLockAcquire(&MyReplicationSlot->mutex);
+ MyReplicationSlot->data.sync_state = SYNCSLOT_STATE_READY;
+ SpinLockRelease(&MyReplicationSlot->mutex);
+
+ /* Mark the slot as PERSISTENT and save the changes to disk */
+ ReplicationSlotPersist();
+ *slot_updated = true;
+
+ ereport(LOG, errmsg("newly locally created slot \"%s\" is sync-ready now",
+ remote_slot->name));
+ }
+
+cleanup:
+
+ ReplicationSlotRelease();
+ CommitTransactionCommand();
+
+ return;
+}
+
+/*
+ * Synchronize slots.
+ *
+ * Gets the failover logical slots info from the primary server and update
+ * the slots locally. Creates the slots if not present on the standby.
+ *
+ * Returns nap time for the next sync-cycle.
+ */
+static long
+synchronize_slots(WalReceiverConn *wrconn)
+{
+#define SLOTSYNC_COLUMN_COUNT 9
+ Oid slotRow[SLOTSYNC_COLUMN_COUNT] = {TEXTOID, TEXTOID, LSNOID,
+ LSNOID, XIDOID, BOOLOID, BOOLOID, TEXTOID, INT2OID};
+
+ WalRcvExecResult *res;
+ TupleTableSlot *slot;
+ StringInfoData s;
+ List *remote_slot_list = NIL;
+ MemoryContext oldctx = CurrentMemoryContext;
+ long naptime = WORKER_DEFAULT_NAPTIME_MS;
+ ListCell *cell;
+ bool slot_updated = false;
+ TimestampTz now;
+
+ /* The primary_slot_name is not set yet or WALs not received yet */
+ if (!WalRcv ||
+ (WalRcv->slotname[0] == '\0') ||
+ XLogRecPtrIsInvalid(WalRcv->latestWalEnd))
+ return naptime;
+
+ /* The syscache access needs a transaction env. */
+ StartTransactionCommand();
+
+ /*
+ * Make result tuples live outside TopTransactionContext to make them
+ * accessible even after transaction is committed.
+ */
+ MemoryContextSwitchTo(oldctx);
+
+ /* Construct query to get slots info from the primary server */
+ initStringInfo(&s);
+ construct_slot_query(&s);
+
+ elog(DEBUG2, "slot sync worker's query:%s \n", s.data);
+
+ /* Execute the query */
+ res = walrcv_exec(wrconn, s.data, SLOTSYNC_COLUMN_COUNT, slotRow);
+ pfree(s.data);
+
+ if (res->status != WALRCV_OK_TUPLES)
+ ereport(ERROR,
+ (errmsg("could not fetch failover logical slots info "
+ "from the primary server: %s", res->err)));
+
+ CommitTransactionCommand();
+
+ /* Switch to oldctx we saved */
+ MemoryContextSwitchTo(oldctx);
+
+ /* Construct the remote_slot tuple and synchronize each slot locally */
+ slot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
+ while (tuplestore_gettupleslot(res->tuplestore, true, false, slot))
+ {
+ bool isnull;
+ RemoteSlot *remote_slot = palloc0(sizeof(RemoteSlot));
+
+ remote_slot->name = TextDatumGetCString(slot_getattr(slot, 1, &isnull));
+ Assert(!isnull);
+
+ remote_slot->plugin = TextDatumGetCString(slot_getattr(slot, 2, &isnull));
+ Assert(!isnull);
+
+ /*
+ * It is possible to get null values for LSN and Xmin if slot is
+ * invalidated on the primary server, so handle accordingly.
+ */
+ remote_slot->confirmed_lsn = DatumGetLSN(slot_getattr(slot, 3, &isnull));
+ if (isnull)
+ remote_slot->confirmed_lsn = InvalidXLogRecPtr;
+
+ remote_slot->restart_lsn = DatumGetLSN(slot_getattr(slot, 4, &isnull));
+ if (isnull)
+ remote_slot->restart_lsn = InvalidXLogRecPtr;
+
+ remote_slot->catalog_xmin = DatumGetTransactionId(slot_getattr(slot,
+ 5, &isnull));
+ if (isnull)
+ remote_slot->catalog_xmin = InvalidTransactionId;
+
+ remote_slot->two_phase = DatumGetBool(slot_getattr(slot, 6, &isnull));
+ Assert(!isnull);
+
+ remote_slot->failover = DatumGetBool(slot_getattr(slot, 7, &isnull));
+ Assert(!isnull);
+
+ remote_slot->database = TextDatumGetCString(slot_getattr(slot,
+ 8, &isnull));
+ Assert(!isnull);
+
+ remote_slot->invalidated = DatumGetInt16(slot_getattr(slot, 9, &isnull));
+ Assert(!isnull);
+
+ /* Create list of remote slots */
+ remote_slot_list = lappend(remote_slot_list, remote_slot);
+
+ ExecClearTuple(slot);
+ }
+
+ /*
+ * Drop local slots that no longer need to be synced. Do it before
+ * synchronize_one_slot to allow dropping of slots before actual sync
+ * which are invalidated locally while still valid on the primary server.
+ */
+ drop_obsolete_slots(remote_slot_list);
+
+ /* Now sync the slots locally */
+ foreach(cell, remote_slot_list)
+ {
+ RemoteSlot *remote_slot = (RemoteSlot *) lfirst(cell);
+
+ synchronize_one_slot(wrconn, remote_slot, &slot_updated);
+ }
+
+ now = GetCurrentTimestamp();
+
+ /*
+ * If any of the slots get updated in this sync-cycle, retain default
+ * naptime and update 'last_update_time' in slot sync worker. But if no
+ * activity is observed in this sync-cycle, then increase naptime provided
+ * inactivity time reaches threshold.
+ */
+ if (slot_updated)
+ SlotSyncWorker->last_update_time = now;
+
+ else if (TimestampDifferenceExceeds(SlotSyncWorker->last_update_time,
+ now, WORKER_INACTIVITY_THRESHOLD_MS))
+ naptime = WORKER_INACTIVITY_NAPTIME_MS;
+
+ /* We are done, free remote_slot_list elements */
+ list_free_deep(remote_slot_list);
+
+ walrcv_clear_result(res);
+
+ return naptime;
+}
+
+/*
+ * Connect to the remote (primary) server.
+ *
+ * This uses GUC primary_conninfo in order to connect to the primary.
+ * For slot sync to work, primary_conninfo is required to specify dbname
+ * as well.
+ */
+static WalReceiverConn *
+remote_connect(void)
+{
+ WalReceiverConn *wrconn = NULL;
+ char *err;
+
+ wrconn = walrcv_connect(PrimaryConnInfo, true, false,
+ cluster_name[0] ? cluster_name : "slotsyncworker", &err);
+ if (wrconn == NULL)
+ ereport(ERROR,
+ (errcode(ERRCODE_CONNECTION_FAILURE),
+ errmsg("could not connect to the primary server: %s", err)));
+ return wrconn;
+}
+
+/*
+ * Re-read the config file.
+ *
+ * If primary_conninfo has changed, reconnect to primary.
+ */
+static void
+slotsync_reread_config(WalReceiverConn **wrconn)
+{
+ char *conninfo = pstrdup(PrimaryConnInfo);
+
+ ConfigReloadPending = false;
+ ProcessConfigFile(PGC_SIGHUP);
+
+ /* Exit if GUC primary_conninfo got changed, let the launcher relaunch it */
+ if (strcmp(conninfo, PrimaryConnInfo) != 0)
+ {
+ if (*wrconn)
+ walrcv_disconnect(*wrconn);
+
+ pfree(conninfo);
+
+ ereport(LOG,
+ (errmsg("replication slot sync worker will stop because the "
+ "primary_conninfo has changed")));
+ proc_exit(0);
+ }
+
+ pfree(conninfo);
+}
+
+/*
+ * Interrupt handler for main loop of slot sync worker.
+ */
+static void
+ProcessSlotSyncInterrupts(WalReceiverConn **wrconn)
+{
+ CHECK_FOR_INTERRUPTS();
+
+ if (ShutdownRequestPending)
+ {
+ ereport(LOG,
+ errmsg("replication slot sync worker is shutting"
+ " down on receiving SIGINT"));
+
+ walrcv_disconnect(*wrconn);
+ proc_exit(0);
+ }
+
+
+ if (ConfigReloadPending)
+ slotsync_reread_config(wrconn);
+}
+
+/*
+ * The main loop of our worker process.
+ */
+void
+ReplSlotSyncWorkerMain(Datum main_arg)
+{
+ WalReceiverConn *wrconn = NULL;
+ char *dbname;
+
+ /* Setup signal handling */
+ pqsignal(SIGHUP, SignalHandlerForConfigReload);
+ pqsignal(SIGINT, SignalHandlerForShutdownRequest);
+ pqsignal(SIGTERM, die);
+ BackgroundWorkerUnblockSignals();
+
+ slotsync_worker_attach();
+
+ /*
+ * If the standby has been promoted, skip the slot synchronization
+ * process.
+ *
+ * Although the startup process stops all the slot sync workers on
+ * promotion, the launcher may not have realized the promotion and could
+ * start additional workers after that. Therefore, this check is still
+ * necessary to prevent these additional workers from running.
+ */
+ if (PromoteIsTriggered())
+ exit(0);
+
+ ereport(LOG, errmsg("replication slot sync worker started"));
+
+ /* Load the libpq-specific functions */
+ load_file("libpqwalreceiver", false);
+
+ /*
+ * Get the user provided dbname from the connection string, if dbname not
+ * provided, skip sync.
+ */
+ dbname = walrcv_get_dbname_from_conninfo(PrimaryConnInfo);
+ if (dbname == NULL)
+ proc_exit(0);
+
+ /*
+ * Connect to the database specified by user in PrimaryConnInfo. We need a
+ * database connection for walrcv_exec to work. Please see comments atop
+ * libpqrcv_exec.
+ */
+ BackgroundWorkerInitializeConnection(dbname, NULL, 0);
+
+ /* Connect to the primary server */
+ wrconn = remote_connect();
+
+ /* Main wait loop. */
+ for (;;)
+ {
+ int rc;
+ long naptime;
+
+ ProcessSlotSyncInterrupts(&wrconn);
+
+ /* Check if got promoted */
+ if (!RecoveryInProgress())
+ {
+ /*
+ * Drop the slots for which sync is initiated but not yet
+ * completed i.e. they are still waiting for the primary server to
+ * catch up.
+ */
+ slotsync_drop_initiated_slots();
+ ereport(LOG,
+ errmsg("exiting slot sync woker on promotion of standby"));
+ proc_exit(0);
+ }
+
+ naptime = synchronize_slots(wrconn);
+
+ rc = WaitLatch(MyLatch,
+ WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
+ naptime,
+ WAIT_EVENT_REPL_SLOTSYNC_MAIN);
+
+ if (rc & WL_LATCH_SET)
+ ResetLatch(MyLatch);
+ }
+
+ /*
+ * The slot sync worker can not get here because it will only stop when it
+ * receives a SIGINT from the logical replication launcher, or when there
+ * is an error.
+ */
+ Assert(false);
+}
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index 1eaee4197b..73aacef25f 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -100,6 +100,7 @@
#include "catalog/pg_subscription_rel.h"
#include "catalog/pg_type.h"
#include "commands/copy.h"
+#include "commands/subscriptioncmds.h"
#include "miscadmin.h"
#include "nodes/makefuncs.h"
#include "parser/parse_relation.h"
@@ -246,7 +247,7 @@ wait_for_worker_state_change(char expected_state)
LWLockAcquire(LogicalRepWorkerLock, LW_SHARED);
worker = logicalrep_worker_find(MyLogicalRepWorker->subid,
InvalidOid, false);
- if (worker && worker->proc)
+ if (worker && worker->hdr.proc)
logicalrep_worker_wakeup_ptr(worker);
LWLockRelease(LogicalRepWorkerLock);
if (!worker)
@@ -535,7 +536,7 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
if (rstate->state == SUBREL_STATE_SYNCWAIT)
{
/* Signal the sync worker, as it may be waiting for us. */
- if (syncworker->proc)
+ if (syncworker->hdr.proc)
logicalrep_worker_wakeup_ptr(syncworker);
/* Now safe to release the LWLock */
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index c96a1700cd..2c059dffca 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -47,6 +47,7 @@
#include "miscadmin.h"
#include "pgstat.h"
#include "replication/slot.h"
+#include "replication/walsender.h"
#include "storage/fd.h"
#include "storage/ipc.h"
#include "storage/proc.h"
@@ -325,6 +326,7 @@ ReplicationSlotCreate(const char *name, bool db_specific,
slot->data.two_phase = two_phase;
slot->data.two_phase_at = InvalidXLogRecPtr;
slot->data.failover = failover;
+ slot->data.sync_state = SYNCSLOT_STATE_NONE;
/* and then data only present in shared memory */
slot->just_dirtied = false;
@@ -684,12 +686,26 @@ restart:
* Permanently drop replication slot identified by the passed in name.
*/
void
-ReplicationSlotDrop(const char *name, bool nowait)
+ReplicationSlotDrop(const char *name, bool nowait, bool user_cmd)
{
Assert(MyReplicationSlot == NULL);
ReplicationSlotAcquire(name, nowait);
+ /*
+ * Do not allow users to drop the slots which are currently being synced
+ * from the primary to the standby.
+ */
+ if (user_cmd && RecoveryInProgress() &&
+ MyReplicationSlot->data.sync_state != SYNCSLOT_STATE_NONE)
+ {
+ ReplicationSlotRelease();
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("cannot drop replication slot \"%s\"", name),
+ errdetail("This slot is being synced from the primary.")));
+ }
+
ReplicationSlotDropAcquired();
}
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index fb6e37d2c3..7b1e0c1552 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -226,11 +226,38 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
CheckSlotRequirements();
- ReplicationSlotDrop(NameStr(*name), true);
+ ReplicationSlotDrop(NameStr(*name), true, true);
PG_RETURN_VOID();
}
+/*
+ * SQL function for getting invalidation cause of a slot.
+ *
+ * Returns ReplicationSlotInvalidationCause enum value for valid slot_name;
+ * returns NULL if slot with given name is not found.
+ *
+ * Returns RS_INVAL_NONE if the given slot is not invalidated.
+ */
+Datum
+pg_get_slot_invalidation_cause(PG_FUNCTION_ARGS)
+{
+ Name name = PG_GETARG_NAME(0);
+ ReplicationSlot *s;
+ ReplicationSlotInvalidationCause cause;
+
+ s = SearchNamedReplicationSlot(NameStr(*name), true);
+
+ if (s == NULL)
+ PG_RETURN_NULL();
+
+ SpinLockAcquire(&s->mutex);
+ cause = s->data.invalidated;
+ SpinLockRelease(&s->mutex);
+
+ PG_RETURN_INT16(cause);
+}
+
/*
* pg_get_replication_slots - SQL SRF showing all replication slots
* that currently exist on the database cluster.
@@ -238,7 +265,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
Datum
pg_get_replication_slots(PG_FUNCTION_ARGS)
{
-#define PG_GET_REPLICATION_SLOTS_COLS 16
+#define PG_GET_REPLICATION_SLOTS_COLS 17
ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
XLogRecPtr currlsn;
int slotno;
@@ -420,6 +447,8 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
values[i++] = BoolGetDatum(slot_contents.data.failover);
+ values[i++] = CharGetDatum(slot_contents.data.sync_state);
+
Assert(i == PG_GET_REPLICATION_SLOTS_COLS);
tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 3e44228bde..edb3656cd2 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -1254,7 +1254,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
static void
DropReplicationSlot(DropReplicationSlotCmd *cmd)
{
- ReplicationSlotDrop(cmd->slotname, !cmd->wait);
+ ReplicationSlotDrop(cmd->slotname, !cmd->wait, false);
}
/*
diff --git a/src/backend/storage/lmgr/lwlocknames.txt b/src/backend/storage/lmgr/lwlocknames.txt
index f72f2906ce..e62a3f1bc0 100644
--- a/src/backend/storage/lmgr/lwlocknames.txt
+++ b/src/backend/storage/lmgr/lwlocknames.txt
@@ -54,3 +54,4 @@ XactTruncationLock 44
WrapLimitsVacuumLock 46
NotifyQueueTailLock 47
WaitEventExtensionLock 48
+SlotSyncWorkerLock 49
diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt
index ede94a1ede..7eb735824c 100644
--- a/src/backend/utils/activity/wait_event_names.txt
+++ b/src/backend/utils/activity/wait_event_names.txt
@@ -53,6 +53,8 @@ LOGICAL_APPLY_MAIN "Waiting in main loop of logical replication apply process."
LOGICAL_LAUNCHER_MAIN "Waiting in main loop of logical replication launcher process."
LOGICAL_PARALLEL_APPLY_MAIN "Waiting in main loop of logical replication parallel apply process."
RECOVERY_WAL_STREAM "Waiting in main loop of startup process for WAL to arrive, during streaming recovery."
+REPL_SLOTSYNC_MAIN "Waiting in main loop of slot sync worker."
+REPL_SLOTSYNC_PRIMARY_CATCHUP "Waiting for the primary to catch-up, in slot sync worker."
SYSLOGGER_MAIN "Waiting in main loop of syslogger process."
WAL_RECEIVER_MAIN "Waiting in main loop of WAL receiver process."
WAL_SENDER_MAIN "Waiting in main loop of WAL sender process."
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index e5e7bb23f9..65a1e544c1 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -65,8 +65,11 @@
#include "postmaster/syslogger.h"
#include "postmaster/walwriter.h"
#include "replication/logicallauncher.h"
+#include "replication/reorderbuffer.h"
#include "replication/slot.h"
#include "replication/syncrep.h"
+#include "replication/walreceiver.h"
+#include "replication/walsender.h"
#include "storage/bufmgr.h"
#include "storage/large_object.h"
#include "storage/pg_shmem.h"
@@ -2021,6 +2024,15 @@ struct config_bool ConfigureNamesBool[] =
NULL, NULL, NULL
},
+ {
+ {"enable_syncslot", PGC_SIGHUP, REPLICATION_STANDBY,
+ gettext_noop("Enables a physical standby to synchronize logical failover slots from the primary server."),
+ },
+ &enable_syncslot,
+ false,
+ NULL, NULL, NULL
+ },
+
/* End-of-list marker */
{
{NULL, 0, 0, NULL, NULL}, NULL, false, NULL, NULL, NULL
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index 998080e0e4..eaafdb3035 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -355,6 +355,7 @@
#wal_retrieve_retry_interval = 5s # time to wait before retrying to
# retrieve WAL after a failed attempt
#recovery_min_apply_delay = 0 # minimum delay for applying changes during recovery
+#enable_syncslot = on # enables slot synchronization on the physical standby from the primary
# - Subscribers -
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index d906734750..6a8192ad83 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11095,14 +11095,18 @@
proname => 'pg_drop_replication_slot', provolatile => 'v', proparallel => 'u',
prorettype => 'void', proargtypes => 'name',
prosrc => 'pg_drop_replication_slot' },
+{ oid => '8484', descr => 'what caused the replication slot to become invalid',
+ proname => 'pg_get_slot_invalidation_cause', provolatile => 's', proisstrict => 't',
+ prorettype => 'int2', proargtypes => 'name',
+ prosrc => 'pg_get_slot_invalidation_cause' },
{ oid => '3781',
descr => 'information about replication slots currently in use',
proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f',
proretset => 't', provolatile => 's', prorettype => 'record',
proargtypes => '',
- proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,bool}',
- proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
- proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,failover}',
+ proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,bool,char}',
+ proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+ proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,failover,sync_state}',
prosrc => 'pg_get_replication_slots' },
{ oid => '3786', descr => 'set up a logical replication slot',
proname => 'pg_create_logical_replication_slot', provolatile => 'v',
diff --git a/src/include/commands/subscriptioncmds.h b/src/include/commands/subscriptioncmds.h
index 214dc6c29e..75b4b2040d 100644
--- a/src/include/commands/subscriptioncmds.h
+++ b/src/include/commands/subscriptioncmds.h
@@ -17,6 +17,7 @@
#include "catalog/objectaddress.h"
#include "parser/parse_node.h"
+#include "replication/walreceiver.h"
extern ObjectAddress CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
bool isTopLevel);
@@ -28,4 +29,7 @@ extern void AlterSubscriptionOwner_oid(Oid subid, Oid newOwnerId);
extern char defGetStreamingMode(DefElem *def);
+extern void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn,
+ char *slotname, bool missing_ok);
+
#endif /* SUBSCRIPTIONCMDS_H */
diff --git a/src/include/nodes/replnodes.h b/src/include/nodes/replnodes.h
index bef8a7162e..085d729620 100644
--- a/src/include/nodes/replnodes.h
+++ b/src/include/nodes/replnodes.h
@@ -33,7 +33,6 @@ typedef struct IdentifySystemCmd
NodeTag type;
} IdentifySystemCmd;
-
/* ----------------------
* BASE_BACKUP command
* ----------------------
diff --git a/src/include/replication/logicallauncher.h b/src/include/replication/logicallauncher.h
index a07c9cb311..02499a7e66 100644
--- a/src/include/replication/logicallauncher.h
+++ b/src/include/replication/logicallauncher.h
@@ -15,9 +15,11 @@
extern PGDLLIMPORT int max_logical_replication_workers;
extern PGDLLIMPORT int max_sync_workers_per_subscription;
extern PGDLLIMPORT int max_parallel_apply_workers_per_subscription;
+extern PGDLLIMPORT bool enable_syncslot;
+
extern void ApplyLauncherRegister(void);
-extern void ApplyLauncherMain(Datum main_arg);
+extern void LauncherMain(Datum main_arg);
extern Size ApplyLauncherShmemSize(void);
extern void ApplyLauncherShmemInit(void);
@@ -31,4 +33,9 @@ extern bool IsLogicalLauncher(void);
extern pid_t GetLeaderApplyWorkerPid(pid_t pid);
+extern void ShutDownSlotSync(void);
+
+extern PGDLLIMPORT char *PrimaryConnInfo;
+extern PGDLLIMPORT char *PrimarySlotName;
+
#endif /* LOGICALLAUNCHER_H */
diff --git a/src/include/replication/logicalworker.h b/src/include/replication/logicalworker.h
index bbd71d0b42..baad5a8f3a 100644
--- a/src/include/replication/logicalworker.h
+++ b/src/include/replication/logicalworker.h
@@ -19,6 +19,7 @@ extern PGDLLIMPORT volatile sig_atomic_t ParallelApplyMessagePending;
extern void ApplyWorkerMain(Datum main_arg);
extern void ParallelApplyWorkerMain(Datum main_arg);
extern void TablesyncWorkerMain(Datum main_arg);
+extern void ReplSlotSyncWorkerMain(Datum main_arg);
extern bool IsLogicalWorker(void);
extern bool IsLogicalParallelApplyWorker(void);
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index ca06e5b1ad..6c300da45a 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -15,7 +15,6 @@
#include "storage/lwlock.h"
#include "storage/shmem.h"
#include "storage/spin.h"
-#include "replication/walreceiver.h"
/*
* Behaviour of replication slots, upon release or crash.
@@ -52,6 +51,14 @@ typedef enum ReplicationSlotInvalidationCause
RS_INVAL_WAL_LEVEL,
} ReplicationSlotInvalidationCause;
+/* The possible values for 'sync_state' in ReplicationSlotPersistentData */
+#define SYNCSLOT_STATE_NONE 'n' /* None for user created slots */
+#define SYNCSLOT_STATE_INITIATED 'i' /* Sync initiated for the slot but
+ * not completed yet, waiting for
+ * the primary server to catch-up */
+#define SYNCSLOT_STATE_READY 'r' /* Initialization complete, ready
+ * to be synced further */
+
/*
* On-Disk data of a replication slot, preserved across restarts.
*/
@@ -112,6 +119,13 @@ typedef struct ReplicationSlotPersistentData
/* plugin name */
NameData plugin;
+ /*
+ * Is this a slot created by a sync-slot worker?
+ *
+ * Relevant for logical slots on the physical standby.
+ */
+ char sync_state;
+
/*
* Is this a failover slot (sync candidate for physical standbys)?
* Only relevant for logical slots on the primary server.
@@ -227,7 +241,7 @@ extern void ReplicationSlotCreate(const char *name, bool db_specific,
ReplicationSlotPersistency persistency,
bool two_phase, bool failover);
extern void ReplicationSlotPersist(void);
-extern void ReplicationSlotDrop(const char *name, bool nowait);
+extern void ReplicationSlotDrop(const char *name, bool nowait, bool user_cmd);
extern void ReplicationSlotAlter(const char *name, bool failover);
extern void ReplicationSlotAcquire(const char *name, bool nowait);
@@ -253,7 +267,6 @@ extern ReplicationSlot *SearchNamedReplicationSlot(const char *name, bool need_l
extern int ReplicationSlotIndex(ReplicationSlot *slot);
extern bool ReplicationSlotName(int index, Name name);
extern void ReplicationSlotNameForTablesync(Oid suboid, Oid relid, char *syncslotname, Size szslot);
-extern void ReplicationSlotDropAtPubNode(WalReceiverConn *wrconn, char *slotname, bool missing_ok);
extern void StartupReplicationSlots(void);
extern void CheckPointReplicationSlots(bool is_shutdown);
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 61bc8de72c..2f7ae9dc69 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -20,6 +20,7 @@
#include "pgtime.h"
#include "port/atomics.h"
#include "replication/logicalproto.h"
+#include "replication/slot.h"
#include "replication/walsender.h"
#include "storage/condition_variable.h"
#include "storage/latch.h"
@@ -280,6 +281,21 @@ typedef void (*walrcv_get_senderinfo_fn) (WalReceiverConn *conn,
typedef char *(*walrcv_identify_system_fn) (WalReceiverConn *conn,
TimeLineID *primary_tli);
+/*
+ * walrcv_get_dbinfo_for_failover_slots_fn
+ *
+ * Run LIST_DBID_FOR_FAILOVER_SLOTS on primary server to get the
+ * list of unique DBIDs for failover logical slots
+ */
+typedef List *(*walrcv_get_dbinfo_for_failover_slots_fn) (WalReceiverConn *conn);
+
+/*
+ * walrcv_get_dbname_from_conninfo_fn
+ *
+ * Returns the dbid from the primary_conninfo
+ */
+typedef char *(*walrcv_get_dbname_from_conninfo_fn) (const char *conninfo);
+
/*
* walrcv_server_version_fn
*
@@ -404,6 +420,7 @@ typedef struct WalReceiverFunctionsType
walrcv_get_conninfo_fn walrcv_get_conninfo;
walrcv_get_senderinfo_fn walrcv_get_senderinfo;
walrcv_identify_system_fn walrcv_identify_system;
+ walrcv_get_dbname_from_conninfo_fn walrcv_get_dbname_from_conninfo;
walrcv_server_version_fn walrcv_server_version;
walrcv_readtimelinehistoryfile_fn walrcv_readtimelinehistoryfile;
walrcv_startstreaming_fn walrcv_startstreaming;
@@ -429,6 +446,8 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
WalReceiverFunctions->walrcv_get_senderinfo(conn, sender_host, sender_port)
#define walrcv_identify_system(conn, primary_tli) \
WalReceiverFunctions->walrcv_identify_system(conn, primary_tli)
+#define walrcv_get_dbname_from_conninfo(conninfo) \
+ WalReceiverFunctions->walrcv_get_dbname_from_conninfo(conninfo)
#define walrcv_server_version(conn) \
WalReceiverFunctions->walrcv_server_version(conn)
#define walrcv_readtimelinehistoryfile(conn, tli, filename, content, size) \
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 4378690ab0..435afb1ee1 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -1,7 +1,8 @@
/*-------------------------------------------------------------------------
*
* worker_internal.h
- * Internal headers shared by logical replication workers.
+ * Internal headers shared by logical replication workers
+ * and slotsync workers.
*
* Portions Copyright (c) 2016-2023, PostgreSQL Global Development Group
*
@@ -36,14 +37,9 @@ typedef enum LogicalRepWorkerType
WORKERTYPE_PARALLEL_APPLY,
} LogicalRepWorkerType;
-typedef struct LogicalRepWorker
+/* Common data for Slotsync and LogicalRep workers */
+typedef struct LogicalWorkerHeader
{
- /* What type of worker is this? */
- LogicalRepWorkerType type;
-
- /* Time at which this worker was launched. */
- TimestampTz launch_time;
-
/* Indicates if this slot is used or free. */
bool in_use;
@@ -53,6 +49,19 @@ typedef struct LogicalRepWorker
/* Pointer to proc array. NULL if not running. */
PGPROC *proc;
+} LogicalWorkerHeader;
+
+/* Shared memory structure for logical replication workers. */
+typedef struct LogicalRepWorker
+{
+ LogicalWorkerHeader hdr;
+
+ /* Time at which this worker was launched. */
+ TimestampTz launch_time;
+
+ /* What type of worker is this? */
+ LogicalRepWorkerType type;
+
/* Database id to connect to. */
Oid dbid;
@@ -96,6 +105,32 @@ typedef struct LogicalRepWorker
TimestampTz reply_time;
} LogicalRepWorker;
+/*
+ * Shared memory structure for Slot-Sync worker. It is allocated by logical
+ * replication launcher and then read by each slot sync worker.
+ *
+ * It is protected by LWLock (SlotSyncWorkerLock). Each slot sync worker
+ * reading the structure needs to hold the lock in shared mode, whereas
+ * the logical replication launcher which updates it needs to hold the lock
+ * in exclusive mode.
+ */
+typedef struct SlotSyncWorkerInfo
+{
+ /*
+ * The header used by slot-sync worker and logical rep worker.
+ *
+ * The header has 'generation'. For slot sync worker, it is not needed
+ * but since slot sync worker uses some of the code infrastructure
+ * provided by logical rep worker which needs 'generation', it is
+ * thus retained for slot sync worker but is always kept as 0.
+ */
+ LogicalWorkerHeader hdr;
+
+ /* The last sync-cycle time when the worker updated any of the slots. */
+ TimestampTz last_update_time;
+
+} SlotSyncWorkerInfo;
+
/*
* State of the transaction in parallel apply worker.
*
@@ -234,12 +269,16 @@ extern PGDLLIMPORT struct WalReceiverConn *LogRepWorkerWalRcvConn;
/* Worker and subscription objects. */
extern PGDLLIMPORT Subscription *MySubscription;
extern PGDLLIMPORT LogicalRepWorker *MyLogicalRepWorker;
+extern PGDLLIMPORT SlotSyncWorkerInfo *SlotSyncWorker;
extern PGDLLIMPORT bool in_remote_transaction;
extern PGDLLIMPORT bool InitializingApplyWorker;
extern void logicalrep_worker_attach(int slot);
+extern void slotsync_worker_attach(void);
+extern void slotsync_worker_detach(int code, Datum arg);
+extern void slotsync_drop_initiated_slots(void);
extern LogicalRepWorker *logicalrep_worker_find(Oid subid, Oid relid,
bool only_running);
extern List *logicalrep_workers_find(Oid subid, bool only_running);
@@ -328,9 +367,9 @@ extern void pa_decr_and_wait_stream_block(void);
extern void pa_xact_finish(ParallelApplyWorkerInfo *winfo,
XLogRecPtr remote_lsn);
-#define isParallelApplyWorker(worker) ((worker)->in_use && \
+#define isParallelApplyWorker(worker) ((worker)->hdr.in_use && \
(worker)->type == WORKERTYPE_PARALLEL_APPLY)
-#define isTablesyncWorker(worker) ((worker)->in_use && \
+#define isTablesyncWorker(worker) ((worker)->hdr.in_use && \
(worker)->type == WORKERTYPE_TABLESYNC)
static inline bool
@@ -342,14 +381,14 @@ am_tablesync_worker(void)
static inline bool
am_leader_apply_worker(void)
{
- Assert(MyLogicalRepWorker->in_use);
+ Assert(MyLogicalRepWorker->hdr.in_use);
return (MyLogicalRepWorker->type == WORKERTYPE_APPLY);
}
static inline bool
am_parallel_apply_worker(void)
{
- Assert(MyLogicalRepWorker->in_use);
+ Assert(MyLogicalRepWorker->hdr.in_use);
return isParallelApplyWorker(MyLogicalRepWorker);
}
diff --git a/src/test/recovery/t/050_verify_slot_order.pl b/src/test/recovery/t/050_verify_slot_order.pl
index bff0f52a46..e7a00bde12 100644
--- a/src/test/recovery/t/050_verify_slot_order.pl
+++ b/src/test/recovery/t/050_verify_slot_order.pl
@@ -146,4 +146,131 @@ $result = $subscriber1->safe_psql('postgres',
"SELECT count(*) = $primary_row_count FROM tab_int;");
is($result, 't', "subscriber1 gets data from primary after standby1 acknowledges changes");
+# Test logical failover slots on the standby
+# Configure standby3 to replicate and synchronize logical slots configured
+# for failover on the primary
+#
+# failover slot lsub1_slot->| ----> subscriber1 (connected via logical replication)
+# primary ---> |
+# physical slot sb3_slot--->| ----> standby3 (connected via streaming replication)
+# | lsub1_slot(synced_slot)
+
+# Cleanup old standby_slot_names
+$primary->stop;
+$primary->append_conf(
+ 'postgresql.conf', qq(
+standby_slot_names = ''
+));
+$primary->start;
+
+$primary->psql('postgres',
+ q{SELECT pg_create_physical_replication_slot('sb3_slot');});
+
+$backup_name = 'backup2';
+$primary->backup($backup_name);
+
+# Create standby3
+my $standby3 = PostgreSQL::Test::Cluster->new('standby3');
+$standby3->init_from_backup(
+ $primary, $backup_name,
+ has_streaming => 1,
+ has_restoring => 1);
+
+my $connstr_1 = $primary->connstr;
+$standby3->stop;
+$standby3->append_conf(
+ 'postgresql.conf', q{
+enable_syncslot = true
+hot_standby_feedback = on
+primary_slot_name = 'sb3_slot'
+});
+$standby3->append_conf(
+ 'postgresql.conf', qq(
+primary_conninfo = '$connstr_1 dbname=postgres'
+));
+$standby3->start;
+
+# Add this standby into the primary's configuration
+$primary->stop;
+$primary->append_conf(
+ 'postgresql.conf', qq(
+standby_slot_names = 'sb3_slot'
+));
+$primary->start;
+
+# Restart the standby
+$standby3->restart;
+
+# Wait for the standby to start sync
+my $offset = -s $standby3->logfile;
+$standby3->wait_for_log(
+ qr/LOG: ( [A-Z0-9]+:)? waiting for remote slot \"lsub1_slot\"/,
+ $offset);
+
+# Advance lsn on the primary
+$primary->safe_psql('postgres',
+ "SELECT pg_log_standby_snapshot();");
+$primary->safe_psql('postgres',
+ "SELECT pg_log_standby_snapshot();");
+$primary->safe_psql('postgres',
+ "SELECT pg_log_standby_snapshot();");
+
+# Wait for the standby to finish sync
+$offset = -s $standby3->logfile;
+$standby3->wait_for_log(
+ qr/LOG: ( [A-Z0-9]+:)? wait over for remote slot \"lsub1_slot\"/,
+ $offset);
+
+# Confirm that logical failover slot is created on the standby
+is( $standby3->safe_psql('postgres',
+ q{SELECT slot_name FROM pg_replication_slots;}
+ ),
+ 'lsub1_slot',
+ 'failover slot was created');
+
+# Verify slot properties on the standby
+is( $standby3->safe_psql('postgres',
+ q{SELECT failover, sync_state FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';}
+ ),
+ "t|r",
+ 'logical slot has sync_state as ready and failover as true on standby');
+
+# Verify slot properties on the primary
+is( $primary->safe_psql('postgres',
+ q{SELECT failover, sync_state FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';}
+ ),
+ "t|n",
+ 'logical slot has sync_state as none and failover as true on primary');
+
+# Test to confirm that restart_lsn of the logical slot on the primary is synced to the standby
+
+# Truncate table on primary
+$primary->safe_psql('postgres',
+ "TRUNCATE TABLE tab_int;");
+
+# Insert data on the primary
+$primary->safe_psql('postgres',
+ "INSERT INTO tab_int SELECT generate_series(1, $primary_row_count);");
+
+# let the slots get synced on the standby
+sleep 2;
+
+# Get the restart_lsn for the logical slot lsub1_slot on the primary
+my $primary_lsn = $primary->safe_psql('postgres',
+ "SELECT restart_lsn from pg_replication_slots WHERE slot_name = 'lsub1_slot';");
+
+# Confirm that restart_lsn of lsub1_slot slot is synced to the standby
+$result = $standby3->safe_psql('postgres',
+ qq[SELECT '$primary_lsn' <= restart_lsn from pg_replication_slots WHERE slot_name = 'lsub1_slot';]);
+is($result, 't', 'restart_lsn of slot lsub1_slot synced to standby');
+
+# Get the confirmed_flush_lsn for the logical slot lsub1_slot on the primary
+$primary_lsn = $primary->safe_psql('postgres',
+ "SELECT confirmed_flush_lsn from pg_replication_slots WHERE slot_name = 'lsub1_slot';");
+
+# Confirm that confirmed_flush_lsn of lsub1_slot slot is synced to the standby
+$result = $standby3->safe_psql('postgres',
+ qq[SELECT '$primary_lsn' <= confirmed_flush_lsn from pg_replication_slots WHERE slot_name = 'lsub1_slot';]);
+is($result, 't', 'confirmed_flush_lsn of slot lsub1_slot synced to the standby');
+
done_testing();
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index cb3b04aa0c..f2e5a3849c 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1474,8 +1474,9 @@ pg_replication_slots| SELECT l.slot_name,
l.safe_wal_size,
l.two_phase,
l.conflicting,
- l.failover
- FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, failover)
+ l.failover,
+ l.sync_state
+ FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, failover, sync_state)
LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
pg_roles| SELECT pg_authid.rolname,
pg_authid.rolsuper,
diff --git a/src/test/regress/expected/sysviews.out b/src/test/regress/expected/sysviews.out
index 271313ebf8..aac83755de 100644
--- a/src/test/regress/expected/sysviews.out
+++ b/src/test/regress/expected/sysviews.out
@@ -132,8 +132,9 @@ select name, setting from pg_settings where name like 'enable%';
enable_self_join_removal | on
enable_seqscan | on
enable_sort | on
+ enable_syncslot | off
enable_tidscan | on
-(22 rows)
+(23 rows)
-- There are always wait event descriptions for various types.
select type, count(*) > 0 as ok FROM pg_wait_events
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index d80d30e99c..d37eeb8a47 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -1432,6 +1432,7 @@ LimitState
LimitStateCond
List
ListCell
+ListDBForLogicalSlotsCmd
ListDictionary
ListParsedLex
ListenAction
@@ -1511,6 +1512,7 @@ LogicalSlotInfo
LogicalSlotInfoArr
LogicalTape
LogicalTapeSet
+LogicalWorkerHeader
LsnReadQueue
LsnReadQueueNextFun
LsnReadQueueNextStatus
@@ -2314,6 +2316,7 @@ RelocationBufferInfo
RelptrFreePageBtree
RelptrFreePageManager
RelptrFreePageSpanLeader
+RemoteSlot
RenameStmt
ReopenPtrType
ReorderBuffer
@@ -2572,6 +2575,8 @@ SlabBlock
SlabContext
SlabSlot
SlotNumber
+SlotSyncWorker
+SlotSyncWorkerInfo
SlruCtl
SlruCtlData
SlruErrorCause
--
2.30.0.windows.2
[application/octet-stream] v39_2-0001-Allow-logical-walsenders-to-wait-for-the-physica.patch (134.3K, ../OS0PR01MB57161938A348977127D56A6694BDA@OS0PR01MB5716.jpnprd01.prod.outlook.com/4-v39_2-0001-Allow-logical-walsenders-to-wait-for-the-physica.patch)
download | inline diff:
From 24dba0abff51672a3505e27adf36b395d7412f4d Mon Sep 17 00:00:00 2001
From: Shveta Malik <[email protected]>
Date: Tue, 21 Nov 2023 14:50:21 +0530
Subject: [PATCH v39 1/3] Allow logical walsenders to wait for the physical
standbys
A new property 'failover' is added at the slot level. This is persistent
information to indicate that this logical slot is enabled to be synced to the
physical standbys so that logical replication can be resumed after failover. It
is always false for physical slots.
Users can set this flag during CREATE SUBSCRIPTION or during
pg_create_logical_replication_slot API. Examples:
CREATE SUBSCRIPTION mysub CONNECTION '..' PUBLICATION mypub
WITH (failover = true);
(failover is the last arg)
SELECT * FROM pg_create_logical_replication_slot('myslot',
'pgoutput', false, true, true);
Altering the failover option of the subscription is currently not
permitted. However, this restriction may be lifted in future versions.
The value of the 'failover' flag is displayed as part of pg_replication_slots
view.
A new GUC standby_slot_names has been added. It is the list of
physical replication slots that logical replication with failover
enabled waits for. The intent of this wait is that no logical
replication subscriptions (with failover=true) should get
ahead of physical replication standbys (corresponding to the
physical slots in standby_slot_names).
A new walreceiver API walrcv_alter_slot has been introduced to
enable the failover of the slot on publisher node.
---
contrib/test_decoding/expected/slot.out | 58 ++++
contrib/test_decoding/sql/slot.sql | 13 +
doc/src/sgml/catalogs.sgml | 12 +
doc/src/sgml/config.sgml | 16 +
doc/src/sgml/func.sgml | 11 +-
doc/src/sgml/protocol.sgml | 51 +++
doc/src/sgml/ref/alter_subscription.sgml | 9 +-
doc/src/sgml/ref/create_subscription.sgml | 23 ++
doc/src/sgml/system-views.sgml | 11 +
src/backend/catalog/pg_subscription.c | 1 +
src/backend/catalog/system_functions.sql | 1 +
src/backend/catalog/system_views.sql | 6 +-
src/backend/commands/subscriptioncmds.c | 91 ++++-
.../libpqwalreceiver/libpqwalreceiver.c | 44 ++-
.../replication/logical/logicalfuncs.c | 13 +
src/backend/replication/logical/tablesync.c | 51 ++-
src/backend/replication/logical/worker.c | 72 +++-
src/backend/replication/repl_gram.y | 18 +-
src/backend/replication/repl_scanner.l | 2 +
src/backend/replication/slot.c | 182 +++++++++-
src/backend/replication/slotfuncs.c | 19 +-
src/backend/replication/walreceiver.c | 2 +-
src/backend/replication/walsender.c | 327 +++++++++++++++++-
.../utils/activity/wait_event_names.txt | 1 +
src/backend/utils/misc/guc_tables.c | 14 +
src/backend/utils/misc/postgresql.conf.sample | 2 +
src/bin/pg_upgrade/info.c | 6 +-
src/bin/pg_upgrade/pg_upgrade.c | 6 +-
src/bin/pg_upgrade/pg_upgrade.h | 2 +
src/bin/pg_upgrade/t/003_logical_slots.pl | 6 +-
src/bin/psql/describe.c | 8 +-
src/bin/psql/tab-complete.c | 2 +-
src/include/catalog/pg_proc.dat | 14 +-
src/include/catalog/pg_subscription.h | 11 +
src/include/nodes/replnodes.h | 12 +
src/include/replication/slot.h | 12 +-
src/include/replication/walreceiver.h | 18 +-
src/include/replication/walsender.h | 4 +
src/include/replication/walsender_private.h | 7 +
src/include/replication/worker_internal.h | 3 +-
src/include/utils/guc_hooks.h | 3 +
src/test/recovery/meson.build | 1 +
src/test/recovery/t/006_logical_decoding.pl | 3 +-
src/test/recovery/t/050_verify_slot_order.pl | 149 ++++++++
src/test/regress/expected/rules.out | 5 +-
src/test/regress/expected/subscription.out | 165 +++++----
src/test/regress/sql/subscription.sql | 8 +
src/tools/pgindent/typedefs.list | 2 +
48 files changed, 1334 insertions(+), 163 deletions(-)
create mode 100644 src/test/recovery/t/050_verify_slot_order.pl
diff --git a/contrib/test_decoding/expected/slot.out b/contrib/test_decoding/expected/slot.out
index 63a9940f73..261d8886d3 100644
--- a/contrib/test_decoding/expected/slot.out
+++ b/contrib/test_decoding/expected/slot.out
@@ -406,3 +406,61 @@ SELECT pg_drop_replication_slot('copied_slot2_notemp');
(1 row)
+-- Test failover option of slots.
+SELECT 'init' FROM pg_create_logical_replication_slot('failover_true_slot', 'test_decoding', false, false, true);
+ ?column?
+----------
+ init
+(1 row)
+
+SELECT 'init' FROM pg_create_logical_replication_slot('failover_false_slot', 'test_decoding', false, false, false);
+ ?column?
+----------
+ init
+(1 row)
+
+SELECT 'init' FROM pg_create_logical_replication_slot('failover_default_slot', 'test_decoding', false, false);
+ ?column?
+----------
+ init
+(1 row)
+
+SELECT 'init' FROM pg_create_physical_replication_slot('physical_slot');
+ ?column?
+----------
+ init
+(1 row)
+
+SELECT slot_name, slot_type, failover FROM pg_replication_slots;
+ slot_name | slot_type | failover
+-----------------------+-----------+----------
+ failover_true_slot | logical | t
+ failover_false_slot | logical | f
+ failover_default_slot | logical | f
+ physical_slot | physical | f
+(4 rows)
+
+SELECT pg_drop_replication_slot('failover_true_slot');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
+SELECT pg_drop_replication_slot('failover_false_slot');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
+SELECT pg_drop_replication_slot('failover_default_slot');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
+SELECT pg_drop_replication_slot('physical_slot');
+ pg_drop_replication_slot
+--------------------------
+
+(1 row)
+
diff --git a/contrib/test_decoding/sql/slot.sql b/contrib/test_decoding/sql/slot.sql
index 1aa27c5667..45aeae7fd5 100644
--- a/contrib/test_decoding/sql/slot.sql
+++ b/contrib/test_decoding/sql/slot.sql
@@ -176,3 +176,16 @@ ORDER BY o.slot_name, c.slot_name;
SELECT pg_drop_replication_slot('orig_slot2');
SELECT pg_drop_replication_slot('copied_slot2_no_change');
SELECT pg_drop_replication_slot('copied_slot2_notemp');
+
+-- Test failover option of slots.
+SELECT 'init' FROM pg_create_logical_replication_slot('failover_true_slot', 'test_decoding', false, false, true);
+SELECT 'init' FROM pg_create_logical_replication_slot('failover_false_slot', 'test_decoding', false, false, false);
+SELECT 'init' FROM pg_create_logical_replication_slot('failover_default_slot', 'test_decoding', false, false);
+SELECT 'init' FROM pg_create_physical_replication_slot('physical_slot');
+
+SELECT slot_name, slot_type, failover FROM pg_replication_slots;
+
+SELECT pg_drop_replication_slot('failover_true_slot');
+SELECT pg_drop_replication_slot('failover_false_slot');
+SELECT pg_drop_replication_slot('failover_default_slot');
+SELECT pg_drop_replication_slot('physical_slot');
diff --git a/doc/src/sgml/catalogs.sgml b/doc/src/sgml/catalogs.sgml
index 3ec7391ec5..e666730c64 100644
--- a/doc/src/sgml/catalogs.sgml
+++ b/doc/src/sgml/catalogs.sgml
@@ -7990,6 +7990,18 @@ SCRAM-SHA-256$<replaceable><iteration count></replaceable>:<replaceable>&l
</para></entry>
</row>
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>subfailoverstate</structfield> <type>char</type>
+ </para>
+ <para>
+ State codes for failover mode:
+ <literal>d</literal> = disabled,
+ <literal>p</literal> = pending enablement,
+ <literal>e</literal> = enabled
+ </para></entry>
+ </row>
+
<row>
<entry role="catalog_table_entry"><para role="column_definition">
<structfield>subconninfo</structfield> <type>text</type>
diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml
index 93735e3aea..6440378d5c 100644
--- a/doc/src/sgml/config.sgml
+++ b/doc/src/sgml/config.sgml
@@ -4344,6 +4344,22 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"' # Windows
</listitem>
</varlistentry>
+ <varlistentry id="guc-standby-slot-names" xreflabel="standby_slot_names">
+ <term><varname>standby_slot_names</varname> (<type>string</type>)
+ <indexterm>
+ <primary><varname>standby_slot_names</varname> configuration parameter</primary>
+ </indexterm>
+ </term>
+ <listitem>
+ <para>
+ List of physical replication slots that logical replication slots with
+ failover enabled waits for. If a logical replication connection is
+ meant to switch to a physical standby after the standby is promoted,
+ the physical replication slot for the standby should be listed here.
+ </para>
+ </listitem>
+ </varlistentry>
+
</variablelist>
</sect2>
diff --git a/doc/src/sgml/func.sgml b/doc/src/sgml/func.sgml
index 20da3ed033..90f1f19018 100644
--- a/doc/src/sgml/func.sgml
+++ b/doc/src/sgml/func.sgml
@@ -27541,7 +27541,7 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
<indexterm>
<primary>pg_create_logical_replication_slot</primary>
</indexterm>
- <function>pg_create_logical_replication_slot</function> ( <parameter>slot_name</parameter> <type>name</type>, <parameter>plugin</parameter> <type>name</type> <optional>, <parameter>temporary</parameter> <type>boolean</type>, <parameter>twophase</parameter> <type>boolean</type> </optional> )
+ <function>pg_create_logical_replication_slot</function> ( <parameter>slot_name</parameter> <type>name</type>, <parameter>plugin</parameter> <type>name</type> <optional>, <parameter>temporary</parameter> <type>boolean</type>, <parameter>twophase</parameter> <type>boolean</type>, <parameter>failover</parameter> <type>boolean</type> </optional> )
<returnvalue>record</returnvalue>
( <parameter>slot_name</parameter> <type>name</type>,
<parameter>lsn</parameter> <type>pg_lsn</type> )
@@ -27556,8 +27556,13 @@ postgres=# SELECT '0/0'::pg_lsn + pd.segment_number * ps.setting::int + :offset
released upon any error. The optional fourth parameter,
<parameter>twophase</parameter>, when set to true, specifies
that the decoding of prepared transactions is enabled for this
- slot. A call to this function has the same effect as the replication
- protocol command <literal>CREATE_REPLICATION_SLOT ... LOGICAL</literal>.
+ slot. The optional fifth parameter,
+ <parameter>failover</parameter>, when set to true,
+ specifies that this slot is enabled to be synced to the
+ physical standbys so that logical replication can be resumed
+ after failover. A call to this function has the same effect as
+ the replication protocol command
+ <literal>CREATE_REPLICATION_SLOT ... LOGICAL</literal>.
</para></entry>
</row>
diff --git a/doc/src/sgml/protocol.sgml b/doc/src/sgml/protocol.sgml
index af3f016f74..bb926ab149 100644
--- a/doc/src/sgml/protocol.sgml
+++ b/doc/src/sgml/protocol.sgml
@@ -2060,6 +2060,16 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
</para>
</listitem>
</varlistentry>
+
+ <varlistentry>
+ <term><literal>FAILOVER { 'true' | 'false' }</literal></term>
+ <listitem>
+ <para>
+ If true, the slot is enabled to be synced to the physical
+ standbys so that logical replication can be resumed after failover.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist>
<para>
@@ -2124,6 +2134,47 @@ psql "dbname=postgres replication=database" -c "IDENTIFY_SYSTEM;"
</listitem>
</varlistentry>
+ <varlistentry id="protocol-replication-alter-replication-slot" xreflabel="ALTER_REPLICATION_SLOT">
+ <term><literal>ALTER_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable> ( <replaceable class="parameter">option</replaceable> [, ...] )
+ <indexterm><primary>ALTER_REPLICATION_SLOT</primary></indexterm>
+ </term>
+ <listitem>
+ <para>
+ Change the definition of a replication slot.
+ See <xref linkend="streaming-replication-slots"/> for more about
+ replication slots. This command is currently only supported for logical
+ replication slots.
+ </para>
+
+ <variablelist>
+ <varlistentry>
+ <term><replaceable class="parameter">slot_name</replaceable></term>
+ <listitem>
+ <para>
+ The name of the slot to alter. Must be a valid replication slot
+ name (see <xref linkend="streaming-replication-slots-manipulation"/>).
+ </para>
+ </listitem>
+ </varlistentry>
+ </variablelist>
+
+ <para>The following options are supported:</para>
+
+ <variablelist>
+ <varlistentry>
+ <term><literal>FAILOVER { 'true' | 'false' }</literal></term>
+ <listitem>
+ <para>
+ If true, the slot is enabled to be synced to the physical
+ standbys so that logical replication can be resumed after failover.
+ </para>
+ </listitem>
+ </varlistentry>
+ </variablelist>
+
+ </listitem>
+ </varlistentry>
+
<varlistentry id="protocol-replication-read-replication-slot">
<term><literal>READ_REPLICATION_SLOT</literal> <replaceable class="parameter">slot_name</replaceable>
<indexterm><primary>READ_REPLICATION_SLOT</primary></indexterm>
diff --git a/doc/src/sgml/ref/alter_subscription.sgml b/doc/src/sgml/ref/alter_subscription.sgml
index 6d36ff0dc9..056404af5d 100644
--- a/doc/src/sgml/ref/alter_subscription.sgml
+++ b/doc/src/sgml/ref/alter_subscription.sgml
@@ -73,11 +73,14 @@ ALTER SUBSCRIPTION <replaceable class="parameter">name</replaceable> RENAME TO <
These commands also cannot be executed when the subscription has
<link linkend="sql-createsubscription-params-with-two-phase"><literal>two_phase</literal></link>
- commit enabled, unless
+ commit enabled or
+ <link linkend="sql-createsubscription-params-with-failover"><literal>failover</literal></link>
+ enabled, unless
<link linkend="sql-createsubscription-params-with-copy-data"><literal>copy_data</literal></link>
is <literal>false</literal>. See column <structfield>subtwophasestate</structfield>
- of <link linkend="catalog-pg-subscription"><structname>pg_subscription</structname></link>
- to know the actual two-phase state.
+ and <structfield>subfailoverstate</structfield> of
+ <link linkend="catalog-pg-subscription"><structname>pg_subscription</structname></link>
+ to know the actual state.
</para>
</refsect1>
diff --git a/doc/src/sgml/ref/create_subscription.sgml b/doc/src/sgml/ref/create_subscription.sgml
index f1c20b3a46..fa6cd1c43f 100644
--- a/doc/src/sgml/ref/create_subscription.sgml
+++ b/doc/src/sgml/ref/create_subscription.sgml
@@ -399,6 +399,29 @@ CREATE SUBSCRIPTION <replaceable class="parameter">subscription_name</replaceabl
</para>
</listitem>
</varlistentry>
+
+ <varlistentry id="sql-createsubscription-params-with-failover">
+ <term><literal>failover</literal> (<type>boolean</type>)</term>
+ <listitem>
+ <para>
+ Specifies whether the replication slot associated with the subscription
+ is enabled to be synced to the physical standbys so that logical
+ replication can be resumed from the new primary after failover.
+ The default is <literal>false</literal>.
+ </para>
+
+ <para>
+ The implementation of failover requires that replication
+ has successfully finished the initial table synchronization
+ phase. So even when <literal>failover</literal> is enabled for a
+ subscription, the internal failover state remains
+ temporarily <quote>pending</quote> until the initialization phase
+ completes. See column <structfield>subfailoverstate</structfield>
+ of <link linkend="catalog-pg-subscription"><structname>pg_subscription</structname></link>
+ to know the actual failover state.
+ </para>
+ </listitem>
+ </varlistentry>
</variablelist></para>
</listitem>
diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml
index 0ef1745631..1dc695fd3a 100644
--- a/doc/src/sgml/system-views.sgml
+++ b/doc/src/sgml/system-views.sgml
@@ -2532,6 +2532,17 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx
invalidated). Always NULL for physical slots.
</para></entry>
</row>
+
+ <row>
+ <entry role="catalog_table_entry"><para role="column_definition">
+ <structfield>failover</structfield> <type>bool</type>
+ </para>
+ <para>
+ True if this logical slot is enabled to be synced to the physical
+ standbys so that logical replication can be resumed from the new primary
+ after failover. Always false for physical slots.
+ </para></entry>
+ </row>
</tbody>
</tgroup>
</table>
diff --git a/src/backend/catalog/pg_subscription.c b/src/backend/catalog/pg_subscription.c
index d6a978f136..18512955ad 100644
--- a/src/backend/catalog/pg_subscription.c
+++ b/src/backend/catalog/pg_subscription.c
@@ -73,6 +73,7 @@ GetSubscription(Oid subid, bool missing_ok)
sub->disableonerr = subform->subdisableonerr;
sub->passwordrequired = subform->subpasswordrequired;
sub->runasowner = subform->subrunasowner;
+ sub->failoverstate = subform->subfailoverstate;
/* Get conninfo */
datum = SysCacheGetAttrNotNull(SUBSCRIPTIONOID,
diff --git a/src/backend/catalog/system_functions.sql b/src/backend/catalog/system_functions.sql
index 4206752881..4db796aa0b 100644
--- a/src/backend/catalog/system_functions.sql
+++ b/src/backend/catalog/system_functions.sql
@@ -479,6 +479,7 @@ CREATE OR REPLACE FUNCTION pg_create_logical_replication_slot(
IN slot_name name, IN plugin name,
IN temporary boolean DEFAULT false,
IN twophase boolean DEFAULT false,
+ IN failover boolean DEFAULT false,
OUT slot_name name, OUT lsn pg_lsn)
RETURNS RECORD
LANGUAGE INTERNAL
diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql
index 11d18ed9dd..63038f87f7 100644
--- a/src/backend/catalog/system_views.sql
+++ b/src/backend/catalog/system_views.sql
@@ -1023,7 +1023,8 @@ CREATE VIEW pg_replication_slots AS
L.wal_status,
L.safe_wal_size,
L.two_phase,
- L.conflicting
+ L.conflicting,
+ L.failover
FROM pg_get_replication_slots() AS L
LEFT JOIN pg_database D ON (L.datoid = D.oid);
@@ -1354,7 +1355,8 @@ REVOKE ALL ON pg_subscription FROM public;
GRANT SELECT (oid, subdbid, subskiplsn, subname, subowner, subenabled,
subbinary, substream, subtwophasestate, subdisableonerr,
subpasswordrequired, subrunasowner,
- subslotname, subsynccommit, subpublications, suborigin)
+ subslotname, subsynccommit, subpublications, suborigin,
+ subfailoverstate)
ON pg_subscription TO public;
CREATE VIEW pg_stat_subscription_stats AS
diff --git a/src/backend/commands/subscriptioncmds.c b/src/backend/commands/subscriptioncmds.c
index edc82c11be..c3187e7b9a 100644
--- a/src/backend/commands/subscriptioncmds.c
+++ b/src/backend/commands/subscriptioncmds.c
@@ -71,6 +71,7 @@
#define SUBOPT_RUN_AS_OWNER 0x00001000
#define SUBOPT_LSN 0x00002000
#define SUBOPT_ORIGIN 0x00004000
+#define SUBOPT_FAILOVER 0x00008000
/* check if the 'val' has 'bits' set */
#define IsSet(val, bits) (((val) & (bits)) == (bits))
@@ -96,6 +97,7 @@ typedef struct SubOpts
bool passwordrequired;
bool runasowner;
char *origin;
+ bool failover;
XLogRecPtr lsn;
} SubOpts;
@@ -157,6 +159,8 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
opts->runasowner = false;
if (IsSet(supported_opts, SUBOPT_ORIGIN))
opts->origin = pstrdup(LOGICALREP_ORIGIN_ANY);
+ if (IsSet(supported_opts, SUBOPT_FAILOVER))
+ opts->failover = false;
/* Parse options */
foreach(lc, stmt_options)
@@ -326,6 +330,15 @@ parse_subscription_options(ParseState *pstate, List *stmt_options,
errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("unrecognized origin value: \"%s\"", opts->origin));
}
+ else if (IsSet(supported_opts, SUBOPT_FAILOVER) &&
+ strcmp(defel->defname, "failover") == 0)
+ {
+ if (IsSet(opts->specified_opts, SUBOPT_FAILOVER))
+ errorConflictingDefElem(defel, pstate);
+
+ opts->specified_opts |= SUBOPT_FAILOVER;
+ opts->failover = defGetBoolean(defel);
+ }
else if (IsSet(supported_opts, SUBOPT_LSN) &&
strcmp(defel->defname, "lsn") == 0)
{
@@ -591,7 +604,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
SUBOPT_SYNCHRONOUS_COMMIT | SUBOPT_BINARY |
SUBOPT_STREAMING | SUBOPT_TWOPHASE_COMMIT |
SUBOPT_DISABLE_ON_ERR | SUBOPT_PASSWORD_REQUIRED |
- SUBOPT_RUN_AS_OWNER | SUBOPT_ORIGIN);
+ SUBOPT_RUN_AS_OWNER | SUBOPT_ORIGIN |
+ SUBOPT_FAILOVER);
parse_subscription_options(pstate, stmt->options, supported_opts, &opts);
/*
@@ -710,6 +724,10 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
publicationListToArray(publications);
values[Anum_pg_subscription_suborigin - 1] =
CStringGetTextDatum(opts.origin);
+ values[Anum_pg_subscription_subfailoverstate - 1] =
+ CharGetDatum(opts.failover ?
+ LOGICALREP_FAILOVER_STATE_PENDING :
+ LOGICALREP_FAILOVER_STATE_DISABLED);
tup = heap_form_tuple(RelationGetDescr(rel), values, nulls);
@@ -746,6 +764,8 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
PG_TRY();
{
+ bool failover_enabled = false;
+
check_publications(wrconn, publications);
check_publications_origin(wrconn, publications, opts.copy_data,
opts.origin, NULL, 0, stmt->subname);
@@ -776,6 +796,19 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
InvalidXLogRecPtr);
}
+ /*
+ * Even if failover is set, don't create the slot with failover
+ * enabled. Will enable it once all the tables are synced and
+ * ready. The intention is that if failover happens at the time of
+ * table-sync, user should re-launch the subscription instead of
+ * relying on main slot (if synced) with no table-sync data
+ * present. When the subscription has no tables, leave failover as
+ * false to allow ALTER SUBSCRIPTION ... REFRESH PUBLICATION to
+ * work.
+ */
+ if (opts.failover && !opts.copy_data && tables != NIL)
+ failover_enabled = true;
+
/*
* If requested, create permanent slot for the subscription. We
* won't use the initial snapshot for anything, so no need to
@@ -807,15 +840,29 @@ CreateSubscription(ParseState *pstate, CreateSubscriptionStmt *stmt,
twophase_enabled = true;
walrcv_create_slot(wrconn, opts.slot_name, false, twophase_enabled,
- CRS_NOEXPORT_SNAPSHOT, NULL);
-
- if (twophase_enabled)
- UpdateTwoPhaseState(subid, LOGICALREP_TWOPHASE_STATE_ENABLED);
+ failover_enabled, CRS_NOEXPORT_SNAPSHOT, NULL);
+ /* Update twophase and/or failover state */
+ EnableTwoPhaseFailoverTriState(subid, twophase_enabled,
+ failover_enabled);
ereport(NOTICE,
(errmsg("created replication slot \"%s\" on publisher",
opts.slot_name)));
}
+
+ /*
+ * If only the slot_name is specified (without create_slot option),
+ * it is possible that the user intends to use an existing slot on
+ * the publisher, so here we enable failover for the slot if
+ * requested.
+ */
+ else if (opts.slot_name && failover_enabled)
+ {
+ walrcv_alter_slot(wrconn, opts.slot_name, failover_enabled);
+ ereport(NOTICE,
+ (errmsg("enabled failover for replication slot \"%s\" on publisher",
+ opts.slot_name)));
+ }
}
PG_FINALLY();
{
@@ -1279,8 +1326,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
errhint("Use ALTER SUBSCRIPTION ... SET PUBLICATION ... WITH (refresh = false).")));
/*
- * See ALTER_SUBSCRIPTION_REFRESH for details why this is
- * not allowed.
+ * See ALTER_SUBSCRIPTION_REFRESH for details why copy_data
+ * is not allowed when twophase or failover is enabled.
*/
if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && opts.copy_data)
ereport(ERROR,
@@ -1288,6 +1335,12 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
errmsg("ALTER SUBSCRIPTION with refresh and copy_data is not allowed when two_phase is enabled"),
errhint("Use ALTER SUBSCRIPTION ... SET PUBLICATION with refresh = false, or with copy_data = false, or use DROP/CREATE SUBSCRIPTION.")));
+ if (sub->failoverstate == LOGICALREP_FAILOVER_STATE_ENABLED && opts.copy_data)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("ALTER SUBSCRIPTION with refresh and copy_data is not allowed when failover is enabled"),
+ errhint("Use ALTER SUBSCRIPTION ... SET PUBLICATION with refresh = false, or with copy_data = false, or use DROP/CREATE SUBSCRIPTION.")));
+
PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION with refresh");
/* Make sure refresh sees the new list of publications. */
@@ -1334,8 +1387,8 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
"ALTER SUBSCRIPTION ... DROP PUBLICATION ... WITH (refresh = false)")));
/*
- * See ALTER_SUBSCRIPTION_REFRESH for details why this is
- * not allowed.
+ * See ALTER_SUBSCRIPTION_REFRESH for details why copy_data
+ * is not allowed when twophase or failover is enabled.
*/
if (sub->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED && opts.copy_data)
ereport(ERROR,
@@ -1347,6 +1400,16 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
"ALTER SUBSCRIPTION ... ADD PUBLICATION" :
"ALTER SUBSCRIPTION ... DROP PUBLICATION")));
+ if (sub->failoverstate == LOGICALREP_FAILOVER_STATE_ENABLED && opts.copy_data)
+ ereport(ERROR,
+ (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
+ errmsg("ALTER SUBSCRIPTION with refresh and copy_data is not allowed when failover is enabled"),
+ /* translator: %s is an SQL ALTER command */
+ errhint("Use %s with refresh = false, or with copy_data = false, or use DROP/CREATE SUBSCRIPTION.",
+ isadd ?
+ "ALTER SUBSCRIPTION ... ADD PUBLICATION" :
+ "ALTER SUBSCRIPTION ... DROP PUBLICATION")));
+
PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION with refresh");
/* Refresh the new list of publications. */
@@ -1392,6 +1455,16 @@ AlterSubscription(ParseState *pstate, AlterSubscriptionStmt *stmt,
errmsg("ALTER SUBSCRIPTION ... REFRESH with copy_data is not allowed when two_phase is enabled"),
errhint("Use ALTER SUBSCRIPTION ... REFRESH with copy_data = false, or use DROP/CREATE SUBSCRIPTION.")));
+ /*
+ * See comments above for twophasestate, same holds true for
+ * 'failover'
+ */
+ if (sub->failoverstate == LOGICALREP_FAILOVER_STATE_ENABLED && opts.copy_data)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("ALTER SUBSCRIPTION ... REFRESH with copy_data is not allowed when failover is enabled"),
+ errhint("Use ALTER SUBSCRIPTION ... REFRESH with copy_data = false, or use DROP/CREATE SUBSCRIPTION.")));
+
PreventInTransactionBlock(isTopLevel, "ALTER SUBSCRIPTION ... REFRESH");
AlterSubscription_refresh(sub, opts.copy_data, NULL);
diff --git a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
index 60d5c1fc40..336c2bec99 100644
--- a/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
+++ b/src/backend/replication/libpqwalreceiver/libpqwalreceiver.c
@@ -74,8 +74,11 @@ static char *libpqrcv_create_slot(WalReceiverConn *conn,
const char *slotname,
bool temporary,
bool two_phase,
+ bool failover,
CRSSnapshotAction snapshot_action,
XLogRecPtr *lsn);
+static void libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
+ bool failover);
static pid_t libpqrcv_get_backend_pid(WalReceiverConn *conn);
static WalRcvExecResult *libpqrcv_exec(WalReceiverConn *conn,
const char *query,
@@ -96,6 +99,7 @@ static WalReceiverFunctionsType PQWalReceiverFunctions = {
.walrcv_receive = libpqrcv_receive,
.walrcv_send = libpqrcv_send,
.walrcv_create_slot = libpqrcv_create_slot,
+ .walrcv_alter_slot = libpqrcv_alter_slot,
.walrcv_get_backend_pid = libpqrcv_get_backend_pid,
.walrcv_exec = libpqrcv_exec,
.walrcv_disconnect = libpqrcv_disconnect
@@ -883,8 +887,8 @@ libpqrcv_send(WalReceiverConn *conn, const char *buffer, int nbytes)
*/
static char *
libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
- bool temporary, bool two_phase, CRSSnapshotAction snapshot_action,
- XLogRecPtr *lsn)
+ bool temporary, bool two_phase, bool failover,
+ CRSSnapshotAction snapshot_action, XLogRecPtr *lsn)
{
PGresult *res;
StringInfoData cmd;
@@ -913,7 +917,14 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
else
appendStringInfoChar(&cmd, ' ');
}
-
+ if (failover)
+ {
+ appendStringInfoString(&cmd, "FAILOVER");
+ if (use_new_options_syntax)
+ appendStringInfoString(&cmd, ", ");
+ else
+ appendStringInfoChar(&cmd, ' ');
+ }
if (use_new_options_syntax)
{
switch (snapshot_action)
@@ -982,6 +993,33 @@ libpqrcv_create_slot(WalReceiverConn *conn, const char *slotname,
return snapshot;
}
+/*
+ * Change the definition of the replication slot.
+ */
+static void
+libpqrcv_alter_slot(WalReceiverConn *conn, const char *slotname,
+ bool failover)
+{
+ StringInfoData cmd;
+ PGresult *res;
+
+ initStringInfo(&cmd);
+ appendStringInfo(&cmd, "ALTER_REPLICATION_SLOT %s ( FAILOVER %s )",
+ quote_identifier(slotname),
+ failover ? "true" : "false");
+
+ res = libpqrcv_PQexec(conn->streamConn, cmd.data);
+ pfree(cmd.data);
+
+ if (PQresultStatus(res) != PGRES_COMMAND_OK)
+ ereport(ERROR,
+ (errcode(ERRCODE_PROTOCOL_VIOLATION),
+ errmsg("could not alter replication slot \"%s\" on publisher: %s",
+ slotname, pchomp(PQerrorMessage(conn->streamConn)))));
+
+ PQclear(res);
+}
+
/*
* Return PID of remote backend process.
*/
diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c
index 1067aca08f..a36366e117 100644
--- a/src/backend/replication/logical/logicalfuncs.c
+++ b/src/backend/replication/logical/logicalfuncs.c
@@ -30,6 +30,7 @@
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/message.h"
+#include "replication/walsender.h"
#include "storage/fd.h"
#include "utils/array.h"
#include "utils/builtins.h"
@@ -109,6 +110,7 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
MemoryContext per_query_ctx;
MemoryContext oldcontext;
XLogRecPtr end_of_wal;
+ XLogRecPtr wait_for_wal_lsn;
LogicalDecodingContext *ctx;
ResourceOwner old_resowner = CurrentResourceOwner;
ArrayType *arr;
@@ -228,6 +230,17 @@ pg_logical_slot_get_changes_guts(FunctionCallInfo fcinfo, bool confirm, bool bin
NameStr(MyReplicationSlot->data.plugin),
format_procedure(fcinfo->flinfo->fn_oid))));
+ if (XLogRecPtrIsInvalid(upto_lsn))
+ wait_for_wal_lsn = end_of_wal;
+ else
+ wait_for_wal_lsn = Min(upto_lsn, end_of_wal);
+
+ /*
+ * Wait for specified streaming replication standby servers (if any)
+ * to confirm receipt of WAL up to wait_for_wal_lsn.
+ */
+ WalSndWaitForStandbyConfirmation(wait_for_wal_lsn);
+
ctx->output_writer_private = p;
/*
diff --git a/src/backend/replication/logical/tablesync.c b/src/backend/replication/logical/tablesync.c
index df3c42eb5d..1eaee4197b 100644
--- a/src/backend/replication/logical/tablesync.c
+++ b/src/backend/replication/logical/tablesync.c
@@ -614,15 +614,26 @@ process_syncing_tables_for_apply(XLogRecPtr current_lsn)
* Note: If the subscription has no tables then leave the state as
* PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to
* work.
+ *
+ * Same goes for 'failover'. Enable it only if subscription has tables
+ * and all the tablesyncs have reached READY state.
*/
- if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING)
+ if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING ||
+ MySubscription->failoverstate == LOGICALREP_FAILOVER_STATE_PENDING)
{
CommandCounterIncrement(); /* make updates visible */
if (AllTablesyncsReady())
{
- ereport(LOG,
- (errmsg("logical replication apply worker for subscription \"%s\" will restart so that two_phase can be enabled",
- MySubscription->name)));
+ if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING)
+ ereport(LOG,
+ (errmsg("logical replication apply worker for subscription \"%s\" will restart so that two_phase can be enabled",
+ MySubscription->name)));
+
+ if (MySubscription->failoverstate == LOGICALREP_FAILOVER_STATE_PENDING)
+ ereport(LOG,
+ (errmsg("logical replication apply worker for subscription \"%s\" will restart so that failover can be enabled",
+ MySubscription->name)));
+
should_exit = true;
}
}
@@ -1420,7 +1431,8 @@ LogicalRepSyncTableStart(XLogRecPtr *origin_startpos)
*/
walrcv_create_slot(LogRepWorkerWalRcvConn,
slotname, false /* permanent */ , false /* two_phase */ ,
- CRS_USE_SNAPSHOT, origin_startpos);
+ false /* failover */ , CRS_USE_SNAPSHOT,
+ origin_startpos);
/*
* Setup replication origin tracking. The purpose of doing this before the
@@ -1722,10 +1734,12 @@ AllTablesyncsReady(void)
}
/*
- * Update the two_phase state of the specified subscription in pg_subscription.
+ * Update the twophase and/or failover state of the specified subscription
+ * in pg_subscription.
*/
void
-UpdateTwoPhaseState(Oid suboid, char new_state)
+EnableTwoPhaseFailoverTriState(Oid suboid, bool enable_twophase,
+ bool enable_failover)
{
Relation rel;
HeapTuple tup;
@@ -1733,9 +1747,8 @@ UpdateTwoPhaseState(Oid suboid, char new_state)
bool replaces[Natts_pg_subscription];
Datum values[Natts_pg_subscription];
- Assert(new_state == LOGICALREP_TWOPHASE_STATE_DISABLED ||
- new_state == LOGICALREP_TWOPHASE_STATE_PENDING ||
- new_state == LOGICALREP_TWOPHASE_STATE_ENABLED);
+ if (!enable_twophase && !enable_failover)
+ return;
rel = table_open(SubscriptionRelationId, RowExclusiveLock);
tup = SearchSysCacheCopy1(SUBSCRIPTIONOID, ObjectIdGetDatum(suboid));
@@ -1749,9 +1762,21 @@ UpdateTwoPhaseState(Oid suboid, char new_state)
memset(nulls, false, sizeof(nulls));
memset(replaces, false, sizeof(replaces));
- /* And update/set two_phase state */
- values[Anum_pg_subscription_subtwophasestate - 1] = CharGetDatum(new_state);
- replaces[Anum_pg_subscription_subtwophasestate - 1] = true;
+ /* Update/set two_phase state if asked by the caller */
+ if (enable_twophase)
+ {
+ values[Anum_pg_subscription_subtwophasestate - 1] =
+ CharGetDatum(LOGICALREP_TWOPHASE_STATE_ENABLED);
+ replaces[Anum_pg_subscription_subtwophasestate - 1] = true;
+ }
+
+ /* Update/set failover state if asked by the caller */
+ if (enable_failover)
+ {
+ values[Anum_pg_subscription_subfailoverstate - 1] =
+ CharGetDatum(LOGICALREP_FAILOVER_STATE_ENABLED);
+ replaces[Anum_pg_subscription_subfailoverstate - 1] = true;
+ }
tup = heap_modify_tuple(tup, RelationGetDescr(rel),
values, nulls, replaces);
diff --git a/src/backend/replication/logical/worker.c b/src/backend/replication/logical/worker.c
index 21abf34ef7..0b7d632922 100644
--- a/src/backend/replication/logical/worker.c
+++ b/src/backend/replication/logical/worker.c
@@ -132,9 +132,41 @@
* avoid such deadlocks, we generate a unique GID (consisting of the
* subscription oid and the xid of the prepared transaction) for each prepare
* transaction on the subscriber.
+ *
+ * FAILOVER
+ * ----------------------
+ * The logical slot on the primary can be synced to the standby by specifying
+ * the failover = true when creating the subscription. Enabling failover allows
+ * us to smoothly transition to the standby in case the primary gets promoted,
+ * ensuring that we can subscribe to the new primary without losing any data.
+ *
+ * However, we do not enable failover for slots created by the table sync
+ * worker. This is because the table sync slot might not be fully synced on the
+ * standby. During syncing, the local restart_lsn and/or local catalog_xmin of
+ * the newly created slot on the standby are typically ahead of those on the
+ * primary. Therefore, the standby needs to wait for the primary server's
+ * restart_lsn and catalog_xmin to catch up, which takes time.
+ *
+ * Additionally, failover is not enabled for the main slot if the table sync is
+ * in progress. This is because if a failover occurs while the table sync
+ * worker has reached a certain state (SUBREL_STATE_FINISHEDCOPY or
+ * SUBREL_STATE_DATASYNC), replication will not be able to continue from the
+ * new primary node.
+ *
+ * As a result, we enable the failover option for the main slot only after the
+ * initial sync is complete. The failover option is implemented as a tri-state
+ * with values DISABLED, PENDING, and ENABLED. The state transition process
+ * between these values is the same as the two_phase option (see TWO_PHASE
+ * TRANSACTIONS for details).
+ *
+ * During the startup of the apply worker, it checks if all table syncs are in
+ * the READY state for a failover tri-state of PENDING. If so, it alters the
+ * main slot's failover property to true and updates the tri-state value from
+ * PENDING to ENABLED.
*-------------------------------------------------------------------------
*/
+
#include "postgres.h"
#include <sys/stat.h>
@@ -3947,6 +3979,7 @@ maybe_reread_subscription(void)
newsub->passwordrequired != MySubscription->passwordrequired ||
strcmp(newsub->origin, MySubscription->origin) != 0 ||
newsub->owner != MySubscription->owner ||
+ newsub->failoverstate != MySubscription->failoverstate ||
!equal(newsub->publications, MySubscription->publications))
{
if (am_parallel_apply_worker())
@@ -4482,6 +4515,8 @@ run_apply_worker()
TimeLineID startpointTLI;
char *err;
bool must_use_password;
+ bool twophase_pending;
+ bool failover_pending;
slotname = MySubscription->slotname;
@@ -4538,17 +4573,38 @@ run_apply_worker()
* Note: If the subscription has no tables then leave the state as
* PENDING, which allows ALTER SUBSCRIPTION ... REFRESH PUBLICATION to
* work.
+ *
+ * Same goes for 'failover'. It is enabled only if subscription has tables
+ * and all the tablesyncs have reached READY state, until then it remains
+ * as PENDING.
*/
- if (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING &&
- AllTablesyncsReady())
+ twophase_pending =
+ (MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING);
+ failover_pending =
+ (MySubscription->failoverstate == LOGICALREP_FAILOVER_STATE_PENDING);
+
+ if ((twophase_pending || failover_pending) && AllTablesyncsReady())
{
/* Start streaming with two_phase enabled */
- options.proto.logical.twophase = true;
+ if (twophase_pending)
+ options.proto.logical.twophase = true;
+
+ if (failover_pending)
+ walrcv_alter_slot(LogRepWorkerWalRcvConn, slotname, true);
+
walrcv_startstreaming(LogRepWorkerWalRcvConn, &options);
StartTransactionCommand();
- UpdateTwoPhaseState(MySubscription->oid, LOGICALREP_TWOPHASE_STATE_ENABLED);
- MySubscription->twophasestate = LOGICALREP_TWOPHASE_STATE_ENABLED;
+
+ /* Update twophase and/or failover */
+ EnableTwoPhaseFailoverTriState(MySubscription->oid, twophase_pending,
+ failover_pending);
+ if (twophase_pending)
+ MySubscription->twophasestate = LOGICALREP_TWOPHASE_STATE_ENABLED;
+
+ if (failover_pending)
+ MySubscription->failoverstate = LOGICALREP_FAILOVER_STATE_ENABLED;
+
CommitTransactionCommand();
}
else
@@ -4557,11 +4613,15 @@ run_apply_worker()
}
ereport(DEBUG1,
- (errmsg_internal("logical replication apply worker for subscription \"%s\" two_phase is %s",
+ (errmsg_internal("logical replication apply worker for subscription \"%s\" two_phase is %s and failover is %s",
MySubscription->name,
MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_DISABLED ? "DISABLED" :
MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_PENDING ? "PENDING" :
MySubscription->twophasestate == LOGICALREP_TWOPHASE_STATE_ENABLED ? "ENABLED" :
+ "?",
+ MySubscription->failoverstate == LOGICALREP_FAILOVER_STATE_DISABLED ? "DISABLED" :
+ MySubscription->failoverstate == LOGICALREP_FAILOVER_STATE_PENDING ? "PENDING" :
+ MySubscription->failoverstate == LOGICALREP_FAILOVER_STATE_ENABLED ? "ENABLED" :
"?")));
/* Run the main loop. */
diff --git a/src/backend/replication/repl_gram.y b/src/backend/replication/repl_gram.y
index 0c874e33cf..b706046811 100644
--- a/src/backend/replication/repl_gram.y
+++ b/src/backend/replication/repl_gram.y
@@ -64,6 +64,7 @@ Node *replication_parse_result;
%token K_START_REPLICATION
%token K_CREATE_REPLICATION_SLOT
%token K_DROP_REPLICATION_SLOT
+%token K_ALTER_REPLICATION_SLOT
%token K_TIMELINE_HISTORY
%token K_WAIT
%token K_TIMELINE
@@ -79,7 +80,8 @@ Node *replication_parse_result;
%type <node> command
%type <node> base_backup start_replication start_logical_replication
- create_replication_slot drop_replication_slot identify_system
+ create_replication_slot drop_replication_slot
+ alter_replication_slot identify_system
read_replication_slot timeline_history show
%type <list> generic_option_list
%type <defelt> generic_option
@@ -111,6 +113,7 @@ command:
| start_logical_replication
| create_replication_slot
| drop_replication_slot
+ | alter_replication_slot
| read_replication_slot
| timeline_history
| show
@@ -257,6 +260,18 @@ drop_replication_slot:
}
;
+/* ALTER_REPLICATION_SLOT slot */
+alter_replication_slot:
+ K_ALTER_REPLICATION_SLOT IDENT '(' generic_option_list ')'
+ {
+ AlterReplicationSlotCmd *cmd;
+ cmd = makeNode(AlterReplicationSlotCmd);
+ cmd->slotname = $2;
+ cmd->options = $4;
+ $$ = (Node *) cmd;
+ }
+ ;
+
/*
* START_REPLICATION [SLOT slot] [PHYSICAL] %X/%X [TIMELINE %d]
*/
@@ -399,6 +414,7 @@ ident_or_keyword:
| K_START_REPLICATION { $$ = "start_replication"; }
| K_CREATE_REPLICATION_SLOT { $$ = "create_replication_slot"; }
| K_DROP_REPLICATION_SLOT { $$ = "drop_replication_slot"; }
+ | K_ALTER_REPLICATION_SLOT { $$ = "alter_replication_slot"; }
| K_TIMELINE_HISTORY { $$ = "timeline_history"; }
| K_WAIT { $$ = "wait"; }
| K_TIMELINE { $$ = "timeline"; }
diff --git a/src/backend/replication/repl_scanner.l b/src/backend/replication/repl_scanner.l
index 1cc7fb858c..0b5ae23195 100644
--- a/src/backend/replication/repl_scanner.l
+++ b/src/backend/replication/repl_scanner.l
@@ -125,6 +125,7 @@ TIMELINE { return K_TIMELINE; }
START_REPLICATION { return K_START_REPLICATION; }
CREATE_REPLICATION_SLOT { return K_CREATE_REPLICATION_SLOT; }
DROP_REPLICATION_SLOT { return K_DROP_REPLICATION_SLOT; }
+ALTER_REPLICATION_SLOT { return K_ALTER_REPLICATION_SLOT; }
TIMELINE_HISTORY { return K_TIMELINE_HISTORY; }
PHYSICAL { return K_PHYSICAL; }
RESERVE_WAL { return K_RESERVE_WAL; }
@@ -301,6 +302,7 @@ replication_scanner_is_replication_command(void)
case K_START_REPLICATION:
case K_CREATE_REPLICATION_SLOT:
case K_DROP_REPLICATION_SLOT:
+ case K_ALTER_REPLICATION_SLOT:
case K_READ_REPLICATION_SLOT:
case K_TIMELINE_HISTORY:
case K_SHOW:
diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c
index 18bc28195b..c96a1700cd 100644
--- a/src/backend/replication/slot.c
+++ b/src/backend/replication/slot.c
@@ -52,6 +52,9 @@
#include "storage/proc.h"
#include "storage/procarray.h"
#include "utils/builtins.h"
+#include "utils/guc_hooks.h"
+#include "utils/memutils.h"
+#include "utils/varlena.h"
/*
* Replication slot on-disk data structure.
@@ -90,7 +93,7 @@ typedef struct ReplicationSlotOnDisk
sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize
#define SLOT_MAGIC 0x1051CA1 /* format identifier */
-#define SLOT_VERSION 3 /* version for new files */
+#define SLOT_VERSION 4 /* version for new files */
/* Control array for replication slot management */
ReplicationSlotCtlData *ReplicationSlotCtl = NULL;
@@ -98,10 +101,19 @@ ReplicationSlotCtlData *ReplicationSlotCtl = NULL;
/* My backend's replication slot in the shared memory array */
ReplicationSlot *MyReplicationSlot = NULL;
-/* GUC variable */
+/* GUC variables */
int max_replication_slots = 10; /* the maximum number of replication
* slots */
+/*
+ * This GUC lists streaming replication standby server slot names that
+ * logical WAL sender processes will wait for.
+ */
+char *standby_slot_names;
+
+/* This is parsed and cached list for raw standby_slot_names. */
+static List *standby_slot_names_list = NIL;
+
static void ReplicationSlotShmemExit(int code, Datum arg);
static void ReplicationSlotDropAcquired(void);
static void ReplicationSlotDropPtr(ReplicationSlot *slot);
@@ -251,7 +263,8 @@ ReplicationSlotValidateName(const char *name, int elevel)
*/
void
ReplicationSlotCreate(const char *name, bool db_specific,
- ReplicationSlotPersistency persistency, bool two_phase)
+ ReplicationSlotPersistency persistency,
+ bool two_phase, bool failover)
{
ReplicationSlot *slot = NULL;
int i;
@@ -311,6 +324,7 @@ ReplicationSlotCreate(const char *name, bool db_specific,
slot->data.persistency = persistency;
slot->data.two_phase = two_phase;
slot->data.two_phase_at = InvalidXLogRecPtr;
+ slot->data.failover = failover;
/* and then data only present in shared memory */
slot->just_dirtied = false;
@@ -679,6 +693,31 @@ ReplicationSlotDrop(const char *name, bool nowait)
ReplicationSlotDropAcquired();
}
+/*
+ * Change the definition of the slot identified by the specified name.
+ */
+void
+ReplicationSlotAlter(const char *name, bool failover)
+{
+ Assert(MyReplicationSlot == NULL);
+
+ ReplicationSlotAcquire(name, true);
+
+ if (SlotIsPhysical(MyReplicationSlot))
+ ereport(ERROR,
+ errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("cannot use %s with a physical replication slot",
+ "ALTER_REPLICATION_SLOT"));
+
+ SpinLockAcquire(&MyReplicationSlot->mutex);
+ MyReplicationSlot->data.failover = failover;
+ SpinLockRelease(&MyReplicationSlot->mutex);
+
+ ReplicationSlotMarkDirty();
+ ReplicationSlotSave();
+ ReplicationSlotRelease();
+}
+
/*
* Permanently drop the currently acquired replication slot.
*/
@@ -2159,3 +2198,140 @@ RestoreSlotFromDisk(const char *name)
(errmsg("too many replication slots active before shutdown"),
errhint("Increase max_replication_slots and try again.")));
}
+
+/*
+ * A helper function to validate slots specified in GUC standby_slot_names.
+ */
+static bool
+validate_standby_slots(char **newval)
+{
+ char *rawname;
+ List *elemlist;
+ ListCell *lc;
+
+ /* Need a modifiable copy of string */
+ rawname = pstrdup(*newval);
+
+ /* Verify syntax and parse string into list of identifiers */
+ if (!SplitIdentifierString(rawname, ',', &elemlist))
+ {
+ /* syntax error in name list */
+ GUC_check_errdetail("List syntax is invalid.");
+ goto ret_standby_slot_names_ng;
+ }
+
+ /*
+ * Verify 'type' of slot now.
+ *
+ * Skip check if replication slots' data is not initialized yet i.e. we
+ * are in startup process.
+ */
+ if (!ReplicationSlotCtl)
+ goto ret_standby_slot_names_ok;
+
+ foreach(lc, elemlist)
+ {
+ char *name = lfirst(lc);
+ ReplicationSlot *slot;
+
+ slot = SearchNamedReplicationSlot(name, true);
+
+ if (!slot)
+ goto ret_standby_slot_names_ng;
+
+ if (SlotIsLogical(slot))
+ {
+ GUC_check_errdetail("\"%s\" is not a physical replication slot",
+ name);
+ goto ret_standby_slot_names_ng;
+ }
+ }
+
+ret_standby_slot_names_ok:
+
+ pfree(rawname);
+ list_free(elemlist);
+ return true;
+
+ret_standby_slot_names_ng:
+
+ pfree(rawname);
+ list_free(elemlist);
+ return false;
+}
+
+/*
+ * GUC check_hook for standby_slot_names
+ */
+bool
+check_standby_slot_names(char **newval, void **extra, GucSource source)
+{
+ if (strcmp(*newval, "") == 0)
+ return true;
+
+ /*
+ * "*" is not accepted as in that case primary will not be able to know
+ * for which all standbys to wait for. Even if we have physical-slots
+ * info, there is no way to confirm whether there is any standby
+ * configured for the known physical slots.
+ */
+ if (strcmp(*newval, "*") == 0)
+ {
+ GUC_check_errdetail("\"%s\" is not accepted for standby_slot_names",
+ *newval);
+ return false;
+ }
+
+ /* Now verify if the specified slots really exist and have correct type */
+ if (!validate_standby_slots(newval))
+ return false;
+
+ *extra = guc_strdup(ERROR, *newval);
+
+ return true;
+}
+
+/*
+ * GUC assign_hook for standby_slot_names
+ */
+void
+assign_standby_slot_names(const char *newval, void *extra)
+{
+ List *standby_slots;
+ MemoryContext oldcxt;
+ char *standby_slot_names_cpy = extra;
+
+ list_free(standby_slot_names_list);
+ standby_slot_names_list = NIL;
+
+ /* No value is specified for standby_slot_names. */
+ if (standby_slot_names_cpy == NULL)
+ return;
+
+ if (!SplitIdentifierString(standby_slot_names_cpy, ',', &standby_slots))
+ {
+ /* This should not happen if GUC checked check_standby_slot_names. */
+ elog(ERROR, "invalid list syntax");
+ }
+
+ /*
+ * Switch to the same memory context under which GUC variables are
+ * allocated (GUCMemoryContext).
+ */
+ oldcxt = MemoryContextSwitchTo(GetMemoryChunkContext(standby_slot_names_cpy));
+ standby_slot_names_list = list_copy(standby_slots);
+ MemoryContextSwitchTo(oldcxt);
+}
+
+/*
+ * Return a copy of standby_slot_names_list if the copy flag is set to true,
+ * otherwise return the original list.
+ */
+List *
+GetStandbySlotList(bool copy)
+{
+ if (copy)
+ return list_copy(standby_slot_names_list);
+ else
+ return standby_slot_names_list;
+}
diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c
index 4b694a03d0..fb6e37d2c3 100644
--- a/src/backend/replication/slotfuncs.c
+++ b/src/backend/replication/slotfuncs.c
@@ -21,6 +21,7 @@
#include "replication/decode.h"
#include "replication/logical.h"
#include "replication/slot.h"
+#include "replication/walsender.h"
#include "utils/builtins.h"
#include "utils/inval.h"
#include "utils/pg_lsn.h"
@@ -42,7 +43,8 @@ create_physical_replication_slot(char *name, bool immediately_reserve,
/* acquire replication slot, this will check for conflicting names */
ReplicationSlotCreate(name, false,
- temporary ? RS_TEMPORARY : RS_PERSISTENT, false);
+ temporary ? RS_TEMPORARY : RS_PERSISTENT, false,
+ false);
if (immediately_reserve)
{
@@ -117,6 +119,7 @@ pg_create_physical_replication_slot(PG_FUNCTION_ARGS)
static void
create_logical_replication_slot(char *name, char *plugin,
bool temporary, bool two_phase,
+ bool failover,
XLogRecPtr restart_lsn,
bool find_startpoint)
{
@@ -133,7 +136,8 @@ create_logical_replication_slot(char *name, char *plugin,
* error as well.
*/
ReplicationSlotCreate(name, true,
- temporary ? RS_TEMPORARY : RS_EPHEMERAL, two_phase);
+ temporary ? RS_TEMPORARY : RS_EPHEMERAL, two_phase,
+ failover);
/*
* Create logical decoding context to find start point or, if we don't
@@ -171,6 +175,7 @@ pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
Name plugin = PG_GETARG_NAME(1);
bool temporary = PG_GETARG_BOOL(2);
bool two_phase = PG_GETARG_BOOL(3);
+ bool failover = PG_GETARG_BOOL(4);
Datum result;
TupleDesc tupdesc;
HeapTuple tuple;
@@ -188,6 +193,7 @@ pg_create_logical_replication_slot(PG_FUNCTION_ARGS)
NameStr(*plugin),
temporary,
two_phase,
+ failover,
InvalidXLogRecPtr,
true);
@@ -232,7 +238,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS)
Datum
pg_get_replication_slots(PG_FUNCTION_ARGS)
{
-#define PG_GET_REPLICATION_SLOTS_COLS 15
+#define PG_GET_REPLICATION_SLOTS_COLS 16
ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
XLogRecPtr currlsn;
int slotno;
@@ -412,6 +418,8 @@ pg_get_replication_slots(PG_FUNCTION_ARGS)
values[i++] = BoolGetDatum(false);
}
+ values[i++] = BoolGetDatum(slot_contents.data.failover);
+
Assert(i == PG_GET_REPLICATION_SLOTS_COLS);
tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc,
@@ -451,6 +459,8 @@ pg_physical_replication_slot_advance(XLogRecPtr moveto)
* crash, but this makes the data consistent after a clean shutdown.
*/
ReplicationSlotMarkDirty();
+
+ PhysicalWakeupLogicalWalSnd();
}
return retlsn;
@@ -679,6 +689,7 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
XLogRecPtr src_restart_lsn;
bool src_islogical;
bool temporary;
+ bool failover;
char *plugin;
Datum values[2];
bool nulls[2];
@@ -734,6 +745,7 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
src_islogical = SlotIsLogical(&first_slot_contents);
src_restart_lsn = first_slot_contents.data.restart_lsn;
temporary = (first_slot_contents.data.persistency == RS_TEMPORARY);
+ failover = first_slot_contents.data.failover;
plugin = logical_slot ? NameStr(first_slot_contents.data.plugin) : NULL;
/* Check type of replication slot */
@@ -773,6 +785,7 @@ copy_replication_slot(FunctionCallInfo fcinfo, bool logical_slot)
plugin,
temporary,
false,
+ failover,
src_restart_lsn,
false);
}
diff --git a/src/backend/replication/walreceiver.c b/src/backend/replication/walreceiver.c
index 2398167f49..e27d231174 100644
--- a/src/backend/replication/walreceiver.c
+++ b/src/backend/replication/walreceiver.c
@@ -386,7 +386,7 @@ WalReceiverMain(void)
"pg_walreceiver_%lld",
(long long int) walrcv_get_backend_pid(wrconn));
- walrcv_create_slot(wrconn, slotname, true, false, 0, NULL);
+ walrcv_create_slot(wrconn, slotname, true, false, false, 0, NULL);
SpinLockAcquire(&walrcv->mutex);
strlcpy(walrcv->slotname, slotname, NAMEDATALEN);
diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c
index 3bc9c82389..3e44228bde 100644
--- a/src/backend/replication/walsender.c
+++ b/src/backend/replication/walsender.c
@@ -974,12 +974,13 @@ static void
parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
bool *reserve_wal,
CRSSnapshotAction *snapshot_action,
- bool *two_phase)
+ bool *two_phase, bool *failover)
{
ListCell *lc;
bool snapshot_action_given = false;
bool reserve_wal_given = false;
bool two_phase_given = false;
+ bool failover_given = false;
/* Parse options */
foreach(lc, cmd->options)
@@ -1029,6 +1030,15 @@ parseCreateReplSlotOptions(CreateReplicationSlotCmd *cmd,
two_phase_given = true;
*two_phase = defGetBoolean(defel);
}
+ else if (strcmp(defel->defname, "failover") == 0)
+ {
+ if (failover_given || cmd->kind != REPLICATION_KIND_LOGICAL)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("conflicting or redundant options")));
+ failover_given = true;
+ *failover = defGetBoolean(defel);
+ }
else
elog(ERROR, "unrecognized option: %s", defel->defname);
}
@@ -1045,6 +1055,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
char *slot_name;
bool reserve_wal = false;
bool two_phase = false;
+ bool failover = false;
CRSSnapshotAction snapshot_action = CRS_EXPORT_SNAPSHOT;
DestReceiver *dest;
TupOutputState *tstate;
@@ -1054,13 +1065,13 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
Assert(!MyReplicationSlot);
- parseCreateReplSlotOptions(cmd, &reserve_wal, &snapshot_action, &two_phase);
-
+ parseCreateReplSlotOptions(cmd, &reserve_wal, &snapshot_action, &two_phase,
+ &failover);
if (cmd->kind == REPLICATION_KIND_PHYSICAL)
{
ReplicationSlotCreate(cmd->slotname, false,
cmd->temporary ? RS_TEMPORARY : RS_PERSISTENT,
- false);
+ false, false);
if (reserve_wal)
{
@@ -1091,7 +1102,7 @@ CreateReplicationSlot(CreateReplicationSlotCmd *cmd)
*/
ReplicationSlotCreate(cmd->slotname, true,
cmd->temporary ? RS_TEMPORARY : RS_EPHEMERAL,
- two_phase);
+ two_phase, failover);
/*
* Do options check early so that we can bail before calling the
@@ -1246,6 +1257,46 @@ DropReplicationSlot(DropReplicationSlotCmd *cmd)
ReplicationSlotDrop(cmd->slotname, !cmd->wait);
}
+/*
+ * Process extra options given to ALTER_REPLICATION_SLOT.
+ */
+static void
+parseAlterReplSlotOptions(AlterReplicationSlotCmd *cmd, bool *failover)
+{
+ ListCell *lc;
+ bool failover_given = false;
+
+ /* Parse options */
+ foreach(lc, cmd->options)
+ {
+ DefElem *defel = (DefElem *) lfirst(lc);
+
+ if (strcmp(defel->defname, "failover") == 0)
+ {
+ if (failover_given)
+ ereport(ERROR,
+ (errcode(ERRCODE_SYNTAX_ERROR),
+ errmsg("conflicting or redundant options")));
+ failover_given = true;
+ *failover = defGetBoolean(defel);
+ }
+ else
+ elog(ERROR, "unrecognized option: %s", defel->defname);
+ }
+}
+
+/*
+ * Change the definition of a replication slot.
+ */
+static void
+AlterReplicationSlot(AlterReplicationSlotCmd *cmd)
+{
+ bool failover = false;
+
+ parseAlterReplSlotOptions(cmd, &failover);
+ ReplicationSlotAlter(cmd->slotname, failover);
+}
+
/*
* Load previously initiated logical slot and prepare for sending data (via
* WalSndLoop).
@@ -1527,27 +1578,233 @@ WalSndUpdateProgress(LogicalDecodingContext *ctx, XLogRecPtr lsn, TransactionId
ProcessPendingWrites();
}
+/*
+ * Wake up logical walsenders with failover-enabled slots if the physical slot
+ * of the current walsender is specified in standby_slot_names GUC.
+ */
+void
+PhysicalWakeupLogicalWalSnd(void)
+{
+ ListCell *lc;
+ List *standby_slots;
+
+ Assert(MyReplicationSlot && SlotIsPhysical(MyReplicationSlot));
+
+ standby_slots = GetStandbySlotList(false);
+
+ foreach(lc, standby_slots)
+ {
+ char *name = lfirst(lc);
+
+ if (strcmp(name, NameStr(MyReplicationSlot->data.name)) == 0)
+ {
+ ConditionVariableBroadcast(&WalSndCtl->wal_confirm_rcv_cv);
+ return;
+ }
+ }
+}
+
+/*
+ * Reload the config file and reinitialize the standby slot list if the GUC
+ * standby_slot_names has changed.
+ */
+static void
+WalSndRereadConfigAndReInitSlotList(List **standby_slots)
+{
+ char *pre_standby_slot_names = pstrdup(standby_slot_names);
+
+ ProcessConfigFile(PGC_SIGHUP);
+
+ if (strcmp(pre_standby_slot_names, standby_slot_names) != 0)
+ {
+ list_free(*standby_slots);
+ *standby_slots = GetStandbySlotList(true);
+ }
+
+ pfree(pre_standby_slot_names);
+}
+
+/*
+ * Filter the standby slots based on the specified log sequence number
+ * (wait_for_lsn).
+ *
+ * This function updates the passed standby_slots list, removing any slots that
+ * have already caught up to or surpassed the given wait_for_lsn.
+ */
+static void
+WalSndFilterStandbySlots(XLogRecPtr wait_for_lsn, List **standby_slots)
+{
+ ListCell *lc;
+ List *standby_slots_cpy = *standby_slots;
+
+ foreach(lc, standby_slots_cpy)
+ {
+ char *name = lfirst(lc);
+ XLogRecPtr restart_lsn = InvalidXLogRecPtr;
+ bool invalidated = false;
+ char *warningfmt = NULL;
+ ReplicationSlot *slot;
+
+ slot = SearchNamedReplicationSlot(name, true);
+
+ if (slot && SlotIsPhysical(slot))
+ {
+ SpinLockAcquire(&slot->mutex);
+ restart_lsn = slot->data.restart_lsn;
+ invalidated = slot->data.invalidated != RS_INVAL_NONE;
+ SpinLockRelease(&slot->mutex);
+ }
+
+ /* Continue if the current slot hasn't caught up. */
+ if (!invalidated && !XLogRecPtrIsInvalid(restart_lsn) &&
+ restart_lsn < wait_for_lsn)
+ {
+ /* Log warning if no active_pid for this physical slot */
+ if (slot->active_pid == 0)
+ ereport(WARNING,
+ errmsg("replication slot \"%s\" specified in parameter \"%s\" does not have active_pid",
+ name, "standby_slot_names"),
+ errdetail("Logical replication is waiting on the "
+ "standby associated with \"%s\"", name),
+ errhint("Consider starting standby associated with "
+ "\"%s\" or amend standby_slot_names", name));
+
+ continue;
+ }
+
+ /*
+ * It may happen that the slot specified in standby_slot_names GUC
+ * value is dropped, so let's skip over it.
+ */
+ else if (!slot)
+ warningfmt = _("replication slot \"%s\" specified in parameter \"%s\" does not exist, ignoring");
+
+ /*
+ * If logical slot name is given in standby_slot_names, give WARNING
+ * and skip it. Since it is harmless, so WARNING should be enough, no
+ * need to error-out.
+ */
+ else if (SlotIsLogical(slot))
+ warningfmt = _("cannot have logical replication slot \"%s\" in parameter \"%s\", ignoring");
+
+ /*
+ * Specified physical slot may have been invalidated, so no point in
+ * waiting for it.
+ */
+ else if (XLogRecPtrIsInvalid(restart_lsn) || invalidated)
+ warningfmt = _("physical slot \"%s\" specified in parameter \"%s\" has been invalidated, ignoring");
+ else
+ Assert(restart_lsn >= wait_for_lsn);
+
+ /*
+ * Reaching here indicates that either the slot has passed the
+ * wait_for_lsn or there is an issue with the slot that requires a
+ * warning to be reported.
+ */
+ if (warningfmt)
+ ereport(WARNING, errmsg(warningfmt, name, "standby_slot_names"));
+
+ standby_slots_cpy = foreach_delete_current(standby_slots_cpy, lc);
+ }
+
+ *standby_slots = standby_slots_cpy;
+}
+
+/*
+ * Wait for physical standby to confirm receiving given lsn.
+ *
+ * Used by logical decoding SQL functions that acquired slot with failover
+ * enabled. It waits for physical standbys corresponding to the physical slots
+ * specified in the standby_slot_names GUC.
+ */
+void
+WalSndWaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
+{
+ List *standby_slots;
+
+ Assert(!am_walsender);
+
+ if (!MyReplicationSlot->data.failover)
+ return;
+
+ standby_slots = GetStandbySlotList(true);
+
+ ConditionVariablePrepareToSleep(&WalSndCtl->wal_confirm_rcv_cv);
+
+ for (;;)
+ {
+ long sleeptime = -1;
+
+ CHECK_FOR_INTERRUPTS();
+
+ if (ConfigReloadPending)
+ {
+ ConfigReloadPending = false;
+ WalSndRereadConfigAndReInitSlotList(&standby_slots);
+ }
+
+ WalSndFilterStandbySlots(wait_for_lsn, &standby_slots);
+
+ /* Exit if done waiting for every slot. */
+ if (standby_slots == NIL)
+ break;
+
+ sleeptime = WalSndComputeSleeptime(GetCurrentTimestamp());
+
+ ConditionVariableTimedSleep(&WalSndCtl->wal_confirm_rcv_cv, sleeptime,
+ WAIT_EVENT_WAL_SENDER_WAIT_FOR_STANDBY_CONFIRMATION);
+ }
+
+ ConditionVariableCancelSleep();
+ list_free(standby_slots);
+}
+
/*
* Wait till WAL < loc is flushed to disk so it can be safely sent to client.
*
- * Returns end LSN of flushed WAL. Normally this will be >= loc, but
- * if we detect a shutdown request (either from postmaster or client)
- * we will return early, so caller must always check.
+ * If the walsender holds a logical slot that has enabled failover, the
+ * function also waits for all the specified streaming replication standby
+ * servers to confirm receipt of WAL up to RecentFlushPtr.
+ *
+ * Returns end LSN of flushed WAL. Normally this will be >= loc, but if we
+ * detect a shutdown request (either from postmaster or client) we will return
+ * early, so caller must always check.
*/
static XLogRecPtr
WalSndWaitForWal(XLogRecPtr loc)
{
int wakeEvents;
+ bool wait_for_standby = false;
+ uint32 wait_event;
+ List *standby_slots = NIL;
static XLogRecPtr RecentFlushPtr = InvalidXLogRecPtr;
+ if (MyReplicationSlot->data.failover)
+ standby_slots = GetStandbySlotList(true);
+
/*
- * Fast path to avoid acquiring the spinlock in case we already know we
- * have enough WAL available. This is particularly interesting if we're
- * far behind.
+ * Check if all the standby servers have confirmed receipt of WAL up to
+ * RecentFlushPtr if we already know we have enough WAL available.
+ *
+ * Note that we cannot directly return without checking the status of
+ * standby servers because the standby_slot_names may have changed, which
+ * means there could be new standby slots in the list that have not yet
+ * caught up to the RecentFlushPtr.
*/
if (RecentFlushPtr != InvalidXLogRecPtr &&
loc <= RecentFlushPtr)
- return RecentFlushPtr;
+ {
+ WalSndFilterStandbySlots(RecentFlushPtr, &standby_slots);
+
+ /*
+ * Fast path to avoid acquiring the spinlock in case we already know we
+ * have enough WAL available and all the standby servers have confirmed
+ * receipt of WAL up to RecentFlushPtr. This is particularly interesting
+ * if we're far behind.
+ */
+ if (standby_slots == NIL)
+ return RecentFlushPtr;
+ }
/* Get a more recent flush pointer. */
if (!RecoveryInProgress())
@@ -1568,7 +1825,7 @@ WalSndWaitForWal(XLogRecPtr loc)
if (ConfigReloadPending)
{
ConfigReloadPending = false;
- ProcessConfigFile(PGC_SIGHUP);
+ WalSndRereadConfigAndReInitSlotList(&standby_slots);
SyncRepInitConfig();
}
@@ -1583,8 +1840,18 @@ WalSndWaitForWal(XLogRecPtr loc)
if (got_STOPPING)
XLogBackgroundFlush();
+ /*
+ * Update the standby slots that have not yet caught up to the flushed
+ * position. It is good to wait up to RecentFlushPtr and then let it
+ * send the changes to logical subscribers one by one which are
+ * already covered in RecentFlushPtr without needing to wait on every
+ * change for standby confirmation.
+ */
+ if (wait_for_standby)
+ WalSndFilterStandbySlots(RecentFlushPtr, &standby_slots);
+
/* Update our idea of the currently flushed position. */
- if (!RecoveryInProgress())
+ else if (!RecoveryInProgress())
RecentFlushPtr = GetFlushRecPtr(NULL);
else
RecentFlushPtr = GetXLogReplayRecPtr(NULL);
@@ -1612,8 +1879,14 @@ WalSndWaitForWal(XLogRecPtr loc)
!waiting_for_ping_response)
WalSndKeepalive(false, InvalidXLogRecPtr);
- /* check whether we're done */
- if (loc <= RecentFlushPtr)
+ if (loc > RecentFlushPtr)
+ wait_event = WAIT_EVENT_WAL_SENDER_WAIT_FOR_WAL;
+ else if (standby_slots)
+ {
+ wait_event = WAIT_EVENT_WAL_SENDER_WAIT_FOR_STANDBY_CONFIRMATION;
+ wait_for_standby = true;
+ }
+ else
break;
/* Waiting for new WAL. Since we need to wait, we're now caught up. */
@@ -1654,9 +1927,11 @@ WalSndWaitForWal(XLogRecPtr loc)
if (pq_is_send_pending())
wakeEvents |= WL_SOCKET_WRITEABLE;
- WalSndWait(wakeEvents, sleeptime, WAIT_EVENT_WAL_SENDER_WAIT_FOR_WAL);
+ WalSndWait(wakeEvents, sleeptime, wait_event);
}
+ list_free(standby_slots);
+
/* reactivate latch so WalSndLoop knows to continue */
SetLatch(MyLatch);
return RecentFlushPtr;
@@ -1819,6 +2094,13 @@ exec_replication_command(const char *cmd_string)
EndReplicationCommand(cmdtag);
break;
+ case T_AlterReplicationSlotCmd:
+ cmdtag = "ALTER_REPLICATION_SLOT";
+ set_ps_display(cmdtag);
+ AlterReplicationSlot((AlterReplicationSlotCmd *) cmd_node);
+ EndReplicationCommand(cmdtag);
+ break;
+
case T_StartReplicationCmd:
{
StartReplicationCmd *cmd = (StartReplicationCmd *) cmd_node;
@@ -2049,6 +2331,7 @@ PhysicalConfirmReceivedLocation(XLogRecPtr lsn)
{
ReplicationSlotMarkDirty();
ReplicationSlotsComputeRequiredLSN();
+ PhysicalWakeupLogicalWalSnd();
}
/*
@@ -3311,6 +3594,8 @@ WalSndShmemInit(void)
ConditionVariableInit(&WalSndCtl->wal_flush_cv);
ConditionVariableInit(&WalSndCtl->wal_replay_cv);
+
+ ConditionVariableInit(&WalSndCtl->wal_confirm_rcv_cv);
}
}
@@ -3380,8 +3665,14 @@ WalSndWait(uint32 socket_events, long timeout, uint32 wait_event)
*
* And, we use separate shared memory CVs for physical and logical
* walsenders for selective wake ups, see WalSndWakeup() for more details.
+ *
+ * When the wait event is WAIT_FOR_STANDBY_CONFIRMATION, wait on another CV
+ * that is woken up by physical walsenders when the walreceiver has
+ * confirmed the receipt of LSN.
*/
- if (MyWalSnd->kind == REPLICATION_KIND_PHYSICAL)
+ if (wait_event == WAIT_EVENT_WAL_SENDER_WAIT_FOR_STANDBY_CONFIRMATION)
+ ConditionVariablePrepareToSleep(&WalSndCtl->wal_confirm_rcv_cv);
+ else if (MyWalSnd->kind == REPLICATION_KIND_PHYSICAL)
ConditionVariablePrepareToSleep(&WalSndCtl->wal_flush_cv);
else if (MyWalSnd->kind == REPLICATION_KIND_LOGICAL)
ConditionVariablePrepareToSleep(&WalSndCtl->wal_replay_cv);
diff --git a/src/backend/utils/activity/wait_event_names.txt b/src/backend/utils/activity/wait_event_names.txt
index d7995931bd..ede94a1ede 100644
--- a/src/backend/utils/activity/wait_event_names.txt
+++ b/src/backend/utils/activity/wait_event_names.txt
@@ -76,6 +76,7 @@ LIBPQWALRECEIVER_CONNECT "Waiting in WAL receiver to establish connection to rem
LIBPQWALRECEIVER_RECEIVE "Waiting in WAL receiver to receive data from remote server."
SSL_OPEN_SERVER "Waiting for SSL while attempting connection."
WAL_SENDER_WAIT_FOR_WAL "Waiting for WAL to be flushed in WAL sender process."
+WAL_SENDER_WAIT_FOR_STANDBY_CONFIRMATION "Waiting for the WAL to be received by physical standby in WAL sender process."
WAL_SENDER_WRITE_DATA "Waiting for any activity when processing replies from WAL receiver in WAL sender process."
diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c
index b764ef6998..e5e7bb23f9 100644
--- a/src/backend/utils/misc/guc_tables.c
+++ b/src/backend/utils/misc/guc_tables.c
@@ -4562,6 +4562,20 @@ struct config_string ConfigureNamesString[] =
check_debug_io_direct, assign_debug_io_direct, NULL
},
+ {
+ {"standby_slot_names", PGC_SIGHUP, REPLICATION_PRIMARY,
+ gettext_noop("Lists streaming replication standby server slot "
+ "names that logical WAL sender processes will wait for."),
+ gettext_noop("Decoded changes are sent out to plugins by logical "
+ "WAL sender processes only after specified "
+ "replication slots confirm receiving WAL."),
+ GUC_LIST_INPUT | GUC_LIST_QUOTE
+ },
+ &standby_slot_names,
+ "",
+ check_standby_slot_names, assign_standby_slot_names, NULL
+ },
+
/* End-of-list marker */
{
{NULL, 0, 0, NULL, NULL}, NULL, NULL, NULL, NULL, NULL
diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample
index e48c066a5b..998080e0e4 100644
--- a/src/backend/utils/misc/postgresql.conf.sample
+++ b/src/backend/utils/misc/postgresql.conf.sample
@@ -326,6 +326,8 @@
# method to choose sync standbys, number of sync standbys,
# and comma-separated list of application_name
# from standby(s); '*' = all
+#standby_slot_names = '' # streaming replication standby server slot names that
+ # logical walsender processes will wait for
# - Standby Servers -
diff --git a/src/bin/pg_upgrade/info.c b/src/bin/pg_upgrade/info.c
index 4878aa22bf..a348932e62 100644
--- a/src/bin/pg_upgrade/info.c
+++ b/src/bin/pg_upgrade/info.c
@@ -661,7 +661,7 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check)
* started and stopped several times causing any temporary slots to be
* removed.
*/
- res = executeQueryOrDie(conn, "SELECT slot_name, plugin, two_phase, "
+ res = executeQueryOrDie(conn, "SELECT slot_name, plugin, two_phase, failover, "
"%s as caught_up, conflicting as invalid "
"FROM pg_catalog.pg_replication_slots "
"WHERE slot_type = 'logical' AND "
@@ -679,14 +679,17 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check)
int i_slotname;
int i_plugin;
int i_twophase;
+ int i_failover;
int i_caught_up;
int i_invalid;
+
slotinfos = (LogicalSlotInfo *) pg_malloc(sizeof(LogicalSlotInfo) * num_slots);
i_slotname = PQfnumber(res, "slot_name");
i_plugin = PQfnumber(res, "plugin");
i_twophase = PQfnumber(res, "two_phase");
+ i_failover = PQfnumber(res, "failover");
i_caught_up = PQfnumber(res, "caught_up");
i_invalid = PQfnumber(res, "invalid");
@@ -697,6 +700,7 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check)
curr->slotname = pg_strdup(PQgetvalue(res, slotnum, i_slotname));
curr->plugin = pg_strdup(PQgetvalue(res, slotnum, i_plugin));
curr->two_phase = (strcmp(PQgetvalue(res, slotnum, i_twophase), "t") == 0);
+ curr->failover = (strcmp(PQgetvalue(res, slotnum, i_failover), "t") == 0);
curr->caught_up = (strcmp(PQgetvalue(res, slotnum, i_caught_up), "t") == 0);
curr->invalid = (strcmp(PQgetvalue(res, slotnum, i_invalid), "t") == 0);
}
diff --git a/src/bin/pg_upgrade/pg_upgrade.c b/src/bin/pg_upgrade/pg_upgrade.c
index 3960af4036..09f7437716 100644
--- a/src/bin/pg_upgrade/pg_upgrade.c
+++ b/src/bin/pg_upgrade/pg_upgrade.c
@@ -916,8 +916,10 @@ create_logical_replication_slots(void)
appendStringLiteralConn(query, slot_info->slotname, conn);
appendPQExpBuffer(query, ", ");
appendStringLiteralConn(query, slot_info->plugin, conn);
- appendPQExpBuffer(query, ", false, %s);",
- slot_info->two_phase ? "true" : "false");
+
+ appendPQExpBuffer(query, ", false, %s, %s);",
+ slot_info->two_phase ? "true" : "false",
+ slot_info->failover ? "true" : "false");
PQclear(executeQueryOrDie(conn, "%s", query->data));
diff --git a/src/bin/pg_upgrade/pg_upgrade.h b/src/bin/pg_upgrade/pg_upgrade.h
index a710f325de..f6ac78418e 100644
--- a/src/bin/pg_upgrade/pg_upgrade.h
+++ b/src/bin/pg_upgrade/pg_upgrade.h
@@ -160,6 +160,8 @@ typedef struct
bool two_phase; /* can the slot decode 2PC? */
bool caught_up; /* has the slot caught up to latest changes? */
bool invalid; /* if true, the slot is unusable */
+ bool failover; /* is the slot designated to be synced
+ * to the physical standby? */
} LogicalSlotInfo;
typedef struct
diff --git a/src/bin/pg_upgrade/t/003_logical_slots.pl b/src/bin/pg_upgrade/t/003_logical_slots.pl
index 5b01cf8c40..0a1c467ed0 100644
--- a/src/bin/pg_upgrade/t/003_logical_slots.pl
+++ b/src/bin/pg_upgrade/t/003_logical_slots.pl
@@ -158,7 +158,7 @@ $sub->start;
$sub->safe_psql(
'postgres', qq[
CREATE TABLE tbl (a int);
- CREATE SUBSCRIPTION regress_sub CONNECTION '$old_connstr' PUBLICATION regress_pub WITH (two_phase = 'true')
+ CREATE SUBSCRIPTION regress_sub CONNECTION '$old_connstr' PUBLICATION regress_pub WITH (two_phase = 'true', failover = 'true')
]);
$sub->wait_for_subscription_sync($oldpub, 'regress_sub');
@@ -172,8 +172,8 @@ command_ok([@pg_upgrade_cmd], 'run of pg_upgrade of old cluster');
# Check that the slot 'regress_sub' has migrated to the new cluster
$newpub->start;
my $result = $newpub->safe_psql('postgres',
- "SELECT slot_name, two_phase FROM pg_replication_slots");
-is($result, qq(regress_sub|t), 'check the slot exists on new cluster');
+ "SELECT slot_name, two_phase, failover FROM pg_replication_slots");
+is($result, qq(regress_sub|t|t), 'check the slot exists on new cluster');
# Update the connection
my $new_connstr = $newpub->connstr . ' dbname=postgres';
diff --git a/src/bin/psql/describe.c b/src/bin/psql/describe.c
index 5077e7b358..36795b1085 100644
--- a/src/bin/psql/describe.c
+++ b/src/bin/psql/describe.c
@@ -6563,7 +6563,8 @@ describeSubscriptions(const char *pattern, bool verbose)
PGresult *res;
printQueryOpt myopt = pset.popt;
static const bool translate_columns[] = {false, false, false, false,
- false, false, false, false, false, false, false, false, false, false};
+ false, false, false, false, false, false, false, false, false, false,
+ false};
if (pset.sversion < 100000)
{
@@ -6627,6 +6628,11 @@ describeSubscriptions(const char *pattern, bool verbose)
gettext_noop("Password required"),
gettext_noop("Run as owner?"));
+ if (pset.sversion >= 170000)
+ appendPQExpBuffer(&buf,
+ ", subfailoverstate AS \"%s\"\n",
+ gettext_noop("Failover"));
+
appendPQExpBuffer(&buf,
", subsynccommit AS \"%s\"\n"
", subconninfo AS \"%s\"\n",
diff --git a/src/bin/psql/tab-complete.c b/src/bin/psql/tab-complete.c
index 006e10f5d2..46e875c48b 100644
--- a/src/bin/psql/tab-complete.c
+++ b/src/bin/psql/tab-complete.c
@@ -3306,7 +3306,7 @@ psql_completion(const char *text, int start, int end)
/* Complete "CREATE SUBSCRIPTION <name> ... WITH ( <opt>" */
else if (HeadMatches("CREATE", "SUBSCRIPTION") && TailMatches("WITH", "("))
COMPLETE_WITH("binary", "connect", "copy_data", "create_slot",
- "disable_on_error", "enabled", "origin",
+ "disable_on_error", "enabled", "failover", "origin",
"password_required", "run_as_owner", "slot_name",
"streaming", "synchronous_commit", "two_phase");
diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat
index fb58dee3bc..d906734750 100644
--- a/src/include/catalog/pg_proc.dat
+++ b/src/include/catalog/pg_proc.dat
@@ -11100,17 +11100,17 @@
proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f',
proretset => 't', provolatile => 's', prorettype => 'record',
proargtypes => '',
- proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool}',
- proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
- proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting}',
+ proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,bool}',
+ proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}',
+ proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,failover}',
prosrc => 'pg_get_replication_slots' },
{ oid => '3786', descr => 'set up a logical replication slot',
proname => 'pg_create_logical_replication_slot', provolatile => 'v',
proparallel => 'u', prorettype => 'record',
- proargtypes => 'name name bool bool',
- proallargtypes => '{name,name,bool,bool,name,pg_lsn}',
- proargmodes => '{i,i,i,i,o,o}',
- proargnames => '{slot_name,plugin,temporary,twophase,slot_name,lsn}',
+ proargtypes => 'name name bool bool bool',
+ proallargtypes => '{name,name,bool,bool,bool,name,pg_lsn}',
+ proargmodes => '{i,i,i,i,i,o,o}',
+ proargnames => '{slot_name,plugin,temporary,twophase,failover,slot_name,lsn}',
prosrc => 'pg_create_logical_replication_slot' },
{ oid => '4222',
descr => 'copy a logical replication slot, changing temporality and plugin',
diff --git a/src/include/catalog/pg_subscription.h b/src/include/catalog/pg_subscription.h
index e0b91eacd2..3190a3889b 100644
--- a/src/include/catalog/pg_subscription.h
+++ b/src/include/catalog/pg_subscription.h
@@ -31,6 +31,14 @@
#define LOGICALREP_TWOPHASE_STATE_PENDING 'p'
#define LOGICALREP_TWOPHASE_STATE_ENABLED 'e'
+/*
+ * failover tri-state values. See comments atop worker.c to know more about
+ * these states.
+ */
+#define LOGICALREP_FAILOVER_STATE_DISABLED 'd'
+#define LOGICALREP_FAILOVER_STATE_PENDING 'p'
+#define LOGICALREP_FAILOVER_STATE_ENABLED 'e'
+
/*
* The subscription will request the publisher to only send changes that do not
* have any origin.
@@ -93,6 +101,8 @@ CATALOG(pg_subscription,6100,SubscriptionRelationId) BKI_SHARED_RELATION BKI_ROW
bool subrunasowner; /* True if replication should execute as the
* subscription owner */
+ char subfailoverstate; /* Failover state */
+
#ifdef CATALOG_VARLEN /* variable-length fields start here */
/* Connection string to the publisher */
text subconninfo BKI_FORCE_NOT_NULL;
@@ -145,6 +155,7 @@ typedef struct Subscription
List *publications; /* List of publication names to subscribe to */
char *origin; /* Only publish data originating from the
* specified origin */
+ char failoverstate; /* Allow slot to be synchronized for failover */
} Subscription;
/* Disallow streaming in-progress transactions. */
diff --git a/src/include/nodes/replnodes.h b/src/include/nodes/replnodes.h
index 5142a08729..bef8a7162e 100644
--- a/src/include/nodes/replnodes.h
+++ b/src/include/nodes/replnodes.h
@@ -72,6 +72,18 @@ typedef struct DropReplicationSlotCmd
} DropReplicationSlotCmd;
+/* ----------------------
+ * ALTER_REPLICATION_SLOT command
+ * ----------------------
+ */
+typedef struct AlterReplicationSlotCmd
+{
+ NodeTag type;
+ char *slotname;
+ List *options;
+} AlterReplicationSlotCmd;
+
+
/* ----------------------
* START_REPLICATION command
* ----------------------
diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h
index d3535eed58..ca06e5b1ad 100644
--- a/src/include/replication/slot.h
+++ b/src/include/replication/slot.h
@@ -111,6 +111,12 @@ typedef struct ReplicationSlotPersistentData
/* plugin name */
NameData plugin;
+
+ /*
+ * Is this a failover slot (sync candidate for physical standbys)?
+ * Only relevant for logical slots on the primary server.
+ */
+ bool failover;
} ReplicationSlotPersistentData;
/*
@@ -210,6 +216,7 @@ extern PGDLLIMPORT ReplicationSlot *MyReplicationSlot;
/* GUCs */
extern PGDLLIMPORT int max_replication_slots;
+extern PGDLLIMPORT char *standby_slot_names;
/* shmem initialization functions */
extern Size ReplicationSlotsShmemSize(void);
@@ -218,9 +225,10 @@ extern void ReplicationSlotsShmemInit(void);
/* management of individual slots */
extern void ReplicationSlotCreate(const char *name, bool db_specific,
ReplicationSlotPersistency persistency,
- bool two_phase);
+ bool two_phase, bool failover);
extern void ReplicationSlotPersist(void);
extern void ReplicationSlotDrop(const char *name, bool nowait);
+extern void ReplicationSlotAlter(const char *name, bool failover);
extern void ReplicationSlotAcquire(const char *name, bool nowait);
extern void ReplicationSlotRelease(void);
@@ -253,4 +261,6 @@ extern void CheckPointReplicationSlots(bool is_shutdown);
extern void CheckSlotRequirements(void);
extern void CheckSlotPermissions(void);
+extern List *GetStandbySlotList(bool copy);
+
#endif /* SLOT_H */
diff --git a/src/include/replication/walreceiver.h b/src/include/replication/walreceiver.h
index 04b439dc50..61bc8de72c 100644
--- a/src/include/replication/walreceiver.h
+++ b/src/include/replication/walreceiver.h
@@ -356,9 +356,20 @@ typedef char *(*walrcv_create_slot_fn) (WalReceiverConn *conn,
const char *slotname,
bool temporary,
bool two_phase,
+ bool failover,
CRSSnapshotAction snapshot_action,
XLogRecPtr *lsn);
+/*
+ * walrcv_alter_slot_fn
+ *
+ * Change the definition of a replication slot. Currently, it only supports
+ * changing the failover property of the slot.
+ */
+typedef void (*walrcv_alter_slot_fn) (WalReceiverConn *conn,
+ const char *slotname,
+ bool failover);
+
/*
* walrcv_get_backend_pid_fn
*
@@ -400,6 +411,7 @@ typedef struct WalReceiverFunctionsType
walrcv_receive_fn walrcv_receive;
walrcv_send_fn walrcv_send;
walrcv_create_slot_fn walrcv_create_slot;
+ walrcv_alter_slot_fn walrcv_alter_slot;
walrcv_get_backend_pid_fn walrcv_get_backend_pid;
walrcv_exec_fn walrcv_exec;
walrcv_disconnect_fn walrcv_disconnect;
@@ -429,8 +441,10 @@ extern PGDLLIMPORT WalReceiverFunctionsType *WalReceiverFunctions;
WalReceiverFunctions->walrcv_receive(conn, buffer, wait_fd)
#define walrcv_send(conn, buffer, nbytes) \
WalReceiverFunctions->walrcv_send(conn, buffer, nbytes)
-#define walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn) \
- WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, snapshot_action, lsn)
+#define walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn) \
+ WalReceiverFunctions->walrcv_create_slot(conn, slotname, temporary, two_phase, failover, snapshot_action, lsn)
+#define walrcv_alter_slot(conn, slotname, failover) \
+ WalReceiverFunctions->walrcv_alter_slot(conn, slotname, failover)
#define walrcv_get_backend_pid(conn) \
WalReceiverFunctions->walrcv_get_backend_pid(conn)
#define walrcv_exec(conn, exec, nRetTypes, retTypes) \
diff --git a/src/include/replication/walsender.h b/src/include/replication/walsender.h
index 268f8e8d0f..1fcc22a127 100644
--- a/src/include/replication/walsender.h
+++ b/src/include/replication/walsender.h
@@ -14,6 +14,8 @@
#include <signal.h>
+#include "access/xlogdefs.h"
+
/*
* What to do with a snapshot in create replication slot command.
*/
@@ -47,6 +49,8 @@ extern void WalSndInitStopping(void);
extern void WalSndWaitStopping(void);
extern void HandleWalSndInitStopping(void);
extern void WalSndRqstFileReload(void);
+extern void PhysicalWakeupLogicalWalSnd(void);
+extern void WalSndWaitForStandbyConfirmation(XLogRecPtr wait_for_lsn);
/*
* Remember that we want to wakeup walsenders later
diff --git a/src/include/replication/walsender_private.h b/src/include/replication/walsender_private.h
index 13fd5877a6..48c6a7a146 100644
--- a/src/include/replication/walsender_private.h
+++ b/src/include/replication/walsender_private.h
@@ -113,6 +113,13 @@ typedef struct
ConditionVariable wal_flush_cv;
ConditionVariable wal_replay_cv;
+ /*
+ * Used by physical walsenders holding slots specified in
+ * standby_slot_names to wake up logical walsenders holding
+ * failover-enabled slots when a walreceiver confirms the receipt of LSN.
+ */
+ ConditionVariable wal_confirm_rcv_cv;
+
WalSnd walsnds[FLEXIBLE_ARRAY_MEMBER];
} WalSndCtlData;
diff --git a/src/include/replication/worker_internal.h b/src/include/replication/worker_internal.h
index 47854b5cd4..4378690ab0 100644
--- a/src/include/replication/worker_internal.h
+++ b/src/include/replication/worker_internal.h
@@ -258,7 +258,8 @@ extern void ReplicationOriginNameForLogicalRep(Oid suboid, Oid relid,
char *originname, Size szoriginname);
extern bool AllTablesyncsReady(void);
-extern void UpdateTwoPhaseState(Oid suboid, char new_state);
+extern void EnableTwoPhaseFailoverTriState(Oid suboid, bool enable_twophase,
+ bool enable_failover);
extern void process_syncing_tables(XLogRecPtr current_lsn);
extern void invalidate_syncing_table_states(Datum arg, int cacheid,
diff --git a/src/include/utils/guc_hooks.h b/src/include/utils/guc_hooks.h
index 3d74483f44..2f3028cc07 100644
--- a/src/include/utils/guc_hooks.h
+++ b/src/include/utils/guc_hooks.h
@@ -162,5 +162,8 @@ extern bool check_wal_consistency_checking(char **newval, void **extra,
extern void assign_wal_consistency_checking(const char *newval, void *extra);
extern bool check_wal_segment_size(int *newval, void **extra, GucSource source);
extern void assign_wal_sync_method(int new_wal_sync_method, void *extra);
+extern bool check_standby_slot_names(char **newval, void **extra,
+ GucSource source);
+extern void assign_standby_slot_names(const char *newval, void *extra);
#endif /* GUC_HOOKS_H */
diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build
index 9d8039684a..3be3ee52fc 100644
--- a/src/test/recovery/meson.build
+++ b/src/test/recovery/meson.build
@@ -45,6 +45,7 @@ tests += {
't/037_invalid_database.pl',
't/038_save_logical_slots_shutdown.pl',
't/039_end_of_wal.pl',
+ 't/050_verify_slot_order.pl',
],
},
}
diff --git a/src/test/recovery/t/006_logical_decoding.pl b/src/test/recovery/t/006_logical_decoding.pl
index 5025d65b1b..a3c3ee3a14 100644
--- a/src/test/recovery/t/006_logical_decoding.pl
+++ b/src/test/recovery/t/006_logical_decoding.pl
@@ -172,9 +172,10 @@ is($node_primary->slot('otherdb_slot')->{'slot_name'},
undef, 'logical slot was actually dropped with DB');
# Test logical slot advancing and its durability.
+# Pass failover=true (last-arg), it should not have any impact on advancing.
my $logical_slot = 'logical_slot';
$node_primary->safe_psql('postgres',
- "SELECT pg_create_logical_replication_slot('$logical_slot', 'test_decoding', false);"
+ "SELECT pg_create_logical_replication_slot('$logical_slot', 'test_decoding', false, false, true);"
);
$node_primary->psql(
'postgres', "
diff --git a/src/test/recovery/t/050_verify_slot_order.pl b/src/test/recovery/t/050_verify_slot_order.pl
new file mode 100644
index 0000000000..bff0f52a46
--- /dev/null
+++ b/src/test/recovery/t/050_verify_slot_order.pl
@@ -0,0 +1,149 @@
+
+# Copyright (c) 2023, PostgreSQL Global Development Group
+
+use strict;
+use warnings;
+use PostgreSQL::Test::Cluster;
+use PostgreSQL::Test::Utils;
+use Test::More;
+
+# Test primary disallowing specified logical replication slots getting ahead of
+# specified physical replication slots. It uses the following set up:
+#
+# | ----> standby1 (primary_slot_name = sb1_slot)
+# | ----> standby2 (primary_slot_name = sb2_slot)
+# primary ----- |
+# | ----> subscriber1 (failover = true)
+# | ----> subscriber2 (failover = false)
+#
+# standby_slot_names = 'sb1_slot'
+#
+# Set up is configured in such a way that the logical slot of subscriber1 is
+# enabled failover, thus it will wait for the physical slot of
+# standby1(sb1_slot) to catch up before sending decoded changes to subscriber1.
+
+# Create primary
+my $primary = PostgreSQL::Test::Cluster->new('primary');
+$primary->init(allows_streaming => 'logical');
+
+# Configure primary to disallow any logical slots that enabled failover from
+# getting ahead of specified physical replication slot (sb1_slot).
+$primary->append_conf(
+ 'postgresql.conf', qq(
+standby_slot_names = 'sb1_slot'
+));
+$primary->start;
+
+$primary->psql('postgres',
+ q{SELECT pg_create_physical_replication_slot('sb1_slot');});
+$primary->psql('postgres',
+ q{SELECT pg_create_physical_replication_slot('sb2_slot');});
+
+$primary->safe_psql('postgres', "CREATE TABLE tab_int (a int PRIMARY KEY);");
+
+my $backup_name = 'backup';
+$primary->backup($backup_name);
+
+# Create a standby
+my $standby1 = PostgreSQL::Test::Cluster->new('standby1');
+$standby1->init_from_backup(
+ $primary, $backup_name,
+ has_streaming => 1,
+ has_restoring => 1);
+$standby1->append_conf(
+ 'postgresql.conf', qq(
+primary_slot_name = 'sb1_slot'
+));
+$standby1->start;
+$primary->wait_for_replay_catchup($standby1);
+
+# Create another standby
+my $standby2 = PostgreSQL::Test::Cluster->new('standby2');
+$standby2->init_from_backup(
+ $primary, $backup_name,
+ has_streaming => 1,
+ has_restoring => 1);
+$standby2->append_conf(
+ 'postgresql.conf', qq(
+primary_slot_name = 'sb2_slot'
+));
+$standby2->start;
+$primary->wait_for_replay_catchup($standby2);
+
+# Create publication on primary
+my $publisher = $primary;
+$publisher->safe_psql('postgres', "CREATE PUBLICATION regress_mypub FOR TABLE tab_int;");
+my $publisher_connstr = $publisher->connstr . ' dbname=postgres';
+
+# Create a subscriber node, wait for sync to complete
+my $subscriber1 = PostgreSQL::Test::Cluster->new('subscriber1');
+$subscriber1->init(allows_streaming => 'logical');
+$subscriber1->start;
+$subscriber1->safe_psql('postgres', "CREATE TABLE tab_int (a int PRIMARY KEY);");
+
+# Create a subscription with failover = true
+$subscriber1->safe_psql('postgres',
+ "CREATE SUBSCRIPTION regress_mysub1 CONNECTION '$publisher_connstr' "
+ . "PUBLICATION regress_mypub WITH (slot_name = lsub1_slot, failover = true);");
+$subscriber1->wait_for_subscription_sync;
+
+# Create another subscriber node without enabling failover, wait for sync to
+# complete
+my $subscriber2 = PostgreSQL::Test::Cluster->new('subscriber2');
+$subscriber2->init(allows_streaming => 'logical');
+$subscriber2->start;
+$subscriber2->safe_psql('postgres', "CREATE TABLE tab_int (a int PRIMARY KEY);");
+$subscriber2->safe_psql('postgres',
+ "CREATE SUBSCRIPTION mysub2 CONNECTION '$publisher_connstr' "
+ . "PUBLICATION regress_mypub WITH (slot_name = lsub2_slot);");
+$subscriber2->wait_for_subscription_sync;
+
+# Stop the standby associated with specified physical replication slot so that
+# the logical replication slot won't receive changes until the standby comes
+# up.
+$standby1->stop;
+
+# Create some data on primary
+my $primary_row_count = 10;
+my $primary_insert_time = time();
+$primary->safe_psql('postgres',
+ "INSERT INTO tab_int SELECT generate_series(1, $primary_row_count);");
+
+# Wait for the standby that's up and running gets the data from primary
+$primary->wait_for_replay_catchup($standby2);
+my $result = $standby2->safe_psql('postgres',
+ "SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't', "standby2 gets data from primary");
+
+# Wait for the subscription that's up and running and is not enabled for failover.
+# It gets the data from primary without waiting for any standbys.
+$publisher->wait_for_catchup('mysub2');
+$result = $subscriber2->safe_psql('postgres',
+ "SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't', "subscriber2 gets data from primary");
+
+# The subscription that's up and running and is enabled for failover
+# doesn't get the data from primary and keeps waiting for the
+# standby specified in standby_slot_names.
+$result = $subscriber1->safe_psql('postgres',
+ "SELECT count(*) = 0 FROM tab_int;");
+is($result, 't', "subscriber1 doesn't get data from primary until standby1 acknowledges changes");
+
+# Start the standby specified in standby_slot_names and wait for it to catch
+# up with the primary.
+$standby1->start;
+$primary->wait_for_replay_catchup($standby1);
+$result = $standby1->safe_psql('postgres',
+ "SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't', "standby1 gets data from primary");
+
+# Now that the standby specified in standby_slot_names is up and running,
+# primary must send the decoded changes to subscription enabled for failover
+# While the standby was down, this subscriber didn't receive any data from
+# primary i.e. the primary didn't allow it to go ahead of standby.
+$publisher->wait_for_catchup('regress_mysub1');
+$result = $subscriber1->safe_psql('postgres',
+ "SELECT count(*) = $primary_row_count FROM tab_int;");
+is($result, 't', "subscriber1 gets data from primary after standby1 acknowledges changes");
+
+done_testing();
diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out
index 05070393b9..cb3b04aa0c 100644
--- a/src/test/regress/expected/rules.out
+++ b/src/test/regress/expected/rules.out
@@ -1473,8 +1473,9 @@ pg_replication_slots| SELECT l.slot_name,
l.wal_status,
l.safe_wal_size,
l.two_phase,
- l.conflicting
- FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting)
+ l.conflicting,
+ l.failover
+ FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, failover)
LEFT JOIN pg_database d ON ((l.datoid = d.oid)));
pg_roles| SELECT pg_authid.rolname,
pg_authid.rolsuper,
diff --git a/src/test/regress/expected/subscription.out b/src/test/regress/expected/subscription.out
index b15eddbff3..96c614332c 100644
--- a/src/test/regress/expected/subscription.out
+++ b/src/test/regress/expected/subscription.out
@@ -116,18 +116,18 @@ CREATE SUBSCRIPTION regress_testsub4 CONNECTION 'dbname=regress_doesnotexist' PU
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | t | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | none | t | f | d | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub4 SET (origin = any);
\dRs+ regress_testsub4
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
-------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+-----------------------------+----------
- regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub4 | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | d | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub3;
@@ -145,10 +145,10 @@ ALTER SUBSCRIPTION regress_testsub CONNECTION 'foobar';
ERROR: invalid connection string syntax: missing "=" after "foobar" in connection info string
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | d | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET PUBLICATION testpub2, testpub3 WITH (refresh = false);
@@ -157,10 +157,10 @@ ALTER SUBSCRIPTION regress_testsub SET (slot_name = 'newname');
ALTER SUBSCRIPTION regress_testsub SET (password_required = false);
ALTER SUBSCRIPTION regress_testsub SET (run_as_owner = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | t | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | f | t | d | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (password_required = true);
@@ -176,10 +176,10 @@ ERROR: unrecognized subscription parameter: "create_slot"
-- ok
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/12345');
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | off | dbname=regress_doesnotexist2 | 0/12345
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | d | off | dbname=regress_doesnotexist2 | 0/12345
(1 row)
-- ok - with lsn = NONE
@@ -188,10 +188,10 @@ ALTER SUBSCRIPTION regress_testsub SKIP (lsn = NONE);
ALTER SUBSCRIPTION regress_testsub SKIP (lsn = '0/0');
ERROR: invalid WAL location (LSN): 0/0
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+------------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | off | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | d | off | dbname=regress_doesnotexist2 | 0/0
(1 row)
BEGIN;
@@ -223,10 +223,10 @@ ALTER SUBSCRIPTION regress_testsub_foo SET (synchronous_commit = foobar);
ERROR: invalid value for parameter "synchronous_commit": "foobar"
HINT: Available values: local, remote_write, remote_apply, on, off.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
----------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+------------------------------+----------
- regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | local | dbname=regress_doesnotexist2 | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+---------------------+---------------------------+---------+---------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+------------------------------+----------
+ regress_testsub_foo | regress_subscription_user | f | {testpub2,testpub3} | f | off | d | f | any | t | f | d | local | dbname=regress_doesnotexist2 | 0/0
(1 row)
-- rename back to keep the rest simple
@@ -255,19 +255,19 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | t | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | t | off | d | f | any | t | f | d | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (binary = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | d | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -279,27 +279,27 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | d | f | any | t | f | d | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = parallel);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | parallel | d | f | any | t | f | d | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (streaming = false);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | d | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication already exists
@@ -314,10 +314,10 @@ ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refr
ALTER SUBSCRIPTION regress_testsub ADD PUBLICATION testpub1, testpub2 WITH (refresh = false);
ERROR: publication "testpub1" is already in subscription "regress_testsub"
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-----------------------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub,testpub1,testpub2} | f | off | d | f | any | t | f | d | off | dbname=regress_doesnotexist | 0/0
(1 row)
-- fail - publication used more than once
@@ -332,10 +332,10 @@ ERROR: publication "testpub3" is not in subscription "regress_testsub"
-- ok - delete publications
ALTER SUBSCRIPTION regress_testsub DROP PUBLICATION testpub1, testpub2 WITH (refresh = false);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | d | off | dbname=regress_doesnotexist | 0/0
(1 row)
DROP SUBSCRIPTION regress_testsub;
@@ -371,10 +371,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | p | f | any | t | f | d | off | dbname=regress_doesnotexist | 0/0
(1 row)
--fail - alter of two_phase option not supported.
@@ -383,10 +383,10 @@ ERROR: unrecognized subscription parameter: "two_phase"
-- but can alter streaming when two_phase enabled
ALTER SUBSCRIPTION regress_testsub SET (streaming = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | d | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -396,10 +396,10 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | on | p | f | any | t | f | d | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
@@ -412,18 +412,31 @@ CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUB
WARNING: subscription was created, but is not connected
HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | d | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
\dRs+
- List of subscriptions
- Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Synchronous commit | Conninfo | Skip LSN
------------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+--------------------+-----------------------------+----------
- regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | t | f | off | dbname=regress_doesnotexist | 0/0
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | t | any | t | f | d | off | dbname=regress_doesnotexist | 0/0
+(1 row)
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+-- test failover option
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, failover = true);
+WARNING: subscription was created, but is not connected
+HINT: To initiate replication, you must manually create the replication slot, enable the subscription, and refresh the subscription.
+\dRs+
+ List of subscriptions
+ Name | Owner | Enabled | Publication | Binary | Streaming | Two-phase commit | Disable on error | Origin | Password required | Run as owner? | Failover | Synchronous commit | Conninfo | Skip LSN
+-----------------+---------------------------+---------+-------------+--------+-----------+------------------+------------------+--------+-------------------+---------------+----------+--------------------+-----------------------------+----------
+ regress_testsub | regress_subscription_user | f | {testpub} | f | off | d | f | any | t | f | p | off | dbname=regress_doesnotexist | 0/0
(1 row)
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
diff --git a/src/test/regress/sql/subscription.sql b/src/test/regress/sql/subscription.sql
index 444e563ff3..e4601158b3 100644
--- a/src/test/regress/sql/subscription.sql
+++ b/src/test/regress/sql/subscription.sql
@@ -290,6 +290,14 @@ ALTER SUBSCRIPTION regress_testsub SET (disable_on_error = true);
ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
DROP SUBSCRIPTION regress_testsub;
+-- test failover option
+CREATE SUBSCRIPTION regress_testsub CONNECTION 'dbname=regress_doesnotexist' PUBLICATION testpub WITH (connect = false, failover = true);
+
+\dRs+
+
+ALTER SUBSCRIPTION regress_testsub SET (slot_name = NONE);
+DROP SUBSCRIPTION regress_testsub;
+
-- let's do some tests with pg_create_subscription rather than superuser
SET SESSION AUTHORIZATION regress_subscription_user3;
diff --git a/src/tools/pgindent/typedefs.list b/src/tools/pgindent/typedefs.list
index 86a9886d4f..d80d30e99c 100644
--- a/src/tools/pgindent/typedefs.list
+++ b/src/tools/pgindent/typedefs.list
@@ -85,6 +85,7 @@ AlterOwnerStmt
AlterPolicyStmt
AlterPublicationAction
AlterPublicationStmt
+AlterReplicationSlotCmd
AlterRoleSetStmt
AlterRoleStmt
AlterSeqStmt
@@ -3862,6 +3863,7 @@ varattrib_1b_e
varattrib_4b
vbits
verifier_context
+walrcv_alter_slot_fn
walrcv_check_conninfo_fn
walrcv_connect_fn
walrcv_create_slot_fn
--
2.30.0.windows.2
view thread (108+ messages) latest in thread
reply
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: [email protected]
Cc: [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected], [email protected]
Subject: RE: Synchronizing slots from primary to standby
In-Reply-To: <OS0PR01MB57161938A348977127D56A6694BDA@OS0PR01MB5716.jpnprd01.prod.outlook.com>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by agora; see mirroring instructions
for how to clone and mirror all data and code used for this inbox