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\"",
792 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
793 errmsg(
"cannot drop replication slot \"%s\"",
name),
794 errdetail(
"This slot is being synced from the primary server."));
811 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
812 errmsg(
"cannot use %s with a physical replication slot",
813 "ALTER_REPLICATION_SLOT"));
823 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
824 errmsg(
"cannot alter replication slot \"%s\"",
name),
825 errdetail(
"This slot is being synced from the primary server."));
833 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
834 errmsg(
"cannot enable failover for a replication slot"
844 errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
845 errmsg(
"cannot enable failover for a temporary replication slot"));
904 if (rename(path, tmppath) == 0)
932 errmsg(
"could not rename file \"%s\" to \"%s\": %m",
963 if (!
rmtree(tmppath,
true))
965 (
errmsg(
"could not remove directory \"%s\"", tmppath)));
1079 agg_xmin = effective_xmin;
1085 agg_catalog_xmin = effective_catalog_xmin;
1129 restart_lsn < min_required))
1130 min_required = restart_lsn;
1190 restart_lsn < result)
1191 result = restart_lsn;
1212 *nslots = *nactive = 0;
1302 if (active_pid == 0)
1332 (
errcode(ERRCODE_OBJECT_IN_USE),
1333 errmsg(
"replication slot \"%s\" is active for PID %d",
1334 slotname, active_pid)));
1367 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1368 errmsg(
"replication slots can only be used if max_replication_slots > 0")));
1372 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
1373 errmsg(
"replication slots can only be used if wal_level >= replica")));
1384 (
errcode(ERRCODE_INSUFFICIENT_PRIVILEGE),
1385 errmsg(
"permission denied to use replication slots"),
1386 errdetail(
"Only roles with the %s attribute may use replication slots.",
1490 unsigned long long ex = oldestLSN - restart_lsn;
1494 ngettext(
"The slot's restart_lsn %X/%X exceeds the limit by %llu byte.",
1495 "The slot's restart_lsn %X/%X exceeds the limit by %llu bytes.",
1503 snapshotConflictHorizon);
1507 appendStringInfoString(&err_detail,
_(
"Logical decoding on standby requires wal_level >= logical on the primary server."));
1515 errmsg(
"terminating process %d to release replication slot \"%s\"",
1517 errmsg(
"invalidating obsolete replication slot \"%s\"",
1520 hint ?
errhint(
"You might need to increase %s.",
"max_slot_wal_keep_size") : 0);
1545 int last_signaled_pid = 0;
1546 bool released_lock =
false;
1547 bool terminated =
false;
1600 initial_restart_lsn < oldestLSN)
1601 invalidation_cause = cause;
1611 snapshotConflictHorizon))
1612 invalidation_cause = cause;
1615 snapshotConflictHorizon))
1616 invalidation_cause = cause;
1620 invalidation_cause = cause;
1632 invalidation_cause_prev != invalidation_cause));
1651 if (active_pid == 0)
1665 *invalidated =
true;
1678 if (active_pid != 0)
1688 released_lock =
true;
1702 if (last_signaled_pid != active_pid)
1705 slotname, restart_lsn,
1706 oldestLSN, snapshotConflictHorizon);
1713 (
void)
kill(active_pid, SIGTERM);
1715 last_signaled_pid = active_pid;
1717 invalidation_cause_prev = invalidation_cause;
1722 WAIT_EVENT_REPLICATION_SLOT_DROP);
1743 released_lock =
true;
1751 slotname, restart_lsn,
1752 oldestLSN, snapshotConflictHorizon);
1761 return released_lock;
1784 bool invalidated =
false;
1805 snapshotConflictHorizon,
1839 elog(
DEBUG1,
"performing replication slot checkpoint");
1896 DIR *replication_dir;
1897 struct dirent *replication_de;
1899 elog(
DEBUG1,
"starting up replication slots");
1903 while ((replication_de =
ReadDir(replication_dir,
"pg_replslot")) != NULL)
1908 if (strcmp(replication_de->
d_name,
".") == 0 ||
1909 strcmp(replication_de->
d_name,
"..") == 0)
1912 snprintf(path,
sizeof(path),
"pg_replslot/%s", replication_de->
d_name);
1925 (
errmsg(
"could not remove directory \"%s\"",
1984 errmsg(
"could not create directory \"%s\": %m",
1993 if (rename(tmppath, path) != 0)
1996 errmsg(
"could not rename file \"%s\" to \"%s\": %m",
2026 was_dirty = slot->
dirty;
2039 sprintf(tmppath,
"%s/state.tmp", dir);
2040 sprintf(path,
"%s/state", dir);
2051 int save_errno = errno;
2057 errmsg(
"could not create file \"%s\": %m",
2080 if ((
write(
fd, &cp,
sizeof(cp))) !=
sizeof(cp))
2082 int save_errno = errno;
2089 errno = save_errno ? save_errno : ENOSPC;
2092 errmsg(
"could not write to file \"%s\": %m",
2102 int save_errno = errno;
2110 errmsg(
"could not fsync file \"%s\": %m",
2118 int save_errno = errno;
2124 errmsg(
"could not close file \"%s\": %m",
2130 if (rename(tmppath, path) != 0)
2132 int save_errno = errno;
2138 errmsg(
"could not rename file \"%s\" to \"%s\": %m",
2160 slot->
dirty =
false;
2178 bool restored =
false;
2186 sprintf(path,
"%s/state.tmp", slotdir);
2187 if (unlink(path) < 0 && errno != ENOENT)
2190 errmsg(
"could not remove file \"%s\": %m", path)));
2192 sprintf(path,
"%s/state", slotdir);
2194 elog(
DEBUG1,
"restoring replication slot from \"%s\"", path);
2206 errmsg(
"could not open file \"%s\": %m", path)));
2216 errmsg(
"could not fsync file \"%s\": %m",
2234 errmsg(
"could not read file \"%s\": %m", path)));
2238 errmsg(
"could not read file \"%s\": read %d of %zu",
2247 errmsg(
"replication slot file \"%s\" has wrong magic number: %u instead of %u",
2254 errmsg(
"replication slot file \"%s\" has unsupported version %u",
2261 errmsg(
"replication slot file \"%s\" has corrupted length %u",
2270 if (readBytes != cp.
length)
2275 errmsg(
"could not read file \"%s\": %m", path)));
2279 errmsg(
"could not read file \"%s\": read %d of %zu",
2286 errmsg(
"could not close file \"%s\": %m", path)));
2297 (
errmsg(
"checksum mismatch for replication slot file \"%s\": is %u, should be %u",
2306 if (!
rmtree(slotdir,
true))
2309 (
errmsg(
"could not remove directory \"%s\"",
2330 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2331 errmsg(
"logical replication slot \"%s\" exists, but wal_level < logical",
2333 errhint(
"Change wal_level to be logical or higher.")));
2336 (
errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2337 errmsg(
"physical replication slot \"%s\" exists, but wal_level < replica",
2339 errhint(
"Change wal_level to be replica or higher.")));
2381 (
errmsg(
"too many replication slots active before shutdown"),
2382 errhint(
"Increase max_replication_slots and try again.")));
2396 Assert(invalidation_reason);
2485 if ((*
newval)[0] ==
'\0')
2494 if (!ok || elemlist ==
NIL)
2504 size += strlen(slot_name) + 1;
2515 strcpy(ptr, slot_name);
2516 ptr += strlen(slot_name) + 1;
2519 *extra = (
void *) config;
2547 const char *standby_slot_name;
2561 if (strcmp(standby_slot_name, slot_name) == 0)
2564 standby_slot_name += strlen(standby_slot_name) + 1;
2581 int caught_up_slot_num = 0;
2634 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2635 errmsg(
"replication slot \"%s\" specified in parameter %s does not exist",
2636 name,
"standby_slot_names"),
2637 errdetail(
"Logical replication is waiting on the standby associated with \"%s\".",
2639 errhint(
"Consider creating the slot \"%s\" or amend parameter %s.",
2640 name,
"standby_slot_names"));
2655 errcode(ERRCODE_INVALID_PARAMETER_VALUE),
2656 errmsg(
"cannot have logical replication slot \"%s\" in parameter %s",
2657 name,
"standby_slot_names"),
2658 errdetail(
"Logical replication is waiting for correction on \"%s\".",
2660 errhint(
"Consider removing logical slot \"%s\" from parameter %s.",
2661 name,
"standby_slot_names"));
2675 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2676 errmsg(
"physical slot \"%s\" specified in parameter %s has been invalidated",
2677 name,
"standby_slot_names"),
2678 errdetail(
"Logical replication is waiting on the standby associated with \"%s\".",
2680 errhint(
"Consider dropping and recreating the slot \"%s\" or amend parameter %s.",
2681 name,
"standby_slot_names"));
2690 errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
2691 errmsg(
"replication slot \"%s\" specified in parameter %s does not have active_pid",
2692 name,
"standby_slot_names"),
2693 errdetail(
"Logical replication is waiting on the standby associated with \"%s\".",
2695 errhint(
"Consider starting standby associated with \"%s\" or amend parameter %s.",
2696 name,
"standby_slot_names"));
2702 Assert(restart_lsn >= wait_for_lsn);
2705 min_restart_lsn > restart_lsn)
2706 min_restart_lsn = restart_lsn;
2708 caught_up_slot_num++;
2770 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,...)
struct dirent * ReadDir(DIR *dir, const char *dirname)
int MakePGDirectory(const char *directoryName)
int CloseTransientFile(int fd)
void fsync_fname(const char *fname, bool isdir)
int OpenTransientFile(const char *fileName, int fileFlags)
DIR * AllocateDir(const char *dirname)
PGFileType get_dirent_type(const char *path, const struct dirent *de, bool look_through_symlinks, int elevel)
void * guc_malloc(int elevel, size_t size)
#define GUC_check_errdetail
void ProcessConfigFile(GucContext context)
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)
#define PROC_IN_LOGICAL_DECODING
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)
void * ShmemInitStruct(const char *name, Size size, bool *foundPtr)
Size mul_size(Size s1, Size s2)
static pg_noinline void Size size
ReplicationSlot * SearchNamedReplicationSlot(const char *name, bool need_lock)
void ReplicationSlotAlter(const char *name, bool failover)
int ReplicationSlotIndex(ReplicationSlot *slot)
#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 ReplicationSlotCleanup(void)
void ReplicationSlotDropAcquired(void)
void ReplicationSlotMarkDirty(void)
void ReplicationSlotReserveWal(void)
char * standby_slot_names
bool ReplicationSlotsCountDBSlots(Oid dboid, int *nslots, int *nactive)
void ReplicationSlotAcquire(const char *name, bool nowait)
bool SlotExistsInStandbySlotNames(const char *slot_name)
bool InvalidateObsoleteReplicationSlots(ReplicationSlotInvalidationCause cause, XLogSegNo oldestSegno, Oid dboid, TransactionId snapshotConflictHorizon)
static bool validate_standby_slots(char *rawname, List **elemlist)
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)
void ReplicationSlotSave(void)
static void CreateSlotOnDisk(ReplicationSlot *slot)
#define ReplicationSlotOnDiskV2Size
void CheckSlotPermissions(void)
bool ReplicationSlotName(int index, Name name)
void ReplicationSlotsShmemInit(void)
const char *const SlotInvalidationCauses[]
static StandbySlotNamesConfigData * standby_slot_names_config
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 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)
void assign_standby_slot_names(const char *newval, void *extra)
bool check_standby_slot_names(char **newval, void **extra, GucSource source)
static void ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, bool terminating, int pid, NameData slotname, XLogRecPtr restart_lsn, XLogRecPtr oldestLSN, TransactionId snapshotConflictHorizon)
#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)