LCOV - code coverage report
Current view: top level - src/backend/replication/logical - slotsync.c (source / functions) Hit Total Coverage
Test: PostgreSQL 18beta1 Lines: 402 455 88.4 %
Date: 2025-05-11 12:15:22 Functions: 27 27 100.0 %
Legend: Lines: hit not hit

          Line data    Source code
       1             : /*-------------------------------------------------------------------------
       2             :  * slotsync.c
       3             :  *     Functionality for synchronizing slots to a standby server from the
       4             :  *         primary server.
       5             :  *
       6             :  * Copyright (c) 2024-2025, PostgreSQL Global Development Group
       7             :  *
       8             :  * IDENTIFICATION
       9             :  *    src/backend/replication/logical/slotsync.c
      10             :  *
      11             :  * This file contains the code for slot synchronization on a physical standby
      12             :  * to fetch logical failover slots information from the primary server, create
      13             :  * the slots on the standby and synchronize them periodically.
      14             :  *
      15             :  * Slot synchronization can be performed either automatically by enabling slot
      16             :  * sync worker or manually by calling SQL function pg_sync_replication_slots().
      17             :  *
      18             :  * If the WAL corresponding to the remote's restart_lsn is not available on the
      19             :  * physical standby or the remote's catalog_xmin precedes the oldest xid for
      20             :  * which it is guaranteed that rows wouldn't have been removed then we cannot
      21             :  * create the local standby slot because that would mean moving the local slot
      22             :  * backward and decoding won't be possible via such a slot. In this case, the
      23             :  * slot will be marked as RS_TEMPORARY. Once the primary server catches up,
      24             :  * the slot will be marked as RS_PERSISTENT (which means sync-ready) after
      25             :  * which slot sync worker can perform the sync periodically or user can call
      26             :  * pg_sync_replication_slots() periodically to perform the syncs.
      27             :  *
      28             :  * If synchronized slots fail to build a consistent snapshot from the
      29             :  * restart_lsn before reaching confirmed_flush_lsn, they would become
      30             :  * unreliable after promotion due to potential data loss from changes
      31             :  * before reaching a consistent point. This can happen because the slots can
      32             :  * be synced at some random time and we may not reach the consistent point
      33             :  * at the same WAL location as the primary. So, we mark such slots as
      34             :  * RS_TEMPORARY. Once the decoding from corresponding LSNs can reach a
      35             :  * consistent point, they will be marked as RS_PERSISTENT.
      36             :  *
      37             :  * The slot sync worker waits for some time before the next synchronization,
      38             :  * with the duration varying based on whether any slots were updated during
      39             :  * the last cycle. Refer to the comments above wait_for_slot_activity() for
      40             :  * more details.
      41             :  *
      42             :  * Any standby synchronized slots will be dropped if they no longer need
      43             :  * to be synchronized. See comment atop drop_local_obsolete_slots() for more
      44             :  * details.
      45             :  *---------------------------------------------------------------------------
      46             :  */
      47             : 
      48             : #include "postgres.h"
      49             : 
      50             : #include <time.h>
      51             : 
      52             : #include "access/xlog_internal.h"
      53             : #include "access/xlogrecovery.h"
      54             : #include "catalog/pg_database.h"
      55             : #include "commands/dbcommands.h"
      56             : #include "libpq/pqsignal.h"
      57             : #include "pgstat.h"
      58             : #include "postmaster/interrupt.h"
      59             : #include "replication/logical.h"
      60             : #include "replication/slotsync.h"
      61             : #include "replication/snapbuild.h"
      62             : #include "storage/ipc.h"
      63             : #include "storage/lmgr.h"
      64             : #include "storage/proc.h"
      65             : #include "storage/procarray.h"
      66             : #include "tcop/tcopprot.h"
      67             : #include "utils/builtins.h"
      68             : #include "utils/pg_lsn.h"
      69             : #include "utils/ps_status.h"
      70             : #include "utils/timeout.h"
      71             : 
      72             : /*
      73             :  * Struct for sharing information to control slot synchronization.
      74             :  *
      75             :  * The slot sync worker's pid is needed by the startup process to shut it
      76             :  * down during promotion. The startup process shuts down the slot sync worker
      77             :  * and also sets stopSignaled=true to handle the race condition when the
      78             :  * postmaster has not noticed the promotion yet and thus may end up restarting
      79             :  * the slot sync worker. If stopSignaled is set, the worker will exit in such a
      80             :  * case. The SQL function pg_sync_replication_slots() will also error out if
      81             :  * this flag is set. Note that we don't need to reset this variable as after
      82             :  * promotion the slot sync worker won't be restarted because the pmState
      83             :  * changes to PM_RUN from PM_HOT_STANDBY and we don't support demoting
      84             :  * primary without restarting the server. See LaunchMissingBackgroundProcesses.
      85             :  *
      86             :  * The 'syncing' flag is needed to prevent concurrent slot syncs to avoid slot
      87             :  * overwrites.
      88             :  *
      89             :  * The 'last_start_time' is needed by postmaster to start the slot sync worker
      90             :  * once per SLOTSYNC_RESTART_INTERVAL_SEC. In cases where an immediate restart
      91             :  * is expected (e.g., slot sync GUCs change), slot sync worker will reset
      92             :  * last_start_time before exiting, so that postmaster can start the worker
      93             :  * without waiting for SLOTSYNC_RESTART_INTERVAL_SEC.
      94             :  */
      95             : typedef struct SlotSyncCtxStruct
      96             : {
      97             :     pid_t       pid;
      98             :     bool        stopSignaled;
      99             :     bool        syncing;
     100             :     time_t      last_start_time;
     101             :     slock_t     mutex;
     102             : } SlotSyncCtxStruct;
     103             : 
     104             : static SlotSyncCtxStruct *SlotSyncCtx = NULL;
     105             : 
     106             : /* GUC variable */
     107             : bool        sync_replication_slots = false;
     108             : 
     109             : /*
     110             :  * The sleep time (ms) between slot-sync cycles varies dynamically
     111             :  * (within a MIN/MAX range) according to slot activity. See
     112             :  * wait_for_slot_activity() for details.
     113             :  */
     114             : #define MIN_SLOTSYNC_WORKER_NAPTIME_MS  200
     115             : #define MAX_SLOTSYNC_WORKER_NAPTIME_MS  30000   /* 30s */
     116             : 
     117             : static long sleep_ms = MIN_SLOTSYNC_WORKER_NAPTIME_MS;
     118             : 
     119             : /* The restart interval for slot sync work used by postmaster */
     120             : #define SLOTSYNC_RESTART_INTERVAL_SEC 10
     121             : 
     122             : /*
     123             :  * Flag to tell if we are syncing replication slots. Unlike the 'syncing' flag
     124             :  * in SlotSyncCtxStruct, this flag is true only if the current process is
     125             :  * performing slot synchronization.
     126             :  */
     127             : static bool syncing_slots = false;
     128             : 
     129             : /*
     130             :  * Structure to hold information fetched from the primary server about a logical
     131             :  * replication slot.
     132             :  */
     133             : typedef struct RemoteSlot
     134             : {
     135             :     char       *name;
     136             :     char       *plugin;
     137             :     char       *database;
     138             :     bool        two_phase;
     139             :     bool        failover;
     140             :     XLogRecPtr  restart_lsn;
     141             :     XLogRecPtr  confirmed_lsn;
     142             :     XLogRecPtr  two_phase_at;
     143             :     TransactionId catalog_xmin;
     144             : 
     145             :     /* RS_INVAL_NONE if valid, or the reason of invalidation */
     146             :     ReplicationSlotInvalidationCause invalidated;
     147             : } RemoteSlot;
     148             : 
     149             : static void slotsync_failure_callback(int code, Datum arg);
     150             : static void update_synced_slots_inactive_since(void);
     151             : 
     152             : /*
     153             :  * If necessary, update the local synced slot's metadata based on the data
     154             :  * from the remote slot.
     155             :  *
     156             :  * If no update was needed (the data of the remote slot is the same as the
     157             :  * local slot) return false, otherwise true.
     158             :  *
     159             :  * *found_consistent_snapshot will be true iff the remote slot's LSN or xmin is
     160             :  * modified, and decoding from the corresponding LSN's can reach a
     161             :  * consistent snapshot.
     162             :  *
     163             :  * *remote_slot_precedes will be true if the remote slot's LSN or xmin
     164             :  * precedes locally reserved position.
     165             :  */
     166             : static bool
     167          64 : update_local_synced_slot(RemoteSlot *remote_slot, Oid remote_dbid,
     168             :                          bool *found_consistent_snapshot,
     169             :                          bool *remote_slot_precedes)
     170             : {
     171          64 :     ReplicationSlot *slot = MyReplicationSlot;
     172          64 :     bool        updated_xmin_or_lsn = false;
     173          64 :     bool        updated_config = false;
     174             : 
     175             :     Assert(slot->data.invalidated == RS_INVAL_NONE);
     176             : 
     177          64 :     if (found_consistent_snapshot)
     178          10 :         *found_consistent_snapshot = false;
     179             : 
     180          64 :     if (remote_slot_precedes)
     181          10 :         *remote_slot_precedes = false;
     182             : 
     183             :     /*
     184             :      * Don't overwrite if we already have a newer catalog_xmin and
     185             :      * restart_lsn.
     186             :      */
     187         128 :     if (remote_slot->restart_lsn < slot->data.restart_lsn ||
     188          64 :         TransactionIdPrecedes(remote_slot->catalog_xmin,
     189             :                               slot->data.catalog_xmin))
     190             :     {
     191             :         /*
     192             :          * This can happen in following situations:
     193             :          *
     194             :          * If the slot is temporary, it means either the initial WAL location
     195             :          * reserved for the local slot is ahead of the remote slot's
     196             :          * restart_lsn or the initial xmin_horizon computed for the local slot
     197             :          * is ahead of the remote slot.
     198             :          *
     199             :          * If the slot is persistent, both restart_lsn and catalog_xmin of the
     200             :          * synced slot could still be ahead of the remote slot. Since we use
     201             :          * slot advance functionality to keep snapbuild/slot updated, it is
     202             :          * possible that the restart_lsn and catalog_xmin are advanced to a
     203             :          * later position than it has on the primary. This can happen when
     204             :          * slot advancing machinery finds running xacts record after reaching
     205             :          * the consistent state at a later point than the primary where it
     206             :          * serializes the snapshot and updates the restart_lsn.
     207             :          *
     208             :          * We LOG the message if the slot is temporary as it can help the user
     209             :          * to understand why the slot is not sync-ready. In the case of a
     210             :          * persistent slot, it would be a more common case and won't directly
     211             :          * impact the users, so we used DEBUG1 level to log the message.
     212             :          */
     213           0 :         ereport(slot->data.persistency == RS_TEMPORARY ? LOG : DEBUG1,
     214             :                 errmsg("could not synchronize replication slot \"%s\" because remote slot precedes local slot",
     215             :                        remote_slot->name),
     216             :                 errdetail("The remote slot has LSN %X/%X and catalog xmin %u, but the local slot has LSN %X/%X and catalog xmin %u.",
     217             :                           LSN_FORMAT_ARGS(remote_slot->restart_lsn),
     218             :                           remote_slot->catalog_xmin,
     219             :                           LSN_FORMAT_ARGS(slot->data.restart_lsn),
     220             :                           slot->data.catalog_xmin));
     221             : 
     222           0 :         if (remote_slot_precedes)
     223           0 :             *remote_slot_precedes = true;
     224             : 
     225             :         /*
     226             :          * Skip updating the configuration. This is required to avoid syncing
     227             :          * two_phase_at without syncing confirmed_lsn. Otherwise, the prepared
     228             :          * transaction between old confirmed_lsn and two_phase_at will
     229             :          * unexpectedly get decoded and sent to the downstream after
     230             :          * promotion. See comments in ReorderBufferFinishPrepared.
     231             :          */
     232           0 :         return false;
     233             :     }
     234             : 
     235             :     /*
     236             :      * Attempt to sync LSNs and xmins only if remote slot is ahead of local
     237             :      * slot.
     238             :      */
     239          64 :     if (remote_slot->confirmed_lsn > slot->data.confirmed_flush ||
     240          86 :         remote_slot->restart_lsn > slot->data.restart_lsn ||
     241          42 :         TransactionIdFollows(remote_slot->catalog_xmin,
     242             :                              slot->data.catalog_xmin))
     243             :     {
     244             :         /*
     245             :          * We can't directly copy the remote slot's LSN or xmin unless there
     246             :          * exists a consistent snapshot at that point. Otherwise, after
     247             :          * promotion, the slots may not reach a consistent point before the
     248             :          * confirmed_flush_lsn which can lead to a data loss. To avoid data
     249             :          * loss, we let slot machinery advance the slot which ensures that
     250             :          * snapbuilder/slot statuses are updated properly.
     251             :          */
     252          22 :         if (SnapBuildSnapshotExists(remote_slot->restart_lsn))
     253             :         {
     254             :             /*
     255             :              * Update the slot info directly if there is a serialized snapshot
     256             :              * at the restart_lsn, as the slot can quickly reach consistency
     257             :              * at restart_lsn by restoring the snapshot.
     258             :              */
     259           4 :             SpinLockAcquire(&slot->mutex);
     260           4 :             slot->data.restart_lsn = remote_slot->restart_lsn;
     261           4 :             slot->data.confirmed_flush = remote_slot->confirmed_lsn;
     262           4 :             slot->data.catalog_xmin = remote_slot->catalog_xmin;
     263           4 :             SpinLockRelease(&slot->mutex);
     264             : 
     265           4 :             if (found_consistent_snapshot)
     266           0 :                 *found_consistent_snapshot = true;
     267             :         }
     268             :         else
     269             :         {
     270          18 :             LogicalSlotAdvanceAndCheckSnapState(remote_slot->confirmed_lsn,
     271             :                                                 found_consistent_snapshot);
     272             : 
     273             :             /* Sanity check */
     274          18 :             if (slot->data.confirmed_flush != remote_slot->confirmed_lsn)
     275           0 :                 ereport(ERROR,
     276             :                         errmsg_internal("synchronized confirmed_flush for slot \"%s\" differs from remote slot",
     277             :                                         remote_slot->name),
     278             :                         errdetail_internal("Remote slot has LSN %X/%X but local slot has LSN %X/%X.",
     279             :                                            LSN_FORMAT_ARGS(remote_slot->confirmed_lsn),
     280             :                                            LSN_FORMAT_ARGS(slot->data.confirmed_flush)));
     281             :         }
     282             : 
     283          22 :         updated_xmin_or_lsn = true;
     284             :     }
     285             : 
     286          64 :     if (remote_dbid != slot->data.database ||
     287          64 :         remote_slot->two_phase != slot->data.two_phase ||
     288          62 :         remote_slot->failover != slot->data.failover ||
     289          62 :         strcmp(remote_slot->plugin, NameStr(slot->data.plugin)) != 0 ||
     290          62 :         remote_slot->two_phase_at != slot->data.two_phase_at)
     291             :     {
     292             :         NameData    plugin_name;
     293             : 
     294             :         /* Avoid expensive operations while holding a spinlock. */
     295           2 :         namestrcpy(&plugin_name, remote_slot->plugin);
     296             : 
     297           2 :         SpinLockAcquire(&slot->mutex);
     298           2 :         slot->data.plugin = plugin_name;
     299           2 :         slot->data.database = remote_dbid;
     300           2 :         slot->data.two_phase = remote_slot->two_phase;
     301           2 :         slot->data.two_phase_at = remote_slot->two_phase_at;
     302           2 :         slot->data.failover = remote_slot->failover;
     303           2 :         SpinLockRelease(&slot->mutex);
     304             : 
     305           2 :         updated_config = true;
     306             : 
     307             :         /*
     308             :          * Ensure that there is no risk of sending prepared transactions
     309             :          * unexpectedly after the promotion.
     310             :          */
     311             :         Assert(slot->data.two_phase_at <= slot->data.confirmed_flush);
     312             :     }
     313             : 
     314             :     /*
     315             :      * We have to write the changed xmin to disk *before* we change the
     316             :      * in-memory value, otherwise after a crash we wouldn't know that some
     317             :      * catalog tuples might have been removed already.
     318             :      */
     319          64 :     if (updated_config || updated_xmin_or_lsn)
     320             :     {
     321          22 :         ReplicationSlotMarkDirty();
     322          22 :         ReplicationSlotSave();
     323             :     }
     324             : 
     325             :     /*
     326             :      * Now the new xmin is safely on disk, we can let the global value
     327             :      * advance. We do not take ProcArrayLock or similar since we only advance
     328             :      * xmin here and there's not much harm done by a concurrent computation
     329             :      * missing that.
     330             :      */
     331          64 :     if (updated_xmin_or_lsn)
     332             :     {
     333          22 :         SpinLockAcquire(&slot->mutex);
     334          22 :         slot->effective_catalog_xmin = remote_slot->catalog_xmin;
     335          22 :         SpinLockRelease(&slot->mutex);
     336             : 
     337          22 :         ReplicationSlotsComputeRequiredXmin(false);
     338          22 :         ReplicationSlotsComputeRequiredLSN();
     339             :     }
     340             : 
     341          64 :     return updated_config || updated_xmin_or_lsn;
     342             : }
     343             : 
     344             : /*
     345             :  * Get the list of local logical slots that are synchronized from the
     346             :  * primary server.
     347             :  */
     348             : static List *
     349          36 : get_local_synced_slots(void)
     350             : {
     351          36 :     List       *local_slots = NIL;
     352             : 
     353          36 :     LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
     354             : 
     355         396 :     for (int i = 0; i < max_replication_slots; i++)
     356             :     {
     357         360 :         ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
     358             : 
     359             :         /* Check if it is a synchronized slot */
     360         360 :         if (s->in_use && s->data.synced)
     361             :         {
     362             :             Assert(SlotIsLogical(s));
     363          60 :             local_slots = lappend(local_slots, s);
     364             :         }
     365             :     }
     366             : 
     367          36 :     LWLockRelease(ReplicationSlotControlLock);
     368             : 
     369          36 :     return local_slots;
     370             : }
     371             : 
     372             : /*
     373             :  * Helper function to check if local_slot is required to be retained.
     374             :  *
     375             :  * Return false either if local_slot does not exist in the remote_slots list
     376             :  * or is invalidated while the corresponding remote slot is still valid,
     377             :  * otherwise true.
     378             :  */
     379             : static bool
     380          60 : local_sync_slot_required(ReplicationSlot *local_slot, List *remote_slots)
     381             : {
     382          60 :     bool        remote_exists = false;
     383          60 :     bool        locally_invalidated = false;
     384             : 
     385         148 :     foreach_ptr(RemoteSlot, remote_slot, remote_slots)
     386             :     {
     387          86 :         if (strcmp(remote_slot->name, NameStr(local_slot->data.name)) == 0)
     388             :         {
     389          58 :             remote_exists = true;
     390             : 
     391             :             /*
     392             :              * If remote slot is not invalidated but local slot is marked as
     393             :              * invalidated, then set locally_invalidated flag.
     394             :              */
     395          58 :             SpinLockAcquire(&local_slot->mutex);
     396          58 :             locally_invalidated =
     397         116 :                 (remote_slot->invalidated == RS_INVAL_NONE) &&
     398          58 :                 (local_slot->data.invalidated != RS_INVAL_NONE);
     399          58 :             SpinLockRelease(&local_slot->mutex);
     400             : 
     401          58 :             break;
     402             :         }
     403             :     }
     404             : 
     405          60 :     return (remote_exists && !locally_invalidated);
     406             : }
     407             : 
     408             : /*
     409             :  * Drop local obsolete slots.
     410             :  *
     411             :  * Drop the local slots that no longer need to be synced i.e. these either do
     412             :  * not exist on the primary or are no longer enabled for failover.
     413             :  *
     414             :  * Additionally, drop any slots that are valid on the primary but got
     415             :  * invalidated on the standby. This situation may occur due to the following
     416             :  * reasons:
     417             :  * - The 'max_slot_wal_keep_size' on the standby is insufficient to retain WAL
     418             :  *   records from the restart_lsn of the slot.
     419             :  * - 'primary_slot_name' is temporarily reset to null and the physical slot is
     420             :  *   removed.
     421             :  * These dropped slots will get recreated in next sync-cycle and it is okay to
     422             :  * drop and recreate such slots as long as these are not consumable on the
     423             :  * standby (which is the case currently).
     424             :  *
     425             :  * Note: Change of 'wal_level' on the primary server to a level lower than
     426             :  * logical may also result in slot invalidation and removal on the standby.
     427             :  * This is because such 'wal_level' change is only possible if the logical
     428             :  * slots are removed on the primary server, so it's expected to see the
     429             :  * slots being invalidated and removed on the standby too (and re-created
     430             :  * if they are re-created on the primary server).
     431             :  */
     432             : static void
     433          36 : drop_local_obsolete_slots(List *remote_slot_list)
     434             : {
     435          36 :     List       *local_slots = get_local_synced_slots();
     436             : 
     437         132 :     foreach_ptr(ReplicationSlot, local_slot, local_slots)
     438             :     {
     439             :         /* Drop the local slot if it is not required to be retained. */
     440          60 :         if (!local_sync_slot_required(local_slot, remote_slot_list))
     441             :         {
     442             :             bool        synced_slot;
     443             : 
     444             :             /*
     445             :              * Use shared lock to prevent a conflict with
     446             :              * ReplicationSlotsDropDBSlots(), trying to drop the same slot
     447             :              * during a drop-database operation.
     448             :              */
     449           4 :             LockSharedObject(DatabaseRelationId, local_slot->data.database,
     450             :                              0, AccessShareLock);
     451             : 
     452             :             /*
     453             :              * In the small window between getting the slot to drop and
     454             :              * locking the database, there is a possibility of a parallel
     455             :              * database drop by the startup process and the creation of a new
     456             :              * slot by the user. This new user-created slot may end up using
     457             :              * the same shared memory as that of 'local_slot'. Thus check if
     458             :              * local_slot is still the synced one before performing actual
     459             :              * drop.
     460             :              */
     461           4 :             SpinLockAcquire(&local_slot->mutex);
     462           4 :             synced_slot = local_slot->in_use && local_slot->data.synced;
     463           4 :             SpinLockRelease(&local_slot->mutex);
     464             : 
     465           4 :             if (synced_slot)
     466             :             {
     467           4 :                 ReplicationSlotAcquire(NameStr(local_slot->data.name), true, false);
     468           4 :                 ReplicationSlotDropAcquired();
     469             :             }
     470             : 
     471           4 :             UnlockSharedObject(DatabaseRelationId, local_slot->data.database,
     472             :                                0, AccessShareLock);
     473             : 
     474           4 :             ereport(LOG,
     475             :                     errmsg("dropped replication slot \"%s\" of database with OID %u",
     476             :                            NameStr(local_slot->data.name),
     477             :                            local_slot->data.database));
     478             :         }
     479             :     }
     480          36 : }
     481             : 
     482             : /*
     483             :  * Reserve WAL for the currently active local slot using the specified WAL
     484             :  * location (restart_lsn).
     485             :  *
     486             :  * If the given WAL location has been removed, reserve WAL using the oldest
     487             :  * existing WAL segment.
     488             :  */
     489             : static void
     490          10 : reserve_wal_for_local_slot(XLogRecPtr restart_lsn)
     491             : {
     492             :     XLogSegNo   oldest_segno;
     493             :     XLogSegNo   segno;
     494          10 :     ReplicationSlot *slot = MyReplicationSlot;
     495             : 
     496             :     Assert(slot != NULL);
     497             :     Assert(XLogRecPtrIsInvalid(slot->data.restart_lsn));
     498             : 
     499             :     while (true)
     500             :     {
     501          10 :         SpinLockAcquire(&slot->mutex);
     502          10 :         slot->data.restart_lsn = restart_lsn;
     503          10 :         SpinLockRelease(&slot->mutex);
     504             : 
     505             :         /* Prevent WAL removal as fast as possible */
     506          10 :         ReplicationSlotsComputeRequiredLSN();
     507             : 
     508          10 :         XLByteToSeg(slot->data.restart_lsn, segno, wal_segment_size);
     509             : 
     510             :         /*
     511             :          * Find the oldest existing WAL segment file.
     512             :          *
     513             :          * Normally, we can determine it by using the last removed segment
     514             :          * number. However, if no WAL segment files have been removed by a
     515             :          * checkpoint since startup, we need to search for the oldest segment
     516             :          * file from the current timeline existing in XLOGDIR.
     517             :          *
     518             :          * XXX: Currently, we are searching for the oldest segment in the
     519             :          * current timeline as there is less chance of the slot's restart_lsn
     520             :          * from being some prior timeline, and even if it happens, in the
     521             :          * worst case, we will wait to sync till the slot's restart_lsn moved
     522             :          * to the current timeline.
     523             :          */
     524          10 :         oldest_segno = XLogGetLastRemovedSegno() + 1;
     525             : 
     526          10 :         if (oldest_segno == 1)
     527             :         {
     528             :             TimeLineID  cur_timeline;
     529             : 
     530           6 :             GetWalRcvFlushRecPtr(NULL, &cur_timeline);
     531           6 :             oldest_segno = XLogGetOldestSegno(cur_timeline);
     532             :         }
     533             : 
     534          10 :         elog(DEBUG1, "segno: " UINT64_FORMAT " of purposed restart_lsn for the synced slot, oldest_segno: " UINT64_FORMAT " available",
     535             :              segno, oldest_segno);
     536             : 
     537             :         /*
     538             :          * If all required WAL is still there, great, otherwise retry. The
     539             :          * slot should prevent further removal of WAL, unless there's a
     540             :          * concurrent ReplicationSlotsComputeRequiredLSN() after we've written
     541             :          * the new restart_lsn above, so normally we should never need to loop
     542             :          * more than twice.
     543             :          */
     544          10 :         if (segno >= oldest_segno)
     545          10 :             break;
     546             : 
     547             :         /* Retry using the location of the oldest wal segment */
     548           0 :         XLogSegNoOffsetToRecPtr(oldest_segno, 0, wal_segment_size, restart_lsn);
     549             :     }
     550          10 : }
     551             : 
     552             : /*
     553             :  * If the remote restart_lsn and catalog_xmin have caught up with the
     554             :  * local ones, then update the LSNs and persist the local synced slot for
     555             :  * future synchronization; otherwise, do nothing.
     556             :  *
     557             :  * Return true if the slot is marked as RS_PERSISTENT (sync-ready), otherwise
     558             :  * false.
     559             :  */
     560             : static bool
     561          10 : update_and_persist_local_synced_slot(RemoteSlot *remote_slot, Oid remote_dbid)
     562             : {
     563          10 :     ReplicationSlot *slot = MyReplicationSlot;
     564          10 :     bool        found_consistent_snapshot = false;
     565          10 :     bool        remote_slot_precedes = false;
     566             : 
     567          10 :     (void) update_local_synced_slot(remote_slot, remote_dbid,
     568             :                                     &found_consistent_snapshot,
     569             :                                     &remote_slot_precedes);
     570             : 
     571             :     /*
     572             :      * Check if the primary server has caught up. Refer to the comment atop
     573             :      * the file for details on this check.
     574             :      */
     575          10 :     if (remote_slot_precedes)
     576             :     {
     577             :         /*
     578             :          * The remote slot didn't catch up to locally reserved position.
     579             :          *
     580             :          * We do not drop the slot because the restart_lsn can be ahead of the
     581             :          * current location when recreating the slot in the next cycle. It may
     582             :          * take more time to create such a slot. Therefore, we keep this slot
     583             :          * and attempt the synchronization in the next cycle.
     584             :          */
     585           0 :         return false;
     586             :     }
     587             : 
     588             :     /*
     589             :      * Don't persist the slot if it cannot reach the consistent point from the
     590             :      * restart_lsn. See comments atop this file.
     591             :      */
     592          10 :     if (!found_consistent_snapshot)
     593             :     {
     594           0 :         ereport(LOG,
     595             :                 errmsg("could not synchronize replication slot \"%s\"", remote_slot->name),
     596             :                 errdetail("Logical decoding could not find consistent point from local slot's LSN %X/%X.",
     597             :                           LSN_FORMAT_ARGS(slot->data.restart_lsn)));
     598             : 
     599           0 :         return false;
     600             :     }
     601             : 
     602          10 :     ReplicationSlotPersist();
     603             : 
     604          10 :     ereport(LOG,
     605             :             errmsg("newly created replication slot \"%s\" is sync-ready now",
     606             :                    remote_slot->name));
     607             : 
     608          10 :     return true;
     609             : }
     610             : 
     611             : /*
     612             :  * Synchronize a single slot to the given position.
     613             :  *
     614             :  * This creates a new slot if there is no existing one and updates the
     615             :  * metadata of the slot as per the data received from the primary server.
     616             :  *
     617             :  * The slot is created as a temporary slot and stays in the same state until the
     618             :  * remote_slot catches up with locally reserved position and local slot is
     619             :  * updated. The slot is then persisted and is considered as sync-ready for
     620             :  * periodic syncs.
     621             :  *
     622             :  * Returns TRUE if the local slot is updated.
     623             :  */
     624             : static bool
     625          66 : synchronize_one_slot(RemoteSlot *remote_slot, Oid remote_dbid)
     626             : {
     627             :     ReplicationSlot *slot;
     628             :     XLogRecPtr  latestFlushPtr;
     629          66 :     bool        slot_updated = false;
     630             : 
     631             :     /*
     632             :      * Make sure that concerned WAL is received and flushed before syncing
     633             :      * slot to target lsn received from the primary server.
     634             :      */
     635          66 :     latestFlushPtr = GetStandbyFlushRecPtr(NULL);
     636          66 :     if (remote_slot->confirmed_lsn > latestFlushPtr)
     637             :     {
     638             :         /*
     639             :          * Can get here only if GUC 'synchronized_standby_slots' on the
     640             :          * primary server was not configured correctly.
     641             :          */
     642           2 :         ereport(AmLogicalSlotSyncWorkerProcess() ? LOG : ERROR,
     643             :                 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
     644             :                 errmsg("skipping slot synchronization because the received slot sync"
     645             :                        " LSN %X/%X for slot \"%s\" is ahead of the standby position %X/%X",
     646             :                        LSN_FORMAT_ARGS(remote_slot->confirmed_lsn),
     647             :                        remote_slot->name,
     648             :                        LSN_FORMAT_ARGS(latestFlushPtr)));
     649             : 
     650           2 :         return false;
     651             :     }
     652             : 
     653             :     /* Search for the named slot */
     654          64 :     if ((slot = SearchNamedReplicationSlot(remote_slot->name, true)))
     655             :     {
     656             :         bool        synced;
     657             : 
     658          54 :         SpinLockAcquire(&slot->mutex);
     659          54 :         synced = slot->data.synced;
     660          54 :         SpinLockRelease(&slot->mutex);
     661             : 
     662             :         /* User-created slot with the same name exists, raise ERROR. */
     663          54 :         if (!synced)
     664           0 :             ereport(ERROR,
     665             :                     errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
     666             :                     errmsg("exiting from slot synchronization because same"
     667             :                            " name slot \"%s\" already exists on the standby",
     668             :                            remote_slot->name));
     669             : 
     670             :         /*
     671             :          * The slot has been synchronized before.
     672             :          *
     673             :          * It is important to acquire the slot here before checking
     674             :          * invalidation. If we don't acquire the slot first, there could be a
     675             :          * race condition that the local slot could be invalidated just after
     676             :          * checking the 'invalidated' flag here and we could end up
     677             :          * overwriting 'invalidated' flag to remote_slot's value. See
     678             :          * InvalidatePossiblyObsoleteSlot() where it invalidates slot directly
     679             :          * if the slot is not acquired by other processes.
     680             :          *
     681             :          * XXX: If it ever turns out that slot acquire/release is costly for
     682             :          * cases when none of the slot properties is changed then we can do a
     683             :          * pre-check to ensure that at least one of the slot properties is
     684             :          * changed before acquiring the slot.
     685             :          */
     686          54 :         ReplicationSlotAcquire(remote_slot->name, true, false);
     687             : 
     688             :         Assert(slot == MyReplicationSlot);
     689             : 
     690             :         /*
     691             :          * Copy the invalidation cause from remote only if local slot is not
     692             :          * invalidated locally, we don't want to overwrite existing one.
     693             :          */
     694          54 :         if (slot->data.invalidated == RS_INVAL_NONE &&
     695          54 :             remote_slot->invalidated != RS_INVAL_NONE)
     696             :         {
     697           0 :             SpinLockAcquire(&slot->mutex);
     698           0 :             slot->data.invalidated = remote_slot->invalidated;
     699           0 :             SpinLockRelease(&slot->mutex);
     700             : 
     701             :             /* Make sure the invalidated state persists across server restart */
     702           0 :             ReplicationSlotMarkDirty();
     703           0 :             ReplicationSlotSave();
     704             : 
     705           0 :             slot_updated = true;
     706             :         }
     707             : 
     708             :         /* Skip the sync of an invalidated slot */
     709          54 :         if (slot->data.invalidated != RS_INVAL_NONE)
     710             :         {
     711           0 :             ReplicationSlotRelease();
     712           0 :             return slot_updated;
     713             :         }
     714             : 
     715             :         /* Slot not ready yet, let's attempt to make it sync-ready now. */
     716          54 :         if (slot->data.persistency == RS_TEMPORARY)
     717             :         {
     718           0 :             slot_updated = update_and_persist_local_synced_slot(remote_slot,
     719             :                                                                 remote_dbid);
     720             :         }
     721             : 
     722             :         /* Slot ready for sync, so sync it. */
     723             :         else
     724             :         {
     725             :             /*
     726             :              * Sanity check: As long as the invalidations are handled
     727             :              * appropriately as above, this should never happen.
     728             :              *
     729             :              * We don't need to check restart_lsn here. See the comments in
     730             :              * update_local_synced_slot() for details.
     731             :              */
     732          54 :             if (remote_slot->confirmed_lsn < slot->data.confirmed_flush)
     733           0 :                 ereport(ERROR,
     734             :                         errmsg_internal("cannot synchronize local slot \"%s\"",
     735             :                                         remote_slot->name),
     736             :                         errdetail_internal("Local slot's start streaming location LSN(%X/%X) is ahead of remote slot's LSN(%X/%X).",
     737             :                                            LSN_FORMAT_ARGS(slot->data.confirmed_flush),
     738             :                                            LSN_FORMAT_ARGS(remote_slot->confirmed_lsn)));
     739             : 
     740          54 :             slot_updated = update_local_synced_slot(remote_slot, remote_dbid,
     741             :                                                     NULL, NULL);
     742             :         }
     743             :     }
     744             :     /* Otherwise create the slot first. */
     745             :     else
     746             :     {
     747             :         NameData    plugin_name;
     748          10 :         TransactionId xmin_horizon = InvalidTransactionId;
     749             : 
     750             :         /* Skip creating the local slot if remote_slot is invalidated already */
     751          10 :         if (remote_slot->invalidated != RS_INVAL_NONE)
     752           0 :             return false;
     753             : 
     754             :         /*
     755             :          * We create temporary slots instead of ephemeral slots here because
     756             :          * we want the slots to survive after releasing them. This is done to
     757             :          * avoid dropping and re-creating the slots in each synchronization
     758             :          * cycle if the restart_lsn or catalog_xmin of the remote slot has not
     759             :          * caught up.
     760             :          */
     761          10 :         ReplicationSlotCreate(remote_slot->name, true, RS_TEMPORARY,
     762          10 :                               remote_slot->two_phase,
     763          10 :                               remote_slot->failover,
     764             :                               true);
     765             : 
     766             :         /* For shorter lines. */
     767          10 :         slot = MyReplicationSlot;
     768             : 
     769             :         /* Avoid expensive operations while holding a spinlock. */
     770          10 :         namestrcpy(&plugin_name, remote_slot->plugin);
     771             : 
     772          10 :         SpinLockAcquire(&slot->mutex);
     773          10 :         slot->data.database = remote_dbid;
     774          10 :         slot->data.plugin = plugin_name;
     775          10 :         SpinLockRelease(&slot->mutex);
     776             : 
     777          10 :         reserve_wal_for_local_slot(remote_slot->restart_lsn);
     778             : 
     779          10 :         LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE);
     780          10 :         xmin_horizon = GetOldestSafeDecodingTransactionId(true);
     781          10 :         SpinLockAcquire(&slot->mutex);
     782          10 :         slot->effective_catalog_xmin = xmin_horizon;
     783          10 :         slot->data.catalog_xmin = xmin_horizon;
     784          10 :         SpinLockRelease(&slot->mutex);
     785          10 :         ReplicationSlotsComputeRequiredXmin(true);
     786          10 :         LWLockRelease(ProcArrayLock);
     787             : 
     788          10 :         update_and_persist_local_synced_slot(remote_slot, remote_dbid);
     789             : 
     790          10 :         slot_updated = true;
     791             :     }
     792             : 
     793          64 :     ReplicationSlotRelease();
     794             : 
     795          64 :     return slot_updated;
     796             : }
     797             : 
     798             : /*
     799             :  * Synchronize slots.
     800             :  *
     801             :  * Gets the failover logical slots info from the primary server and updates
     802             :  * the slots locally. Creates the slots if not present on the standby.
     803             :  *
     804             :  * Returns TRUE if any of the slots gets updated in this sync-cycle.
     805             :  */
     806             : static bool
     807          36 : synchronize_slots(WalReceiverConn *wrconn)
     808             : {
     809             : #define SLOTSYNC_COLUMN_COUNT 10
     810          36 :     Oid         slotRow[SLOTSYNC_COLUMN_COUNT] = {TEXTOID, TEXTOID, LSNOID,
     811             :     LSNOID, XIDOID, BOOLOID, LSNOID, BOOLOID, TEXTOID, TEXTOID};
     812             : 
     813             :     WalRcvExecResult *res;
     814             :     TupleTableSlot *tupslot;
     815          36 :     List       *remote_slot_list = NIL;
     816          36 :     bool        some_slot_updated = false;
     817          36 :     bool        started_tx = false;
     818          36 :     const char *query = "SELECT slot_name, plugin, confirmed_flush_lsn,"
     819             :         " restart_lsn, catalog_xmin, two_phase, two_phase_at, failover,"
     820             :         " database, invalidation_reason"
     821             :         " FROM pg_catalog.pg_replication_slots"
     822             :         " WHERE failover and NOT temporary";
     823             : 
     824             :     /* The syscache access in walrcv_exec() needs a transaction env. */
     825          36 :     if (!IsTransactionState())
     826             :     {
     827          22 :         StartTransactionCommand();
     828          22 :         started_tx = true;
     829             :     }
     830             : 
     831             :     /* Execute the query */
     832          36 :     res = walrcv_exec(wrconn, query, SLOTSYNC_COLUMN_COUNT, slotRow);
     833          36 :     if (res->status != WALRCV_OK_TUPLES)
     834           0 :         ereport(ERROR,
     835             :                 errmsg("could not fetch failover logical slots info from the primary server: %s",
     836             :                        res->err));
     837             : 
     838             :     /* Construct the remote_slot tuple and synchronize each slot locally */
     839          36 :     tupslot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
     840         102 :     while (tuplestore_gettupleslot(res->tuplestore, true, false, tupslot))
     841             :     {
     842             :         bool        isnull;
     843          66 :         RemoteSlot *remote_slot = palloc0(sizeof(RemoteSlot));
     844             :         Datum       d;
     845          66 :         int         col = 0;
     846             : 
     847          66 :         remote_slot->name = TextDatumGetCString(slot_getattr(tupslot, ++col,
     848             :                                                              &isnull));
     849             :         Assert(!isnull);
     850             : 
     851          66 :         remote_slot->plugin = TextDatumGetCString(slot_getattr(tupslot, ++col,
     852             :                                                                &isnull));
     853             :         Assert(!isnull);
     854             : 
     855             :         /*
     856             :          * It is possible to get null values for LSN and Xmin if slot is
     857             :          * invalidated on the primary server, so handle accordingly.
     858             :          */
     859          66 :         d = slot_getattr(tupslot, ++col, &isnull);
     860          66 :         remote_slot->confirmed_lsn = isnull ? InvalidXLogRecPtr :
     861          66 :             DatumGetLSN(d);
     862             : 
     863          66 :         d = slot_getattr(tupslot, ++col, &isnull);
     864          66 :         remote_slot->restart_lsn = isnull ? InvalidXLogRecPtr : DatumGetLSN(d);
     865             : 
     866          66 :         d = slot_getattr(tupslot, ++col, &isnull);
     867          66 :         remote_slot->catalog_xmin = isnull ? InvalidTransactionId :
     868          66 :             DatumGetTransactionId(d);
     869             : 
     870          66 :         remote_slot->two_phase = DatumGetBool(slot_getattr(tupslot, ++col,
     871             :                                                            &isnull));
     872             :         Assert(!isnull);
     873             : 
     874          66 :         d = slot_getattr(tupslot, ++col, &isnull);
     875          66 :         remote_slot->two_phase_at = isnull ? InvalidXLogRecPtr : DatumGetLSN(d);
     876             : 
     877          66 :         remote_slot->failover = DatumGetBool(slot_getattr(tupslot, ++col,
     878             :                                                           &isnull));
     879             :         Assert(!isnull);
     880             : 
     881          66 :         remote_slot->database = TextDatumGetCString(slot_getattr(tupslot,
     882             :                                                                  ++col, &isnull));
     883             :         Assert(!isnull);
     884             : 
     885          66 :         d = slot_getattr(tupslot, ++col, &isnull);
     886          66 :         remote_slot->invalidated = isnull ? RS_INVAL_NONE :
     887           0 :             GetSlotInvalidationCause(TextDatumGetCString(d));
     888             : 
     889             :         /* Sanity check */
     890             :         Assert(col == SLOTSYNC_COLUMN_COUNT);
     891             : 
     892             :         /*
     893             :          * If restart_lsn, confirmed_lsn or catalog_xmin is invalid but the
     894             :          * slot is valid, that means we have fetched the remote_slot in its
     895             :          * RS_EPHEMERAL state. In such a case, don't sync it; we can always
     896             :          * sync it in the next sync cycle when the remote_slot is persisted
     897             :          * and has valid lsn(s) and xmin values.
     898             :          *
     899             :          * XXX: In future, if we plan to expose 'slot->data.persistency' in
     900             :          * pg_replication_slots view, then we can avoid fetching RS_EPHEMERAL
     901             :          * slots in the first place.
     902             :          */
     903          66 :         if ((XLogRecPtrIsInvalid(remote_slot->restart_lsn) ||
     904          66 :              XLogRecPtrIsInvalid(remote_slot->confirmed_lsn) ||
     905          66 :              !TransactionIdIsValid(remote_slot->catalog_xmin)) &&
     906           0 :             remote_slot->invalidated == RS_INVAL_NONE)
     907           0 :             pfree(remote_slot);
     908             :         else
     909             :             /* Create list of remote slots */
     910          66 :             remote_slot_list = lappend(remote_slot_list, remote_slot);
     911             : 
     912          66 :         ExecClearTuple(tupslot);
     913             :     }
     914             : 
     915             :     /* Drop local slots that no longer need to be synced. */
     916          36 :     drop_local_obsolete_slots(remote_slot_list);
     917             : 
     918             :     /* Now sync the slots locally */
     919         138 :     foreach_ptr(RemoteSlot, remote_slot, remote_slot_list)
     920             :     {
     921          66 :         Oid         remote_dbid = get_database_oid(remote_slot->database, false);
     922             : 
     923             :         /*
     924             :          * Use shared lock to prevent a conflict with
     925             :          * ReplicationSlotsDropDBSlots(), trying to drop the same slot during
     926             :          * a drop-database operation.
     927             :          */
     928          66 :         LockSharedObject(DatabaseRelationId, remote_dbid, 0, AccessShareLock);
     929             : 
     930          66 :         some_slot_updated |= synchronize_one_slot(remote_slot, remote_dbid);
     931             : 
     932          66 :         UnlockSharedObject(DatabaseRelationId, remote_dbid, 0, AccessShareLock);
     933             :     }
     934             : 
     935             :     /* We are done, free remote_slot_list elements */
     936          36 :     list_free_deep(remote_slot_list);
     937             : 
     938          36 :     walrcv_clear_result(res);
     939             : 
     940          36 :     if (started_tx)
     941          22 :         CommitTransactionCommand();
     942             : 
     943          36 :     return some_slot_updated;
     944             : }
     945             : 
     946             : /*
     947             :  * Checks the remote server info.
     948             :  *
     949             :  * We ensure that the 'primary_slot_name' exists on the remote server and the
     950             :  * remote server is not a standby node.
     951             :  */
     952             : static void
     953          24 : validate_remote_info(WalReceiverConn *wrconn)
     954             : {
     955             : #define PRIMARY_INFO_OUTPUT_COL_COUNT 2
     956             :     WalRcvExecResult *res;
     957          24 :     Oid         slotRow[PRIMARY_INFO_OUTPUT_COL_COUNT] = {BOOLOID, BOOLOID};
     958             :     StringInfoData cmd;
     959             :     bool        isnull;
     960             :     TupleTableSlot *tupslot;
     961             :     bool        remote_in_recovery;
     962             :     bool        primary_slot_valid;
     963          24 :     bool        started_tx = false;
     964             : 
     965          24 :     initStringInfo(&cmd);
     966          24 :     appendStringInfo(&cmd,
     967             :                      "SELECT pg_is_in_recovery(), count(*) = 1"
     968             :                      " FROM pg_catalog.pg_replication_slots"
     969             :                      " WHERE slot_type='physical' AND slot_name=%s",
     970             :                      quote_literal_cstr(PrimarySlotName));
     971             : 
     972             :     /* The syscache access in walrcv_exec() needs a transaction env. */
     973          24 :     if (!IsTransactionState())
     974             :     {
     975           8 :         StartTransactionCommand();
     976           8 :         started_tx = true;
     977             :     }
     978             : 
     979          24 :     res = walrcv_exec(wrconn, cmd.data, PRIMARY_INFO_OUTPUT_COL_COUNT, slotRow);
     980          24 :     pfree(cmd.data);
     981             : 
     982          24 :     if (res->status != WALRCV_OK_TUPLES)
     983           0 :         ereport(ERROR,
     984             :                 errmsg("could not fetch primary slot name \"%s\" info from the primary server: %s",
     985             :                        PrimarySlotName, res->err),
     986             :                 errhint("Check if \"primary_slot_name\" is configured correctly."));
     987             : 
     988          24 :     tupslot = MakeSingleTupleTableSlot(res->tupledesc, &TTSOpsMinimalTuple);
     989          24 :     if (!tuplestore_gettupleslot(res->tuplestore, true, false, tupslot))
     990           0 :         elog(ERROR,
     991             :              "failed to fetch tuple for the primary server slot specified by \"primary_slot_name\"");
     992             : 
     993          24 :     remote_in_recovery = DatumGetBool(slot_getattr(tupslot, 1, &isnull));
     994             :     Assert(!isnull);
     995             : 
     996             :     /*
     997             :      * Slot sync is currently not supported on a cascading standby. This is
     998             :      * because if we allow it, the primary server needs to wait for all the
     999             :      * cascading standbys, otherwise, logical subscribers can still be ahead
    1000             :      * of one of the cascading standbys which we plan to promote. Thus, to
    1001             :      * avoid this additional complexity, we restrict it for the time being.
    1002             :      */
    1003          24 :     if (remote_in_recovery)
    1004           2 :         ereport(ERROR,
    1005             :                 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
    1006             :                 errmsg("cannot synchronize replication slots from a standby server"));
    1007             : 
    1008          22 :     primary_slot_valid = DatumGetBool(slot_getattr(tupslot, 2, &isnull));
    1009             :     Assert(!isnull);
    1010             : 
    1011          22 :     if (!primary_slot_valid)
    1012           0 :         ereport(ERROR,
    1013             :                 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
    1014             :         /* translator: second %s is a GUC variable name */
    1015             :                 errmsg("replication slot \"%s\" specified by \"%s\" does not exist on primary server",
    1016             :                        PrimarySlotName, "primary_slot_name"));
    1017             : 
    1018          22 :     ExecClearTuple(tupslot);
    1019          22 :     walrcv_clear_result(res);
    1020             : 
    1021          22 :     if (started_tx)
    1022           8 :         CommitTransactionCommand();
    1023          22 : }
    1024             : 
    1025             : /*
    1026             :  * Checks if dbname is specified in 'primary_conninfo'.
    1027             :  *
    1028             :  * Error out if not specified otherwise return it.
    1029             :  */
    1030             : char *
    1031          26 : CheckAndGetDbnameFromConninfo(void)
    1032             : {
    1033             :     char       *dbname;
    1034             : 
    1035             :     /*
    1036             :      * The slot synchronization needs a database connection for walrcv_exec to
    1037             :      * work.
    1038             :      */
    1039          26 :     dbname = walrcv_get_dbname_from_conninfo(PrimaryConnInfo);
    1040          26 :     if (dbname == NULL)
    1041           2 :         ereport(ERROR,
    1042             :                 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
    1043             : 
    1044             :         /*
    1045             :          * translator: first %s is a connection option; second %s is a GUC
    1046             :          * variable name
    1047             :          */
    1048             :                 errmsg("replication slot synchronization requires \"%s\" to be specified in \"%s\"",
    1049             :                        "dbname", "primary_conninfo"));
    1050          24 :     return dbname;
    1051             : }
    1052             : 
    1053             : /*
    1054             :  * Return true if all necessary GUCs for slot synchronization are set
    1055             :  * appropriately, otherwise, return false.
    1056             :  */
    1057             : bool
    1058          30 : ValidateSlotSyncParams(int elevel)
    1059             : {
    1060             :     /*
    1061             :      * Logical slot sync/creation requires wal_level >= logical.
    1062             :      *
    1063             :      * Since altering the wal_level requires a server restart, so error out in
    1064             :      * this case regardless of elevel provided by caller.
    1065             :      */
    1066          30 :     if (wal_level < WAL_LEVEL_LOGICAL)
    1067           0 :         ereport(ERROR,
    1068             :                 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
    1069             :                 errmsg("replication slot synchronization requires \"wal_level\" >= \"logical\""));
    1070             : 
    1071             :     /*
    1072             :      * A physical replication slot(primary_slot_name) is required on the
    1073             :      * primary to ensure that the rows needed by the standby are not removed
    1074             :      * after restarting, so that the synchronized slot on the standby will not
    1075             :      * be invalidated.
    1076             :      */
    1077          30 :     if (PrimarySlotName == NULL || *PrimarySlotName == '\0')
    1078             :     {
    1079           0 :         ereport(elevel,
    1080             :                 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
    1081             :         /* translator: %s is a GUC variable name */
    1082             :                 errmsg("replication slot synchronization requires \"%s\" to be set", "primary_slot_name"));
    1083           0 :         return false;
    1084             :     }
    1085             : 
    1086             :     /*
    1087             :      * hot_standby_feedback must be enabled to cooperate with the physical
    1088             :      * replication slot, which allows informing the primary about the xmin and
    1089             :      * catalog_xmin values on the standby.
    1090             :      */
    1091          30 :     if (!hot_standby_feedback)
    1092             :     {
    1093           2 :         ereport(elevel,
    1094             :                 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
    1095             :         /* translator: %s is a GUC variable name */
    1096             :                 errmsg("replication slot synchronization requires \"%s\" to be enabled",
    1097             :                        "hot_standby_feedback"));
    1098           2 :         return false;
    1099             :     }
    1100             : 
    1101             :     /*
    1102             :      * The primary_conninfo is required to make connection to primary for
    1103             :      * getting slots information.
    1104             :      */
    1105          28 :     if (PrimaryConnInfo == NULL || *PrimaryConnInfo == '\0')
    1106             :     {
    1107           0 :         ereport(elevel,
    1108             :                 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
    1109             :         /* translator: %s is a GUC variable name */
    1110             :                 errmsg("replication slot synchronization requires \"%s\" to be set",
    1111             :                        "primary_conninfo"));
    1112           0 :         return false;
    1113             :     }
    1114             : 
    1115          28 :     return true;
    1116             : }
    1117             : 
    1118             : /*
    1119             :  * Re-read the config file.
    1120             :  *
    1121             :  * Exit if any of the slot sync GUCs have changed. The postmaster will
    1122             :  * restart it.
    1123             :  */
    1124             : static void
    1125           2 : slotsync_reread_config(void)
    1126             : {
    1127           2 :     char       *old_primary_conninfo = pstrdup(PrimaryConnInfo);
    1128           2 :     char       *old_primary_slotname = pstrdup(PrimarySlotName);
    1129           2 :     bool        old_sync_replication_slots = sync_replication_slots;
    1130           2 :     bool        old_hot_standby_feedback = hot_standby_feedback;
    1131             :     bool        conninfo_changed;
    1132             :     bool        primary_slotname_changed;
    1133             : 
    1134             :     Assert(sync_replication_slots);
    1135             : 
    1136           2 :     ConfigReloadPending = false;
    1137           2 :     ProcessConfigFile(PGC_SIGHUP);
    1138             : 
    1139           2 :     conninfo_changed = strcmp(old_primary_conninfo, PrimaryConnInfo) != 0;
    1140           2 :     primary_slotname_changed = strcmp(old_primary_slotname, PrimarySlotName) != 0;
    1141           2 :     pfree(old_primary_conninfo);
    1142           2 :     pfree(old_primary_slotname);
    1143             : 
    1144           2 :     if (old_sync_replication_slots != sync_replication_slots)
    1145             :     {
    1146           0 :         ereport(LOG,
    1147             :         /* translator: %s is a GUC variable name */
    1148             :                 errmsg("replication slot synchronization worker will shut down because \"%s\" is disabled", "sync_replication_slots"));
    1149           0 :         proc_exit(0);
    1150             :     }
    1151             : 
    1152           2 :     if (conninfo_changed ||
    1153           2 :         primary_slotname_changed ||
    1154           2 :         (old_hot_standby_feedback != hot_standby_feedback))
    1155             :     {
    1156           2 :         ereport(LOG,
    1157             :                 errmsg("replication slot synchronization worker will restart because of a parameter change"));
    1158             : 
    1159             :         /*
    1160             :          * Reset the last-start time for this worker so that the postmaster
    1161             :          * can restart it without waiting for SLOTSYNC_RESTART_INTERVAL_SEC.
    1162             :          */
    1163           2 :         SlotSyncCtx->last_start_time = 0;
    1164             : 
    1165           2 :         proc_exit(0);
    1166             :     }
    1167             : 
    1168           0 : }
    1169             : 
    1170             : /*
    1171             :  * Interrupt handler for main loop of slot sync worker.
    1172             :  */
    1173             : static void
    1174          30 : ProcessSlotSyncInterrupts(WalReceiverConn *wrconn)
    1175             : {
    1176          30 :     CHECK_FOR_INTERRUPTS();
    1177             : 
    1178          26 :     if (ShutdownRequestPending)
    1179             :     {
    1180           2 :         ereport(LOG,
    1181             :                 errmsg("replication slot synchronization worker is shutting down on receiving SIGINT"));
    1182             : 
    1183           2 :         proc_exit(0);
    1184             :     }
    1185             : 
    1186          24 :     if (ConfigReloadPending)
    1187           2 :         slotsync_reread_config();
    1188          22 : }
    1189             : 
    1190             : /*
    1191             :  * Connection cleanup function for slotsync worker.
    1192             :  *
    1193             :  * Called on slotsync worker exit.
    1194             :  */
    1195             : static void
    1196           8 : slotsync_worker_disconnect(int code, Datum arg)
    1197             : {
    1198           8 :     WalReceiverConn *wrconn = (WalReceiverConn *) DatumGetPointer(arg);
    1199             : 
    1200           8 :     walrcv_disconnect(wrconn);
    1201           8 : }
    1202             : 
    1203             : /*
    1204             :  * Cleanup function for slotsync worker.
    1205             :  *
    1206             :  * Called on slotsync worker exit.
    1207             :  */
    1208             : static void
    1209           8 : slotsync_worker_onexit(int code, Datum arg)
    1210             : {
    1211             :     /*
    1212             :      * We need to do slots cleanup here just like WalSndErrorCleanup() does.
    1213             :      *
    1214             :      * The startup process during promotion invokes ShutDownSlotSync() which
    1215             :      * waits for slot sync to finish and it does that by checking the
    1216             :      * 'syncing' flag. Thus the slot sync worker must be done with slots'
    1217             :      * release and cleanup to avoid any dangling temporary slots or active
    1218             :      * slots before it marks itself as finished syncing.
    1219             :      */
    1220             : 
    1221             :     /* Make sure active replication slots are released */
    1222           8 :     if (MyReplicationSlot != NULL)
    1223           0 :         ReplicationSlotRelease();
    1224             : 
    1225             :     /* Also cleanup the temporary slots. */
    1226           8 :     ReplicationSlotCleanup(false);
    1227             : 
    1228           8 :     SpinLockAcquire(&SlotSyncCtx->mutex);
    1229             : 
    1230           8 :     SlotSyncCtx->pid = InvalidPid;
    1231             : 
    1232             :     /*
    1233             :      * If syncing_slots is true, it indicates that the process errored out
    1234             :      * without resetting the flag. So, we need to clean up shared memory and
    1235             :      * reset the flag here.
    1236             :      */
    1237           8 :     if (syncing_slots)
    1238             :     {
    1239           8 :         SlotSyncCtx->syncing = false;
    1240           8 :         syncing_slots = false;
    1241             :     }
    1242             : 
    1243           8 :     SpinLockRelease(&SlotSyncCtx->mutex);
    1244           8 : }
    1245             : 
    1246             : /*
    1247             :  * Sleep for long enough that we believe it's likely that the slots on primary
    1248             :  * get updated.
    1249             :  *
    1250             :  * If there is no slot activity the wait time between sync-cycles will double
    1251             :  * (to a maximum of 30s). If there is some slot activity the wait time between
    1252             :  * sync-cycles is reset to the minimum (200ms).
    1253             :  */
    1254             : static void
    1255          22 : wait_for_slot_activity(bool some_slot_updated)
    1256             : {
    1257             :     int         rc;
    1258             : 
    1259          22 :     if (!some_slot_updated)
    1260             :     {
    1261             :         /*
    1262             :          * No slots were updated, so double the sleep time, but not beyond the
    1263             :          * maximum allowable value.
    1264             :          */
    1265          12 :         sleep_ms = Min(sleep_ms * 2, MAX_SLOTSYNC_WORKER_NAPTIME_MS);
    1266             :     }
    1267             :     else
    1268             :     {
    1269             :         /*
    1270             :          * Some slots were updated since the last sleep, so reset the sleep
    1271             :          * time.
    1272             :          */
    1273          10 :         sleep_ms = MIN_SLOTSYNC_WORKER_NAPTIME_MS;
    1274             :     }
    1275             : 
    1276          22 :     rc = WaitLatch(MyLatch,
    1277             :                    WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
    1278             :                    sleep_ms,
    1279             :                    WAIT_EVENT_REPLICATION_SLOTSYNC_MAIN);
    1280             : 
    1281          22 :     if (rc & WL_LATCH_SET)
    1282           8 :         ResetLatch(MyLatch);
    1283          22 : }
    1284             : 
    1285             : /*
    1286             :  * Emit an error if a promotion or a concurrent sync call is in progress.
    1287             :  * Otherwise, advertise that a sync is in progress.
    1288             :  */
    1289             : static void
    1290          24 : check_and_set_sync_info(pid_t worker_pid)
    1291             : {
    1292          24 :     SpinLockAcquire(&SlotSyncCtx->mutex);
    1293             : 
    1294             :     /* The worker pid must not be already assigned in SlotSyncCtx */
    1295             :     Assert(worker_pid == InvalidPid || SlotSyncCtx->pid == InvalidPid);
    1296             : 
    1297             :     /*
    1298             :      * Emit an error if startup process signaled the slot sync machinery to
    1299             :      * stop. See comments atop SlotSyncCtxStruct.
    1300             :      */
    1301          24 :     if (SlotSyncCtx->stopSignaled)
    1302             :     {
    1303           0 :         SpinLockRelease(&SlotSyncCtx->mutex);
    1304           0 :         ereport(ERROR,
    1305             :                 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    1306             :                 errmsg("cannot synchronize replication slots when standby promotion is ongoing"));
    1307             :     }
    1308             : 
    1309          24 :     if (SlotSyncCtx->syncing)
    1310             :     {
    1311           0 :         SpinLockRelease(&SlotSyncCtx->mutex);
    1312           0 :         ereport(ERROR,
    1313             :                 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
    1314             :                 errmsg("cannot synchronize replication slots concurrently"));
    1315             :     }
    1316             : 
    1317          24 :     SlotSyncCtx->syncing = true;
    1318             : 
    1319             :     /*
    1320             :      * Advertise the required PID so that the startup process can kill the
    1321             :      * slot sync worker on promotion.
    1322             :      */
    1323          24 :     SlotSyncCtx->pid = worker_pid;
    1324             : 
    1325          24 :     SpinLockRelease(&SlotSyncCtx->mutex);
    1326             : 
    1327          24 :     syncing_slots = true;
    1328          24 : }
    1329             : 
    1330             : /*
    1331             :  * Reset syncing flag.
    1332             :  */
    1333             : static void
    1334          16 : reset_syncing_flag()
    1335             : {
    1336          16 :     SpinLockAcquire(&SlotSyncCtx->mutex);
    1337          16 :     SlotSyncCtx->syncing = false;
    1338          16 :     SpinLockRelease(&SlotSyncCtx->mutex);
    1339             : 
    1340          16 :     syncing_slots = false;
    1341          16 : };
    1342             : 
    1343             : /*
    1344             :  * The main loop of our worker process.
    1345             :  *
    1346             :  * It connects to the primary server, fetches logical failover slots
    1347             :  * information periodically in order to create and sync the slots.
    1348             :  */
    1349             : void
    1350           8 : ReplSlotSyncWorkerMain(const void *startup_data, size_t startup_data_len)
    1351             : {
    1352           8 :     WalReceiverConn *wrconn = NULL;
    1353             :     char       *dbname;
    1354             :     char       *err;
    1355             :     sigjmp_buf  local_sigjmp_buf;
    1356             :     StringInfoData app_name;
    1357             : 
    1358             :     Assert(startup_data_len == 0);
    1359             : 
    1360           8 :     MyBackendType = B_SLOTSYNC_WORKER;
    1361             : 
    1362           8 :     init_ps_display(NULL);
    1363             : 
    1364             :     Assert(GetProcessingMode() == InitProcessing);
    1365             : 
    1366             :     /*
    1367             :      * Create a per-backend PGPROC struct in shared memory.  We must do this
    1368             :      * before we access any shared memory.
    1369             :      */
    1370           8 :     InitProcess();
    1371             : 
    1372             :     /*
    1373             :      * Early initialization.
    1374             :      */
    1375           8 :     BaseInit();
    1376             : 
    1377             :     Assert(SlotSyncCtx != NULL);
    1378             : 
    1379             :     /*
    1380             :      * If an exception is encountered, processing resumes here.
    1381             :      *
    1382             :      * We just need to clean up, report the error, and go away.
    1383             :      *
    1384             :      * If we do not have this handling here, then since this worker process
    1385             :      * operates at the bottom of the exception stack, ERRORs turn into FATALs.
    1386             :      * Therefore, we create our own exception handler to catch ERRORs.
    1387             :      */
    1388           8 :     if (sigsetjmp(local_sigjmp_buf, 1) != 0)
    1389             :     {
    1390             :         /* since not using PG_TRY, must reset error stack by hand */
    1391           0 :         error_context_stack = NULL;
    1392             : 
    1393             :         /* Prevents interrupts while cleaning up */
    1394           0 :         HOLD_INTERRUPTS();
    1395             : 
    1396             :         /* Report the error to the server log */
    1397           0 :         EmitErrorReport();
    1398             : 
    1399             :         /*
    1400             :          * We can now go away.  Note that because we called InitProcess, a
    1401             :          * callback was registered to do ProcKill, which will clean up
    1402             :          * necessary state.
    1403             :          */
    1404           0 :         proc_exit(0);
    1405             :     }
    1406             : 
    1407             :     /* We can now handle ereport(ERROR) */
    1408           8 :     PG_exception_stack = &local_sigjmp_buf;
    1409             : 
    1410             :     /* Setup signal handling */
    1411           8 :     pqsignal(SIGHUP, SignalHandlerForConfigReload);
    1412           8 :     pqsignal(SIGINT, SignalHandlerForShutdownRequest);
    1413           8 :     pqsignal(SIGTERM, die);
    1414           8 :     pqsignal(SIGFPE, FloatExceptionHandler);
    1415           8 :     pqsignal(SIGUSR1, procsignal_sigusr1_handler);
    1416           8 :     pqsignal(SIGUSR2, SIG_IGN);
    1417           8 :     pqsignal(SIGPIPE, SIG_IGN);
    1418           8 :     pqsignal(SIGCHLD, SIG_DFL);
    1419             : 
    1420           8 :     check_and_set_sync_info(MyProcPid);
    1421             : 
    1422           8 :     ereport(LOG, errmsg("slot sync worker started"));
    1423             : 
    1424             :     /* Register it as soon as SlotSyncCtx->pid is initialized. */
    1425           8 :     before_shmem_exit(slotsync_worker_onexit, (Datum) 0);
    1426             : 
    1427             :     /*
    1428             :      * Establishes SIGALRM handler and initialize timeout module. It is needed
    1429             :      * by InitPostgres to register different timeouts.
    1430             :      */
    1431           8 :     InitializeTimeouts();
    1432             : 
    1433             :     /* Load the libpq-specific functions */
    1434           8 :     load_file("libpqwalreceiver", false);
    1435             : 
    1436             :     /*
    1437             :      * Unblock signals (they were blocked when the postmaster forked us)
    1438             :      */
    1439           8 :     sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
    1440             : 
    1441             :     /*
    1442             :      * Set always-secure search path, so malicious users can't redirect user
    1443             :      * code (e.g. operators).
    1444             :      *
    1445             :      * It's not strictly necessary since we won't be scanning or writing to
    1446             :      * any user table locally, but it's good to retain it here for added
    1447             :      * precaution.
    1448             :      */
    1449           8 :     SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
    1450             : 
    1451           8 :     dbname = CheckAndGetDbnameFromConninfo();
    1452             : 
    1453             :     /*
    1454             :      * Connect to the database specified by the user in primary_conninfo. We
    1455             :      * need a database connection for walrcv_exec to work which we use to
    1456             :      * fetch slot information from the remote node. See comments atop
    1457             :      * libpqrcv_exec.
    1458             :      *
    1459             :      * We do not specify a specific user here since the slot sync worker will
    1460             :      * operate as a superuser. This is safe because the slot sync worker does
    1461             :      * not interact with user tables, eliminating the risk of executing
    1462             :      * arbitrary code within triggers.
    1463             :      */
    1464           8 :     InitPostgres(dbname, InvalidOid, NULL, InvalidOid, 0, NULL);
    1465             : 
    1466           8 :     SetProcessingMode(NormalProcessing);
    1467             : 
    1468           8 :     initStringInfo(&app_name);
    1469           8 :     if (cluster_name[0])
    1470           8 :         appendStringInfo(&app_name, "%s_%s", cluster_name, "slotsync worker");
    1471             :     else
    1472           0 :         appendStringInfoString(&app_name, "slotsync worker");
    1473             : 
    1474             :     /*
    1475             :      * Establish the connection to the primary server for slot
    1476             :      * synchronization.
    1477             :      */
    1478           8 :     wrconn = walrcv_connect(PrimaryConnInfo, false, false, false,
    1479             :                             app_name.data, &err);
    1480           8 :     pfree(app_name.data);
    1481             : 
    1482           8 :     if (!wrconn)
    1483           0 :         ereport(ERROR,
    1484             :                 errcode(ERRCODE_CONNECTION_FAILURE),
    1485             :                 errmsg("synchronization worker \"%s\" could not connect to the primary server: %s",
    1486             :                        app_name.data, err));
    1487             : 
    1488             :     /*
    1489             :      * Register the disconnection callback.
    1490             :      *
    1491             :      * XXX: This can be combined with previous cleanup registration of
    1492             :      * slotsync_worker_onexit() but that will need the connection to be made
    1493             :      * global and we want to avoid introducing global for this purpose.
    1494             :      */
    1495           8 :     before_shmem_exit(slotsync_worker_disconnect, PointerGetDatum(wrconn));
    1496             : 
    1497             :     /*
    1498             :      * Using the specified primary server connection, check that we are not a
    1499             :      * cascading standby and slot configured in 'primary_slot_name' exists on
    1500             :      * the primary server.
    1501             :      */
    1502           8 :     validate_remote_info(wrconn);
    1503             : 
    1504             :     /* Main loop to synchronize slots */
    1505             :     for (;;)
    1506          22 :     {
    1507          30 :         bool        some_slot_updated = false;
    1508             : 
    1509          30 :         ProcessSlotSyncInterrupts(wrconn);
    1510             : 
    1511          22 :         some_slot_updated = synchronize_slots(wrconn);
    1512             : 
    1513          22 :         wait_for_slot_activity(some_slot_updated);
    1514             :     }
    1515             : 
    1516             :     /*
    1517             :      * The slot sync worker can't get here because it will only stop when it
    1518             :      * receives a SIGINT from the startup process, or when there is an error.
    1519             :      */
    1520             :     Assert(false);
    1521             : }
    1522             : 
    1523             : /*
    1524             :  * Update the inactive_since property for synced slots.
    1525             :  *
    1526             :  * Note that this function is currently called when we shutdown the slot
    1527             :  * sync machinery.
    1528             :  */
    1529             : static void
    1530        1698 : update_synced_slots_inactive_since(void)
    1531             : {
    1532        1698 :     TimestampTz now = 0;
    1533             : 
    1534             :     /*
    1535             :      * We need to update inactive_since only when we are promoting standby to
    1536             :      * correctly interpret the inactive_since if the standby gets promoted
    1537             :      * without a restart. We don't want the slots to appear inactive for a
    1538             :      * long time after promotion if they haven't been synchronized recently.
    1539             :      * Whoever acquires the slot, i.e., makes the slot active, will reset it.
    1540             :      */
    1541        1698 :     if (!StandbyMode)
    1542        1606 :         return;
    1543             : 
    1544             :     /* The slot sync worker or SQL function mustn't be running by now */
    1545             :     Assert((SlotSyncCtx->pid == InvalidPid) && !SlotSyncCtx->syncing);
    1546             : 
    1547          92 :     LWLockAcquire(ReplicationSlotControlLock, LW_SHARED);
    1548             : 
    1549         988 :     for (int i = 0; i < max_replication_slots; i++)
    1550             :     {
    1551         896 :         ReplicationSlot *s = &ReplicationSlotCtl->replication_slots[i];
    1552             : 
    1553             :         /* Check if it is a synchronized slot */
    1554         896 :         if (s->in_use && s->data.synced)
    1555             :         {
    1556             :             Assert(SlotIsLogical(s));
    1557             : 
    1558             :             /* The slot must not be acquired by any process */
    1559             :             Assert(s->active_pid == 0);
    1560             : 
    1561             :             /* Use the same inactive_since time for all the slots. */
    1562           6 :             if (now == 0)
    1563           4 :                 now = GetCurrentTimestamp();
    1564             : 
    1565           6 :             ReplicationSlotSetInactiveSince(s, now, true);
    1566             :         }
    1567             :     }
    1568             : 
    1569          92 :     LWLockRelease(ReplicationSlotControlLock);
    1570             : }
    1571             : 
    1572             : /*
    1573             :  * Shut down the slot sync worker.
    1574             :  *
    1575             :  * This function sends signal to shutdown slot sync worker, if required. It
    1576             :  * also waits till the slot sync worker has exited or
    1577             :  * pg_sync_replication_slots() has finished.
    1578             :  */
    1579             : void
    1580        1698 : ShutDownSlotSync(void)
    1581             : {
    1582             :     pid_t       worker_pid;
    1583             : 
    1584        1698 :     SpinLockAcquire(&SlotSyncCtx->mutex);
    1585             : 
    1586        1698 :     SlotSyncCtx->stopSignaled = true;
    1587             : 
    1588             :     /*
    1589             :      * Return if neither the slot sync worker is running nor the function
    1590             :      * pg_sync_replication_slots() is executing.
    1591             :      */
    1592        1698 :     if (!SlotSyncCtx->syncing)
    1593             :     {
    1594        1696 :         SpinLockRelease(&SlotSyncCtx->mutex);
    1595        1696 :         update_synced_slots_inactive_since();
    1596        1696 :         return;
    1597             :     }
    1598             : 
    1599           2 :     worker_pid = SlotSyncCtx->pid;
    1600             : 
    1601           2 :     SpinLockRelease(&SlotSyncCtx->mutex);
    1602             : 
    1603           2 :     if (worker_pid != InvalidPid)
    1604           2 :         kill(worker_pid, SIGINT);
    1605             : 
    1606             :     /* Wait for slot sync to end */
    1607             :     for (;;)
    1608           0 :     {
    1609             :         int         rc;
    1610             : 
    1611             :         /* Wait a bit, we don't expect to have to wait long */
    1612           2 :         rc = WaitLatch(MyLatch,
    1613             :                        WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
    1614             :                        10L, WAIT_EVENT_REPLICATION_SLOTSYNC_SHUTDOWN);
    1615             : 
    1616           2 :         if (rc & WL_LATCH_SET)
    1617             :         {
    1618           0 :             ResetLatch(MyLatch);
    1619           0 :             CHECK_FOR_INTERRUPTS();
    1620             :         }
    1621             : 
    1622           2 :         SpinLockAcquire(&SlotSyncCtx->mutex);
    1623             : 
    1624             :         /* Ensure that no process is syncing the slots. */
    1625           2 :         if (!SlotSyncCtx->syncing)
    1626           2 :             break;
    1627             : 
    1628           0 :         SpinLockRelease(&SlotSyncCtx->mutex);
    1629             :     }
    1630             : 
    1631           2 :     SpinLockRelease(&SlotSyncCtx->mutex);
    1632             : 
    1633           2 :     update_synced_slots_inactive_since();
    1634             : }
    1635             : 
    1636             : /*
    1637             :  * SlotSyncWorkerCanRestart
    1638             :  *
    1639             :  * Returns true if enough time (SLOTSYNC_RESTART_INTERVAL_SEC) has passed
    1640             :  * since it was launched last. Otherwise returns false.
    1641             :  *
    1642             :  * This is a safety valve to protect against continuous respawn attempts if the
    1643             :  * worker is dying immediately at launch. Note that since we will retry to
    1644             :  * launch the worker from the postmaster main loop, we will get another
    1645             :  * chance later.
    1646             :  */
    1647             : bool
    1648          10 : SlotSyncWorkerCanRestart(void)
    1649             : {
    1650          10 :     time_t      curtime = time(NULL);
    1651             : 
    1652             :     /* Return false if too soon since last start. */
    1653          10 :     if ((unsigned int) (curtime - SlotSyncCtx->last_start_time) <
    1654             :         (unsigned int) SLOTSYNC_RESTART_INTERVAL_SEC)
    1655           2 :         return false;
    1656             : 
    1657           8 :     SlotSyncCtx->last_start_time = curtime;
    1658             : 
    1659           8 :     return true;
    1660             : }
    1661             : 
    1662             : /*
    1663             :  * Is current process syncing replication slots?
    1664             :  *
    1665             :  * Could be either backend executing SQL function or slot sync worker.
    1666             :  */
    1667             : bool
    1668          42 : IsSyncingReplicationSlots(void)
    1669             : {
    1670          42 :     return syncing_slots;
    1671             : }
    1672             : 
    1673             : /*
    1674             :  * Amount of shared memory required for slot synchronization.
    1675             :  */
    1676             : Size
    1677        6006 : SlotSyncShmemSize(void)
    1678             : {
    1679        6006 :     return sizeof(SlotSyncCtxStruct);
    1680             : }
    1681             : 
    1682             : /*
    1683             :  * Allocate and initialize the shared memory of slot synchronization.
    1684             :  */
    1685             : void
    1686        2100 : SlotSyncShmemInit(void)
    1687             : {
    1688        2100 :     Size        size = SlotSyncShmemSize();
    1689             :     bool        found;
    1690             : 
    1691        2100 :     SlotSyncCtx = (SlotSyncCtxStruct *)
    1692        2100 :         ShmemInitStruct("Slot Sync Data", size, &found);
    1693             : 
    1694        2100 :     if (!found)
    1695             :     {
    1696        2100 :         memset(SlotSyncCtx, 0, size);
    1697        2100 :         SlotSyncCtx->pid = InvalidPid;
    1698        2100 :         SpinLockInit(&SlotSyncCtx->mutex);
    1699             :     }
    1700        2100 : }
    1701             : 
    1702             : /*
    1703             :  * Error cleanup callback for slot sync SQL function.
    1704             :  */
    1705             : static void
    1706           2 : slotsync_failure_callback(int code, Datum arg)
    1707             : {
    1708           2 :     WalReceiverConn *wrconn = (WalReceiverConn *) DatumGetPointer(arg);
    1709             : 
    1710             :     /*
    1711             :      * We need to do slots cleanup here just like WalSndErrorCleanup() does.
    1712             :      *
    1713             :      * The startup process during promotion invokes ShutDownSlotSync() which
    1714             :      * waits for slot sync to finish and it does that by checking the
    1715             :      * 'syncing' flag. Thus the SQL function must be done with slots' release
    1716             :      * and cleanup to avoid any dangling temporary slots or active slots
    1717             :      * before it marks itself as finished syncing.
    1718             :      */
    1719             : 
    1720             :     /* Make sure active replication slots are released */
    1721           2 :     if (MyReplicationSlot != NULL)
    1722           0 :         ReplicationSlotRelease();
    1723             : 
    1724             :     /* Also cleanup the synced temporary slots. */
    1725           2 :     ReplicationSlotCleanup(true);
    1726             : 
    1727             :     /*
    1728             :      * The set syncing_slots indicates that the process errored out without
    1729             :      * resetting the flag. So, we need to clean up shared memory and reset the
    1730             :      * flag here.
    1731             :      */
    1732           2 :     if (syncing_slots)
    1733           2 :         reset_syncing_flag();
    1734             : 
    1735           2 :     walrcv_disconnect(wrconn);
    1736           2 : }
    1737             : 
    1738             : /*
    1739             :  * Synchronize the failover enabled replication slots using the specified
    1740             :  * primary server connection.
    1741             :  */
    1742             : void
    1743          16 : SyncReplicationSlots(WalReceiverConn *wrconn)
    1744             : {
    1745          16 :     PG_ENSURE_ERROR_CLEANUP(slotsync_failure_callback, PointerGetDatum(wrconn));
    1746             :     {
    1747          16 :         check_and_set_sync_info(InvalidPid);
    1748             : 
    1749          16 :         validate_remote_info(wrconn);
    1750             : 
    1751          14 :         synchronize_slots(wrconn);
    1752             : 
    1753             :         /* Cleanup the synced temporary slots */
    1754          14 :         ReplicationSlotCleanup(true);
    1755             : 
    1756             :         /* We are done with sync, so reset sync flag */
    1757          14 :         reset_syncing_flag();
    1758             :     }
    1759          16 :     PG_END_ENSURE_ERROR_CLEANUP(slotsync_failure_callback, PointerGetDatum(wrconn));
    1760          14 : }

Generated by: LCOV version 1.14