PostgreSQL Source Code  git master
slotsync.h File Reference
Include dependency graph for slotsync.h:
This graph shows which files directly or indirectly include this file:

Go to the source code of this file.

Functions

char * CheckAndGetDbnameFromConninfo (void)
 
bool ValidateSlotSyncParams (int elevel)
 
void ReplSlotSyncWorkerMain (char *startup_data, size_t startup_data_len) pg_attribute_noreturn()
 
void ShutDownSlotSync (void)
 
bool SlotSyncWorkerCanRestart (void)
 
bool IsSyncingReplicationSlots (void)
 
Size SlotSyncShmemSize (void)
 
void SlotSyncShmemInit (void)
 
void SyncReplicationSlots (WalReceiverConn *wrconn)
 

Variables

PGDLLIMPORT bool sync_replication_slots
 
PGDLLIMPORT char * PrimaryConnInfo
 
PGDLLIMPORT char * PrimarySlotName
 

Function Documentation

◆ CheckAndGetDbnameFromConninfo()

char* CheckAndGetDbnameFromConninfo ( void  )

Definition at line 1013 of file slotsync.c.

1014 {
1015  char *dbname;
1016 
1017  /*
1018  * The slot synchronization needs a database connection for walrcv_exec to
1019  * work.
1020  */
1022  if (dbname == NULL)
1023  ereport(ERROR,
1024 
1025  /*
1026  * translator: dbname is a specific option; %s is a GUC variable name
1027  */
1028  errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1029  errmsg("slot synchronization requires dbname to be specified in %s",
1030  "primary_conninfo"));
1031  return dbname;
1032 }
int errcode(int sqlerrcode)
Definition: elog.c:855
int errmsg(const char *fmt,...)
Definition: elog.c:1068
#define ERROR
Definition: elog.h:39
#define ereport(elevel,...)
Definition: elog.h:149
char * dbname
Definition: streamutil.c:52
#define walrcv_get_dbname_from_conninfo(conninfo)
Definition: walreceiver.h:442
char * PrimaryConnInfo
Definition: xlogrecovery.c:96

References dbname, ereport, errcode(), errmsg(), ERROR, PrimaryConnInfo, and walrcv_get_dbname_from_conninfo.

Referenced by pg_sync_replication_slots(), and ReplSlotSyncWorkerMain().

◆ IsSyncingReplicationSlots()

bool IsSyncingReplicationSlots ( void  )

Definition at line 1651 of file slotsync.c.

1652 {
1653  return syncing_slots;
1654 }
static bool syncing_slots
Definition: slotsync.c:129

References syncing_slots.

Referenced by CreateDecodingContext(), GetStandbyFlushRecPtr(), and ReplicationSlotCreate().

◆ ReplSlotSyncWorkerMain()

void ReplSlotSyncWorkerMain ( char *  startup_data,
size_t  startup_data_len 
)

Definition at line 1331 of file slotsync.c.

1332 {
1333  WalReceiverConn *wrconn = NULL;
1334  char *dbname;
1335  char *err;
1336  sigjmp_buf local_sigjmp_buf;
1337  StringInfoData app_name;
1338 
1339  Assert(startup_data_len == 0);
1340 
1342 
1343  init_ps_display(NULL);
1344 
1346 
1347  /*
1348  * Create a per-backend PGPROC struct in shared memory. We must do this
1349  * before we access any shared memory.
1350  */
1351  InitProcess();
1352 
1353  /*
1354  * Early initialization.
1355  */
1356  BaseInit();
1357 
1358  Assert(SlotSyncCtx != NULL);
1359 
1360  /*
1361  * If an exception is encountered, processing resumes here.
1362  *
1363  * We just need to clean up, report the error, and go away.
1364  *
1365  * If we do not have this handling here, then since this worker process
1366  * operates at the bottom of the exception stack, ERRORs turn into FATALs.
1367  * Therefore, we create our own exception handler to catch ERRORs.
1368  */
1369  if (sigsetjmp(local_sigjmp_buf, 1) != 0)
1370  {
1371  /* since not using PG_TRY, must reset error stack by hand */
1372  error_context_stack = NULL;
1373 
1374  /* Prevents interrupts while cleaning up */
1375  HOLD_INTERRUPTS();
1376 
1377  /* Report the error to the server log */
1378  EmitErrorReport();
1379 
1380  /*
1381  * We can now go away. Note that because we called InitProcess, a
1382  * callback was registered to do ProcKill, which will clean up
1383  * necessary state.
1384  */
1385  proc_exit(0);
1386  }
1387 
1388  /* We can now handle ereport(ERROR) */
1389  PG_exception_stack = &local_sigjmp_buf;
1390 
1391  /* Setup signal handling */
1394  pqsignal(SIGTERM, die);
1400 
1402 
1403  ereport(LOG, errmsg("slot sync worker started"));
1404 
1405  /* Register it as soon as SlotSyncCtx->pid is initialized. */
1407 
1408  /*
1409  * Establishes SIGALRM handler and initialize timeout module. It is needed
1410  * by InitPostgres to register different timeouts.
1411  */
1413 
1414  /* Load the libpq-specific functions */
1415  load_file("libpqwalreceiver", false);
1416 
1417  /*
1418  * Unblock signals (they were blocked when the postmaster forked us)
1419  */
1420  sigprocmask(SIG_SETMASK, &UnBlockSig, NULL);
1421 
1422  /*
1423  * Set always-secure search path, so malicious users can't redirect user
1424  * code (e.g. operators).
1425  *
1426  * It's not strictly necessary since we won't be scanning or writing to
1427  * any user table locally, but it's good to retain it here for added
1428  * precaution.
1429  */
1430  SetConfigOption("search_path", "", PGC_SUSET, PGC_S_OVERRIDE);
1431 
1433 
1434  /*
1435  * Connect to the database specified by the user in primary_conninfo. We
1436  * need a database connection for walrcv_exec to work which we use to
1437  * fetch slot information from the remote node. See comments atop
1438  * libpqrcv_exec.
1439  *
1440  * We do not specify a specific user here since the slot sync worker will
1441  * operate as a superuser. This is safe because the slot sync worker does
1442  * not interact with user tables, eliminating the risk of executing
1443  * arbitrary code within triggers.
1444  */
1445  InitPostgres(dbname, InvalidOid, NULL, InvalidOid, 0, NULL);
1446 
1448 
1449  initStringInfo(&app_name);
1450  if (cluster_name[0])
1451  appendStringInfo(&app_name, "%s_%s", cluster_name, "slotsync worker");
1452  else
1453  appendStringInfoString(&app_name, "slotsync worker");
1454 
1455  /*
1456  * Establish the connection to the primary server for slot
1457  * synchronization.
1458  */
1459  wrconn = walrcv_connect(PrimaryConnInfo, false, false, false,
1460  app_name.data, &err);
1461  pfree(app_name.data);
1462 
1463  if (!wrconn)
1464  ereport(ERROR,
1465  errcode(ERRCODE_CONNECTION_FAILURE),
1466  errmsg("synchronization worker \"%s\" could not connect to the primary server: %s",
1467  app_name.data, err));
1468 
1469  /*
1470  * Register the disconnection callback.
1471  *
1472  * XXX: This can be combined with previous cleanup registration of
1473  * slotsync_worker_onexit() but that will need the connection to be made
1474  * global and we want to avoid introducing global for this purpose.
1475  */
1477 
1478  /*
1479  * Using the specified primary server connection, check that we are not a
1480  * cascading standby and slot configured in 'primary_slot_name' exists on
1481  * the primary server.
1482  */
1484 
1485  /* Main loop to synchronize slots */
1486  for (;;)
1487  {
1488  bool some_slot_updated = false;
1489 
1491 
1492  some_slot_updated = synchronize_slots(wrconn);
1493 
1494  wait_for_slot_activity(some_slot_updated);
1495  }
1496 
1497  /*
1498  * The slot sync worker can't get here because it will only stop when it
1499  * receives a SIGINT from the startup process, or when there is an error.
1500  */
1501  Assert(false);
1502 }
sigset_t UnBlockSig
Definition: pqsignal.c:22
#define Assert(condition)
Definition: c.h:858
void load_file(const char *filename, bool restricted)
Definition: dfmgr.c:144
void EmitErrorReport(void)
Definition: elog.c:1668
ErrorContextCallback * error_context_stack
Definition: elog.c:94
sigjmp_buf * PG_exception_stack
Definition: elog.c:96
#define LOG
Definition: elog.h:31
void err(int eval, const char *fmt,...)
Definition: err.c:43
int MyProcPid
Definition: globals.c:46
void SetConfigOption(const char *name, const char *value, GucContext context, GucSource source)
Definition: guc.c:4285
@ PGC_S_OVERRIDE
Definition: guc.h:119
@ PGC_SUSET
Definition: guc.h:74
char * cluster_name
Definition: guc_tables.c:541
void SignalHandlerForShutdownRequest(SIGNAL_ARGS)
Definition: interrupt.c:105
void SignalHandlerForConfigReload(SIGNAL_ARGS)
Definition: interrupt.c:61
void before_shmem_exit(pg_on_exit_callback function, Datum arg)
Definition: ipc.c:337
void proc_exit(int code)
Definition: ipc.c:104
void pfree(void *pointer)
Definition: mcxt.c:1521
@ NormalProcessing
Definition: miscadmin.h:449
@ InitProcessing
Definition: miscadmin.h:448
#define GetProcessingMode()
Definition: miscadmin.h:458
#define HOLD_INTERRUPTS()
Definition: miscadmin.h:133
#define SetProcessingMode(mode)
Definition: miscadmin.h:460
@ B_SLOTSYNC_WORKER
Definition: miscadmin.h:343
BackendType MyBackendType
Definition: miscinit.c:63
#define die(msg)
pqsigfunc pqsignal(int signo, pqsigfunc func)
void FloatExceptionHandler(SIGNAL_ARGS)
Definition: postgres.c:3019
static Datum PointerGetDatum(const void *X)
Definition: postgres.h:322
uintptr_t Datum
Definition: postgres.h:64
#define InvalidOid
Definition: postgres_ext.h:36
void BaseInit(void)
Definition: postinit.c:602
void InitPostgres(const char *in_dbname, Oid dboid, const char *username, Oid useroid, bits32 flags, char *out_dbname)
Definition: postinit.c:693
void procsignal_sigusr1_handler(SIGNAL_ARGS)
Definition: procsignal.c:635
void init_ps_display(const char *fixed_part)
Definition: ps_status.c:267
static void slotsync_worker_disconnect(int code, Datum arg)
Definition: slotsync.c:1177
static SlotSyncCtxStruct * SlotSyncCtx
Definition: slotsync.c:106
static bool synchronize_slots(WalReceiverConn *wrconn)
Definition: slotsync.c:791
static void wait_for_slot_activity(bool some_slot_updated)
Definition: slotsync.c:1236
char * CheckAndGetDbnameFromConninfo(void)
Definition: slotsync.c:1013
static void slotsync_worker_onexit(int code, Datum arg)
Definition: slotsync.c:1190
static void check_and_set_sync_info(pid_t worker_pid)
Definition: slotsync.c:1271
static void validate_remote_info(WalReceiverConn *wrconn)
Definition: slotsync.c:934
static void ProcessSlotSyncInterrupts(WalReceiverConn *wrconn)
Definition: slotsync.c:1155
void InitProcess(void)
Definition: proc.c:296
void appendStringInfo(StringInfo str, const char *fmt,...)
Definition: stringinfo.c:97
void appendStringInfoString(StringInfo str, const char *s)
Definition: stringinfo.c:182
void initStringInfo(StringInfo str)
Definition: stringinfo.c:59
void InitializeTimeouts(void)
Definition: timeout.c:470
static WalReceiverConn * wrconn
Definition: walreceiver.c:92
#define walrcv_connect(conninfo, replication, logical, must_use_password, appname, err)
Definition: walreceiver.h:432
#define SIGCHLD
Definition: win32_port.h:178
#define SIGHUP
Definition: win32_port.h:168
#define SIG_DFL
Definition: win32_port.h:163
#define SIGPIPE
Definition: win32_port.h:173
#define SIGUSR1
Definition: win32_port.h:180
#define SIGUSR2
Definition: win32_port.h:181
#define SIG_IGN
Definition: win32_port.h:165

References appendStringInfo(), appendStringInfoString(), Assert, B_SLOTSYNC_WORKER, BaseInit(), before_shmem_exit(), check_and_set_sync_info(), CheckAndGetDbnameFromConninfo(), cluster_name, StringInfoData::data, dbname, die, EmitErrorReport(), ereport, err(), errcode(), errmsg(), ERROR, error_context_stack, FloatExceptionHandler(), GetProcessingMode, HOLD_INTERRUPTS, init_ps_display(), InitializeTimeouts(), InitPostgres(), InitProcess(), InitProcessing, initStringInfo(), InvalidOid, load_file(), LOG, MyBackendType, MyProcPid, NormalProcessing, pfree(), PG_exception_stack, PGC_S_OVERRIDE, PGC_SUSET, PointerGetDatum(), pqsignal(), PrimaryConnInfo, proc_exit(), ProcessSlotSyncInterrupts(), procsignal_sigusr1_handler(), SetConfigOption(), SetProcessingMode, SIG_DFL, SIG_IGN, SIGCHLD, SIGHUP, SignalHandlerForConfigReload(), SignalHandlerForShutdownRequest(), SIGPIPE, SIGUSR1, SIGUSR2, slotsync_worker_disconnect(), slotsync_worker_onexit(), SlotSyncCtx, synchronize_slots(), UnBlockSig, validate_remote_info(), wait_for_slot_activity(), walrcv_connect, and wrconn.

◆ ShutDownSlotSync()

void ShutDownSlotSync ( void  )

Definition at line 1563 of file slotsync.c.

1564 {
1565  pid_t worker_pid;
1566 
1568 
1569  SlotSyncCtx->stopSignaled = true;
1570 
1571  /*
1572  * Return if neither the slot sync worker is running nor the function
1573  * pg_sync_replication_slots() is executing.
1574  */
1575  if (!SlotSyncCtx->syncing)
1576  {
1579  return;
1580  }
1581 
1582  worker_pid = SlotSyncCtx->pid;
1583 
1585 
1586  if (worker_pid != InvalidPid)
1587  kill(worker_pid, SIGINT);
1588 
1589  /* Wait for slot sync to end */
1590  for (;;)
1591  {
1592  int rc;
1593 
1594  /* Wait a bit, we don't expect to have to wait long */
1595  rc = WaitLatch(MyLatch,
1597  10L, WAIT_EVENT_REPLICATION_SLOTSYNC_SHUTDOWN);
1598 
1599  if (rc & WL_LATCH_SET)
1600  {
1603  }
1604 
1606 
1607  /* Ensure that no process is syncing the slots. */
1608  if (!SlotSyncCtx->syncing)
1609  break;
1610 
1612  }
1613 
1615 
1617 }
struct Latch * MyLatch
Definition: globals.c:61
void ResetLatch(Latch *latch)
Definition: latch.c:724
int WaitLatch(Latch *latch, int wakeEvents, long timeout, uint32 wait_event_info)
Definition: latch.c:517
#define WL_TIMEOUT
Definition: latch.h:130
#define WL_EXIT_ON_PM_DEATH
Definition: latch.h:132
#define WL_LATCH_SET
Definition: latch.h:127
#define CHECK_FOR_INTERRUPTS()
Definition: miscadmin.h:122
#define InvalidPid
Definition: miscadmin.h:32
static void update_synced_slots_inactive_since(void)
Definition: slotsync.c:1511
#define SpinLockRelease(lock)
Definition: spin.h:64
#define SpinLockAcquire(lock)
Definition: spin.h:62
#define kill(pid, sig)
Definition: win32_port.h:485

References CHECK_FOR_INTERRUPTS, InvalidPid, kill, SlotSyncCtxStruct::mutex, MyLatch, SlotSyncCtxStruct::pid, ResetLatch(), SlotSyncCtx, SpinLockAcquire, SpinLockRelease, SlotSyncCtxStruct::stopSignaled, SlotSyncCtxStruct::syncing, update_synced_slots_inactive_since(), WaitLatch(), WL_EXIT_ON_PM_DEATH, WL_LATCH_SET, and WL_TIMEOUT.

Referenced by FinishWalRecovery().

◆ SlotSyncShmemInit()

void SlotSyncShmemInit ( void  )

Definition at line 1669 of file slotsync.c.

1670 {
1672  bool found;
1673 
1675  ShmemInitStruct("Slot Sync Data", size, &found);
1676 
1677  if (!found)
1678  {
1679  memset(SlotSyncCtx, 0, size);
1682  }
1683 }
size_t Size
Definition: c.h:605
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Definition: shmem.c:387
static pg_noinline void Size size
Definition: slab.c:607
Size SlotSyncShmemSize(void)
Definition: slotsync.c:1660
#define SpinLockInit(lock)
Definition: spin.h:60

References InvalidPid, SlotSyncCtxStruct::mutex, SlotSyncCtxStruct::pid, ShmemInitStruct(), size, SlotSyncCtx, SlotSyncShmemSize(), and SpinLockInit.

Referenced by CreateOrAttachShmemStructs().

◆ SlotSyncShmemSize()

Size SlotSyncShmemSize ( void  )

Definition at line 1660 of file slotsync.c.

1661 {
1662  return sizeof(SlotSyncCtxStruct);
1663 }
struct SlotSyncCtxStruct SlotSyncCtxStruct

Referenced by CalculateShmemSize(), and SlotSyncShmemInit().

◆ SlotSyncWorkerCanRestart()

bool SlotSyncWorkerCanRestart ( void  )

Definition at line 1631 of file slotsync.c.

1632 {
1633  time_t curtime = time(NULL);
1634 
1635  /* Return false if too soon since last start. */
1636  if ((unsigned int) (curtime - SlotSyncCtx->last_start_time) <
1637  (unsigned int) SLOTSYNC_RESTART_INTERVAL_SEC)
1638  return false;
1639 
1640  SlotSyncCtx->last_start_time = curtime;
1641 
1642  return true;
1643 }
#define SLOTSYNC_RESTART_INTERVAL_SEC
Definition: slotsync.c:122
time_t last_start_time
Definition: slotsync.c:102

References SlotSyncCtxStruct::last_start_time, SLOTSYNC_RESTART_INTERVAL_SEC, and SlotSyncCtx.

Referenced by MaybeStartSlotSyncWorker().

◆ SyncReplicationSlots()

void SyncReplicationSlots ( WalReceiverConn wrconn)

Definition at line 1726 of file slotsync.c.

1727 {
1729  {
1731 
1733 
1735 
1736  /* Cleanup the synced temporary slots */
1737  ReplicationSlotCleanup(true);
1738 
1739  /* We are done with sync, so reset sync flag */
1741  }
1743 }
#define PG_ENSURE_ERROR_CLEANUP(cleanup_function, arg)
Definition: ipc.h:47
#define PG_END_ENSURE_ERROR_CLEANUP(cleanup_function, arg)
Definition: ipc.h:52
void ReplicationSlotCleanup(bool synced_only)
Definition: slot.c:745
static void slotsync_failure_callback(int code, Datum arg)
Definition: slotsync.c:1689
static void reset_syncing_flag()
Definition: slotsync.c:1315

References check_and_set_sync_info(), InvalidPid, PG_END_ENSURE_ERROR_CLEANUP, PG_ENSURE_ERROR_CLEANUP, PointerGetDatum(), ReplicationSlotCleanup(), reset_syncing_flag(), slotsync_failure_callback(), synchronize_slots(), validate_remote_info(), and wrconn.

Referenced by pg_sync_replication_slots().

◆ ValidateSlotSyncParams()

bool ValidateSlotSyncParams ( int  elevel)

Definition at line 1039 of file slotsync.c.

1040 {
1041  /*
1042  * Logical slot sync/creation requires wal_level >= logical.
1043  *
1044  * Since altering the wal_level requires a server restart, so error out in
1045  * this case regardless of elevel provided by caller.
1046  */
1048  ereport(ERROR,
1049  errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1050  errmsg("slot synchronization requires \"wal_level\" >= \"logical\""));
1051 
1052  /*
1053  * A physical replication slot(primary_slot_name) is required on the
1054  * primary to ensure that the rows needed by the standby are not removed
1055  * after restarting, so that the synchronized slot on the standby will not
1056  * be invalidated.
1057  */
1058  if (PrimarySlotName == NULL || *PrimarySlotName == '\0')
1059  {
1060  ereport(elevel,
1061  /* translator: %s is a GUC variable name */
1062  errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1063  errmsg("slot synchronization requires %s to be defined", "primary_slot_name"));
1064  return false;
1065  }
1066 
1067  /*
1068  * hot_standby_feedback must be enabled to cooperate with the physical
1069  * replication slot, which allows informing the primary about the xmin and
1070  * catalog_xmin values on the standby.
1071  */
1072  if (!hot_standby_feedback)
1073  {
1074  ereport(elevel,
1075  /* translator: %s is a GUC variable name */
1076  errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1077  errmsg("slot synchronization requires %s to be enabled",
1078  "hot_standby_feedback"));
1079  return false;
1080  }
1081 
1082  /*
1083  * The primary_conninfo is required to make connection to primary for
1084  * getting slots information.
1085  */
1086  if (PrimaryConnInfo == NULL || *PrimaryConnInfo == '\0')
1087  {
1088  ereport(elevel,
1089  /* translator: %s is a GUC variable name */
1090  errcode(ERRCODE_INVALID_PARAMETER_VALUE),
1091  errmsg("slot synchronization requires %s to be defined",
1092  "primary_conninfo"));
1093  return false;
1094  }
1095 
1096  return true;
1097 }
bool hot_standby_feedback
Definition: walreceiver.c:89
int wal_level
Definition: xlog.c:131
@ WAL_LEVEL_LOGICAL
Definition: xlog.h:74
char * PrimarySlotName
Definition: xlogrecovery.c:97

References ereport, errcode(), errmsg(), ERROR, hot_standby_feedback, PrimaryConnInfo, PrimarySlotName, wal_level, and WAL_LEVEL_LOGICAL.

Referenced by MaybeStartSlotSyncWorker(), and pg_sync_replication_slots().

Variable Documentation

◆ PrimaryConnInfo

◆ PrimarySlotName

◆ sync_replication_slots

PGDLLIMPORT bool sync_replication_slots
extern

Definition at line 109 of file slotsync.c.

Referenced by MaybeStartSlotSyncWorker(), and slotsync_reread_config().