113#define RS_INVAL_MAX_CAUSES RS_INVAL_WAL_LEVEL
116 "array length mismatch");
119#define ReplicationSlotOnDiskConstantSize \
120 offsetof(ReplicationSlotOnDisk, slotdata)
122#define ReplicationSlotOnDiskNotChecksummedSize \
123 offsetof(ReplicationSlotOnDisk, version)
125#define ReplicationSlotOnDiskChecksummedSize \
126 sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskNotChecksummedSize
128#define ReplicationSlotOnDiskV2Size \
129 sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize
131#define SLOT_MAGIC 0x1051CA1
132#define SLOT_VERSION 5
256 if (strlen(
name) == 0)
259 (
errcode(ERRCODE_INVALID_NAME),
260 errmsg(
"replication slot name \"%s\" is too short",
268 (
errcode(ERRCODE_NAME_TOO_LONG),
269 errmsg(
"replication slot name \"%s\" is too long",
274 for (cp =
name; *cp; cp++)
276 if (!((*cp >=
'a' && *cp <=
'z')
277 || (*cp >=
'0' && *cp <=
'9')
281 (
errcode(ERRCODE_INVALID_NAME),
282 errmsg(
"replication slot name \"%s\" contains invalid character",
284 errhint(
"Replication slot names may only contain lower case letters, numbers, and the underscore character.")));
311 bool two_phase,
bool failover,
bool synced)
332 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
333 errmsg(
"cannot enable failover for a replication slot created on the standby"));
344 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
345 errmsg(
"cannot enable failover for a temporary replication slot"));
370 errmsg(
"replication slot \"%s\" already exists",
name)));
371 if (!s->
in_use && slot == NULL)
379 (
errcode(ERRCODE_CONFIGURATION_LIMIT_EXCEEDED),
380 errmsg(
"all replication slots are in use"),
381 errhint(
"Free one or increase \"max_replication_slots\".")));
554 if (s == NULL || !s->
in_use)
559 (
errcode(ERRCODE_UNDEFINED_OBJECT),
560 errmsg(
"replication slot \"%s\" does not exist",
599 WAIT_EVENT_REPLICATION_SLOT_DROP);
605 (
errcode(ERRCODE_OBJECT_IN_USE),
606 errmsg(
"replication slot \"%s\" is active for PID %d",
638 ?
errmsg(
"acquired logical replication slot \"%s\"",
640 :
errmsg(
"acquired physical replication slot \"%s\"",
655 char *slotname = NULL;
656 bool is_logical =
false;
729 ?
errmsg(
"released logical replication slot \"%s\"",
731 :
errmsg(
"released physical replication slot \"%s\"",
796 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
797 errmsg(
"cannot drop replication slot \"%s\"",
name),
798 errdetail(
"This replication slot is being synchronized from the primary server."));
810 bool update_slot =
false;
819 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
820 errmsg(
"cannot use %s with a physical replication slot",
821 "ALTER_REPLICATION_SLOT"));
825 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
826 errmsg(
"cannot alter invalid replication slot \"%s\"",
name),
827 errdetail(
"This replication slot has been invalidated due to \"%s\".",
838 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
839 errmsg(
"cannot alter replication slot \"%s\"",
name),
840 errdetail(
"This replication slot is being synchronized from the primary server."));
846 if (failover && *failover)
848 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
849 errmsg(
"cannot enable failover for a replication slot"
861 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
862 errmsg(
"cannot enable failover for a temporary replication slot"));
936 if (rename(path, tmppath) == 0)
964 errmsg(
"could not rename file \"%s\" to \"%s\": %m",
995 if (!
rmtree(tmppath,
true))
997 (
errmsg(
"could not remove directory \"%s\"", tmppath)));
1111 agg_xmin = effective_xmin;
1117 agg_catalog_xmin = effective_catalog_xmin;
1161 restart_lsn < min_required))
1162 min_required = restart_lsn;
1222 restart_lsn < result)
1223 result = restart_lsn;
1244 *nslots = *nactive = 0;
1334 if (active_pid == 0)
1364 (
errcode(ERRCODE_OBJECT_IN_USE),
1365 errmsg(
"replication slot \"%s\" is active for PID %d",
1366 slotname, active_pid)));
1399 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1400 errmsg(
"replication slots can only be used if \"max_replication_slots\" > 0")));
1404 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1405 errmsg(
"replication slots can only be used if \"wal_level\" >= \"replica\"")));
1416 (
errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1417 errmsg(
"permission denied to use replication slots"),
1418 errdetail(
"Only roles with the %s attribute may use replication slots.",
1522 unsigned long long ex = oldestLSN - restart_lsn;
1526 ngettext(
"The slot's restart_lsn %X/%X exceeds the limit by %llu byte.",
1527 "The slot's restart_lsn %X/%X exceeds the limit by %llu bytes.",
1535 snapshotConflictHorizon);
1539 appendStringInfoString(&err_detail,
_(
"Logical decoding on standby requires \"wal_level\" >= \"logical\" on the primary server."));
1547 errmsg(
"terminating process %d to release replication slot \"%s\"",
1549 errmsg(
"invalidating obsolete replication slot \"%s\"",
1552 hint ?
errhint(
"You might need to increase \"%s\".",
"max_slot_wal_keep_size") : 0);
1577 int last_signaled_pid = 0;
1578 bool released_lock =
false;
1579 bool terminated =
false;
1632 initial_restart_lsn < oldestLSN)
1633 invalidation_cause = cause;
1643 snapshotConflictHorizon))
1644 invalidation_cause = cause;
1647 snapshotConflictHorizon))
1648 invalidation_cause = cause;
1652 invalidation_cause = cause;
1664 invalidation_cause_prev != invalidation_cause));
1683 if (active_pid == 0)
1697 *invalidated =
true;
1710 if (active_pid != 0)
1720 released_lock =
true;
1734 if (last_signaled_pid != active_pid)
1737 slotname, restart_lsn,
1738 oldestLSN, snapshotConflictHorizon);
1745 (
void)
kill(active_pid, SIGTERM);
1747 last_signaled_pid = active_pid;
1749 invalidation_cause_prev = invalidation_cause;
1754 WAIT_EVENT_REPLICATION_SLOT_DROP);
1775 released_lock =
true;
1783 slotname, restart_lsn,
1784 oldestLSN, snapshotConflictHorizon);
1793 return released_lock;
1816 bool invalidated =
false;
1837 snapshotConflictHorizon,
1871 elog(
DEBUG1,
"performing replication slot checkpoint");
1926 DIR *replication_dir;
1927 struct dirent *replication_de;
1929 elog(
DEBUG1,
"starting up replication slots");
1938 if (strcmp(replication_de->
d_name,
".") == 0 ||
1939 strcmp(replication_de->
d_name,
"..") == 0)
1955 (
errmsg(
"could not remove directory \"%s\"",
2014 errmsg(
"could not create directory \"%s\": %m",
2023 if (rename(tmppath, path) != 0)
2026 errmsg(
"could not rename file \"%s\" to \"%s\": %m",
2056 was_dirty = slot->
dirty;
2069 sprintf(tmppath,
"%s/state.tmp", dir);
2070 sprintf(path,
"%s/state", dir);
2081 int save_errno = errno;
2087 errmsg(
"could not create file \"%s\": %m",
2110 if ((
write(
fd, &cp,
sizeof(cp))) !=
sizeof(cp))
2112 int save_errno = errno;
2119 errno = save_errno ? save_errno : ENOSPC;
2122 errmsg(
"could not write to file \"%s\": %m",
2132 int save_errno = errno;
2140 errmsg(
"could not fsync file \"%s\": %m",
2148 int save_errno = errno;
2154 errmsg(
"could not close file \"%s\": %m",
2160 if (rename(tmppath, path) != 0)
2162 int save_errno = errno;
2168 errmsg(
"could not rename file \"%s\" to \"%s\": %m",
2190 slot->
dirty =
false;
2208 bool restored =
false;
2216 sprintf(path,
"%s/state.tmp", slotdir);
2217 if (unlink(path) < 0 && errno != ENOENT)
2220 errmsg(
"could not remove file \"%s\": %m", path)));
2222 sprintf(path,
"%s/state", slotdir);
2224 elog(
DEBUG1,
"restoring replication slot from \"%s\"", path);
2236 errmsg(
"could not open file \"%s\": %m", path)));
2246 errmsg(
"could not fsync file \"%s\": %m",
2264 errmsg(
"could not read file \"%s\": %m", path)));
2268 errmsg(
"could not read file \"%s\": read %d of %zu",
2277 errmsg(
"replication slot file \"%s\" has wrong magic number: %u instead of %u",
2284 errmsg(
"replication slot file \"%s\" has unsupported version %u",
2291 errmsg(
"replication slot file \"%s\" has corrupted length %u",
2300 if (readBytes != cp.
length)
2305 errmsg(
"could not read file \"%s\": %m", path)));
2309 errmsg(
"could not read file \"%s\": read %d of %zu",
2316 errmsg(
"could not close file \"%s\": %m", path)));
2327 (
errmsg(
"checksum mismatch for replication slot file \"%s\": is %u, should be %u",
2336 if (!
rmtree(slotdir,
true))
2339 (
errmsg(
"could not remove directory \"%s\"",
2360 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2361 errmsg(
"logical replication slot \"%s\" exists, but \"wal_level\" < \"logical\"",
2363 errhint(
"Change \"wal_level\" to be \"logical\" or higher.")));
2366 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2367 errmsg(
"physical replication slot \"%s\" exists, but \"wal_level\" < \"replica\"",
2369 errhint(
"Change \"wal_level\" to be \"replica\" or higher.")));
2411 (
errmsg(
"too many replication slots active before shutdown"),
2412 errhint(
"Increase \"max_replication_slots\" and try again.")));
2426 Assert(invalidation_reason);
2512 if ((*
newval)[0] ==
'\0')
2521 if (!ok || elemlist ==
NIL)
2531 size += strlen(slot_name) + 1;
2542 strcpy(ptr, slot_name);
2543 ptr += strlen(slot_name) + 1;
2574 const char *standby_slot_name;
2588 if (strcmp(standby_slot_name, slot_name) == 0)
2591 standby_slot_name += strlen(standby_slot_name) + 1;
2608 int caught_up_slot_num = 0;
2662 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2663 errmsg(
"replication slot \"%s\" specified in parameter \"%s\" does not exist",
2664 name,
"synchronized_standby_slots"),
2665 errdetail(
"Logical replication is waiting on the standby associated with replication slot \"%s\".",
2667 errhint(
"Create the replication slot \"%s\" or amend parameter \"%s\".",
2668 name,
"synchronized_standby_slots"));
2676 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2677 errmsg(
"cannot specify logical replication slot \"%s\" in parameter \"%s\"",
2678 name,
"synchronized_standby_slots"),
2679 errdetail(
"Logical replication is waiting for correction on replication slot \"%s\".",
2681 errhint(
"Remove the logical replication slot \"%s\" from parameter \"%s\".",
2682 name,
"synchronized_standby_slots"));
2696 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2697 errmsg(
"physical replication slot \"%s\" specified in parameter \"%s\" has been invalidated",
2698 name,
"synchronized_standby_slots"),
2699 errdetail(
"Logical replication is waiting on the standby associated with replication slot \"%s\".",
2701 errhint(
"Drop and recreate the replication slot \"%s\", or amend parameter \"%s\".",
2702 name,
"synchronized_standby_slots"));
2711 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2712 errmsg(
"replication slot \"%s\" specified in parameter \"%s\" does not have active_pid",
2713 name,
"synchronized_standby_slots"),
2714 errdetail(
"Logical replication is waiting on the standby associated with replication slot \"%s\".",
2716 errhint(
"Start the standby associated with the replication slot \"%s\", or amend parameter \"%s\".",
2717 name,
"synchronized_standby_slots"));
2723 Assert(restart_lsn >= wait_for_lsn);
2726 min_restart_lsn > restart_lsn)
2727 min_restart_lsn = restart_lsn;
2729 caught_up_slot_num++;
2791 WAIT_EVENT_WAIT_FOR_STANDBY_CONFIRMATION);
TimestampTz GetCurrentTimestamp(void)
Datum now(PG_FUNCTION_ARGS)
#define ngettext(s, p, n)
#define PG_USED_FOR_ASSERTS_ONLY
#define Assert(condition)
#define FLEXIBLE_ARRAY_MEMBER
#define MemSet(start, val, len)
bool ConditionVariableCancelSleep(void)
bool ConditionVariableTimedSleep(ConditionVariable *cv, long timeout, uint32 wait_event_info)
void ConditionVariableBroadcast(ConditionVariable *cv)
void ConditionVariablePrepareToSleep(ConditionVariable *cv)
void ConditionVariableInit(ConditionVariable *cv)
void ConditionVariableSleep(ConditionVariable *cv, uint32 wait_event_info)
int errdetail_internal(const char *fmt,...)
int errcode_for_file_access(void)
int errdetail(const char *fmt,...)
int errhint(const char *fmt,...)
int errcode(int sqlerrcode)
int errmsg(const char *fmt,...)
#define ereport(elevel,...)
int MakePGDirectory(const char *directoryName)
int CloseTransientFile(int fd)
void fsync_fname(const char *fname, bool isdir)
DIR * AllocateDir(const char *dirname)
struct dirent * ReadDir(DIR *dir, const char *dirname)
int OpenTransientFile(const char *fileName, int fileFlags)
PGFileType get_dirent_type(const char *path, const struct dirent *de, bool look_through_symlinks, int elevel)
void ProcessConfigFile(GucContext context)
void * guc_malloc(int elevel, size_t size)
#define GUC_check_errdetail
volatile sig_atomic_t ConfigReloadPending
void before_shmem_exit(pg_on_exit_callback function, Datum arg)
void list_free(List *list)
bool LWLockHeldByMe(LWLock *lock)
bool LWLockAcquire(LWLock *lock, LWLockMode mode)
bool LWLockHeldByMeInMode(LWLock *lock, LWLockMode mode)
void LWLockRelease(LWLock *lock)
void LWLockInitialize(LWLock *lock, int tranche_id)
@ LWTRANCHE_REPLICATION_SLOT_IO
char * pstrdup(const char *in)
void pfree(void *pointer)
#define START_CRIT_SECTION()
#define CHECK_FOR_INTERRUPTS()
#define END_CRIT_SECTION()
BackendType MyBackendType
bool has_rolreplication(Oid roleid)
void namestrcpy(Name name, const char *str)
#define ERRCODE_DATA_CORRUPTED
#define COMP_CRC32C(crc, data, len)
#define EQ_CRC32C(c1, c2)
static int list_length(const List *l)
#define foreach_ptr(type, var, lst)
static rewind_source * source
void pgstat_create_replslot(ReplicationSlot *slot)
void pgstat_acquire_replslot(ReplicationSlot *slot)
void pgstat_drop_replslot(ReplicationSlot *slot)
static int fd(const char *x, int i)
void ProcArraySetReplicationSlotXmin(TransactionId xmin, TransactionId catalog_xmin, bool already_locked)
#define INVALID_PROC_NUMBER
int SendProcSignal(pid_t pid, ProcSignalReason reason, ProcNumber procNumber)
@ PROCSIG_RECOVERY_CONFLICT_LOGICALSLOT
bool rmtree(const char *path, bool rmtopdir)
Size add_size(Size s1, Size s2)
Size mul_size(Size s1, Size s2)
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
static pg_noinline void Size size
int ReplicationSlotIndex(ReplicationSlot *slot)
char * synchronized_standby_slots
void assign_synchronized_standby_slots(const char *newval, void *extra)
#define ReplicationSlotOnDiskChecksummedSize
void CheckPointReplicationSlots(bool is_shutdown)
void ReplicationSlotCreate(const char *name, bool db_specific, ReplicationSlotPersistency persistency, bool two_phase, bool failover, bool synced)
void ReplicationSlotDropAcquired(void)
void ReplicationSlotMarkDirty(void)
void ReplicationSlotReserveWal(void)
bool ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive)
void ReplicationSlotAcquire(const char *name, bool nowait)
bool InvalidateObsoleteReplicationSlots(ReplicationSlotInvalidationCause cause, XLogSegNo oldestSegno, Oid dboid, TransactionId snapshotConflictHorizon)
static XLogRecPtr ss_oldest_flush_lsn
ReplicationSlotInvalidationCause GetSlotInvalidationCause(const char *invalidation_reason)
void ReplicationSlotsDropDBSlots(Oid dboid)
#define ReplicationSlotOnDiskNotChecksummedSize
XLogRecPtr ReplicationSlotsComputeLogicalRestartLSN(void)
void ReplicationSlotsComputeRequiredXmin(bool already_locked)
static void RestoreSlotFromDisk(const char *name)
#define RS_INVAL_MAX_CAUSES
void ReplicationSlotPersist(void)
ReplicationSlot * MyReplicationSlot
static void SaveSlotToPath(ReplicationSlot *slot, const char *dir, int elevel)
void ReplicationSlotDrop(const char *name, bool nowait)
bool SlotExistsInSyncStandbySlots(const char *slot_name)
static bool validate_sync_standby_slots(char *rawname, List **elemlist)
void ReplicationSlotSave(void)
ReplicationSlot * SearchNamedReplicationSlot(const char *name, bool need_lock)
static void CreateSlotOnDisk(ReplicationSlot *slot)
#define ReplicationSlotOnDiskV2Size
void CheckSlotPermissions(void)
bool ReplicationSlotName(int index, Name name)
bool check_synchronized_standby_slots(char **newval, void **extra, GucSource source)
void ReplicationSlotsShmemInit(void)
const char *const SlotInvalidationCauses[]
void ReplicationSlotAlter(const char *name, const bool *failover, const bool *two_phase)
void ReplicationSlotRelease(void)
int max_replication_slots
StaticAssertDecl(lengthof(SlotInvalidationCauses)==(RS_INVAL_MAX_CAUSES+1), "array length mismatch")
ReplicationSlotCtlData * ReplicationSlotCtl
struct ReplicationSlotOnDisk ReplicationSlotOnDisk
void WaitForStandbyConfirmation(XLogRecPtr wait_for_lsn)
bool StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel)
void ReplicationSlotsComputeRequiredLSN(void)
void ReplicationSlotCleanup(bool synced_only)
void ReplicationSlotInitialize(void)
static void ReplicationSlotDropPtr(ReplicationSlot *slot)
static bool InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, ReplicationSlot *s, XLogRecPtr oldestLSN, Oid dboid, TransactionId snapshotConflictHorizon, bool *invalidated)
void StartupReplicationSlots(void)
void CheckSlotRequirements(void)
static void ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, bool terminating, int pid, NameData slotname, XLogRecPtr restart_lsn, XLogRecPtr oldestLSN, TransactionId snapshotConflictHorizon)
static SyncStandbySlotsConfigData * synchronized_standby_slots_config
#define ReplicationSlotOnDiskConstantSize
Size ReplicationSlotsShmemSize(void)
bool ReplicationSlotValidateName(const char *name, int elevel)
static void ReplicationSlotShmemExit(int code, Datum arg)
ReplicationSlotPersistency
#define SlotIsPhysical(slot)
ReplicationSlotInvalidationCause
#define SlotIsLogical(slot)
bool IsSyncingReplicationSlots(void)
#define SpinLockInit(lock)
#define SpinLockRelease(lock)
#define SpinLockAcquire(lock)
XLogRecPtr LogStandbySnapshot(void)
#define ERRCODE_DUPLICATE_OBJECT
bool pg_str_endswith(const char *str, const char *end)
void appendStringInfo(StringInfo str, const char *fmt,...)
void appendStringInfoString(StringInfo str, const char *s)
void initStringInfo(StringInfo str)
ReplicationSlot replication_slots[1]
ReplicationSlotPersistentData slotdata
TransactionId catalog_xmin
XLogRecPtr confirmed_flush
ReplicationSlotPersistency persistency
ReplicationSlotInvalidationCause invalidated
XLogRecPtr candidate_xmin_lsn
TransactionId effective_catalog_xmin
XLogRecPtr candidate_restart_valid
XLogRecPtr last_saved_confirmed_flush
TransactionId effective_xmin
XLogRecPtr candidate_restart_lsn
LWLock io_in_progress_lock
ConditionVariable active_cv
TransactionId candidate_catalog_xmin
ReplicationSlotPersistentData data
TimestampTz inactive_since
char slot_names[FLEXIBLE_ARRAY_MEMBER]
ConditionVariable wal_confirm_rcv_cv
bool TransactionIdPrecedes(TransactionId id1, TransactionId id2)
bool TransactionIdPrecedesOrEquals(TransactionId id1, TransactionId id2)
#define InvalidTransactionId
#define TransactionIdIsValid(xid)
bool SplitIdentifierString(char *rawstring, char separator, List **namelist)
static void pgstat_report_wait_start(uint32 wait_event_info)
static void pgstat_report_wait_end(void)
bool log_replication_commands
WalSndCtlData * WalSndCtl
bool RecoveryInProgress(void)
XLogSegNo XLogGetLastRemovedSegno(void)
XLogRecPtr GetRedoRecPtr(void)
void XLogSetReplicationSlotMinimumLSN(XLogRecPtr lsn)
XLogRecPtr GetXLogInsertRecPtr(void)
void XLogFlush(XLogRecPtr record)
#define XLogSegNoOffsetToRecPtr(segno, offset, wal_segsz_bytes, dest)
#define XLByteToSeg(xlrp, logSegNo, wal_segsz_bytes)
#define LSN_FORMAT_ARGS(lsn)
#define XLogRecPtrIsInvalid(r)
#define InvalidXLogRecPtr
XLogRecPtr GetXLogReplayRecPtr(TimeLineID *replayTLI)