public inbox for [email protected]help / color / mirror / Atom feed
[PATCH v8 1/4] Move callback-call from ReadPageInternal to XLogReadRecord. 18+ messages / 7 participants [nested] [flat]
* [PATCH v8 1/4] Move callback-call from ReadPageInternal to XLogReadRecord. @ 2019-09-05 11:21 Kyotaro Horiguchi <[email protected]> 0 siblings, 0 replies; 18+ messages in thread From: Kyotaro Horiguchi @ 2019-09-05 11:21 UTC (permalink / raw) The current WAL record reader reads page data using a call back function. Although it is not so problematic alone, it would be a problem if we are going to do add tasks like encryption which is performed on page data before WAL reader reads them. To avoid that the record reader facility has to have a new code path corresponds to every new callback, this patch separates page reader from WAL record reading facility by modifying the current WAL record reader to a state machine. As the first step of that change, this patch moves the page reader function out of ReadPageInternal, then the remaining tasks of the function are taken over by the new function XLogNeedData. As the result XLogPageRead directly calls the page reader callback function according to the feedback from XLogNeedData. --- src/backend/access/transam/xlog.c | 16 +- src/backend/access/transam/xlogreader.c | 306 +++++++++++++++---------- src/backend/access/transam/xlogutils.c | 10 +- src/backend/replication/logical/logicalfuncs.c | 2 +- src/backend/replication/walsender.c | 10 +- src/bin/pg_rewind/parsexlog.c | 16 +- src/bin/pg_waldump/pg_waldump.c | 8 +- src/include/access/xlogreader.h | 23 +- src/include/access/xlogutils.h | 2 +- src/include/replication/logicalfuncs.h | 2 +- src/test/recovery/t/011_crash_recovery.pl | 1 + 11 files changed, 239 insertions(+), 157 deletions(-) diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 6c69eb6dd7..5dcb2e500c 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -884,7 +884,7 @@ static bool InstallXLogFileSegment(XLogSegNo *segno, char *tmppath, static int XLogFileRead(XLogSegNo segno, int emode, TimeLineID tli, int source, bool notfoundOk); static int XLogFileReadAnyTLI(XLogSegNo segno, int emode, int source); -static int XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, +static bool XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *readBuf); static bool WaitForWALToBecomeAvailable(XLogRecPtr RecPtr, bool randAccess, bool fetching_ckpt, XLogRecPtr tliRecPtr); @@ -4249,7 +4249,6 @@ ReadRecord(XLogReaderState *xlogreader, XLogRecPtr RecPtr, int emode, XLogRecord *record; XLogPageReadPrivate *private = (XLogPageReadPrivate *) xlogreader->private_data; - /* Pass through parameters to XLogPageRead */ private->fetching_ckpt = fetching_ckpt; private->emode = emode; private->randAccess = (RecPtr != InvalidXLogRecPtr); @@ -11522,7 +11521,7 @@ CancelBackup(void) * XLogPageRead() to try fetching the record from another source, or to * sleep and retry. */ -static int +static bool XLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *readBuf) { @@ -11581,7 +11580,8 @@ retry: readLen = 0; readSource = 0; - return -1; + xlogreader->readLen = -1; + return false; } } @@ -11676,7 +11676,8 @@ retry: goto next_record_is_invalid; } - return readLen; + xlogreader->readLen = readLen; + return true; next_record_is_invalid: lastSourceFailed = true; @@ -11690,8 +11691,9 @@ next_record_is_invalid: /* In standby-mode, keep trying */ if (StandbyMode) goto retry; - else - return -1; + + xlogreader->readLen = -1; + return false; } /* diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.c index 27c27303d6..900a628752 100644 --- a/src/backend/access/transam/xlogreader.c +++ b/src/backend/access/transam/xlogreader.c @@ -34,8 +34,8 @@ static void report_invalid_record(XLogReaderState *state, const char *fmt,...) pg_attribute_printf(2, 3); static bool allocate_recordbuf(XLogReaderState *state, uint32 reclength); -static int ReadPageInternal(XLogReaderState *state, XLogRecPtr pageptr, - int reqLen); +static bool XLogNeedData(XLogReaderState *state, XLogRecPtr pageptr, + int reqLen, bool header_inclusive); static void XLogReaderInvalReadState(XLogReaderState *state); static bool ValidXLogRecordHeader(XLogReaderState *state, XLogRecPtr RecPtr, XLogRecPtr PrevRecPtr, XLogRecord *record, bool randAccess); @@ -104,7 +104,7 @@ XLogReaderAllocate(int wal_segment_size, const char *waldir, /* system_identifier initialized to zeroes above */ state->private_data = private_data; /* ReadRecPtr and EndRecPtr initialized to zeroes above */ - /* readSegNo, readOff, readLen, readPageTLI initialized to zeroes above */ + /* readSegNo, readLen, readPageTLI initialized to zeroes above */ state->errormsg_buf = palloc_extended(MAX_ERRORMSG_LEN + 1, MCXT_ALLOC_NO_OOM); if (!state->errormsg_buf) @@ -245,7 +245,6 @@ XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg) uint32 targetRecOff; uint32 pageHeaderSize; bool gotheader; - int readOff; /* * randAccess indicates whether to verify the previous-record pointer of @@ -297,15 +296,20 @@ XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg) * byte to cover the whole record header, or at least the part of it that * fits on the same page. */ - readOff = ReadPageInternal(state, - targetPagePtr, - Min(targetRecOff + SizeOfXLogRecord, XLOG_BLCKSZ)); - if (readOff < 0) + while (XLogNeedData(state, targetPagePtr, + Min(targetRecOff + SizeOfXLogRecord, XLOG_BLCKSZ), + targetRecOff != 0)) + { + if (!state->read_page(state, state->readPagePtr, state->readLen, + RecPtr, state->readBuf)) + break; + } + + if (!state->page_verified) goto err; /* - * ReadPageInternal always returns at least the page header, so we can - * examine it now. + * We have at least the page header, so we can examine it now. */ pageHeaderSize = XLogPageHeaderSize((XLogPageHeader) state->readBuf); if (targetRecOff == 0) @@ -331,8 +335,8 @@ XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg) goto err; } - /* ReadPageInternal has verified the page header */ - Assert(pageHeaderSize <= readOff); + /* XLogNeedData has verified the page header */ + Assert(pageHeaderSize <= state->readLen); /* * Read the record length. @@ -405,18 +409,25 @@ XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg) do { + int rest_len = total_len - gotlen; + /* Calculate pointer to beginning of next page */ targetPagePtr += XLOG_BLCKSZ; /* Wait for the next page to become available */ - readOff = ReadPageInternal(state, targetPagePtr, - Min(total_len - gotlen + SizeOfXLogShortPHD, - XLOG_BLCKSZ)); + while (XLogNeedData(state, targetPagePtr, + Min(rest_len, XLOG_BLCKSZ), + false)) + { + if (!state->read_page(state, state->readPagePtr, state->readLen, + state->ReadRecPtr, state->readBuf)) + break; + } - if (readOff < 0) + if (!state->page_verified) goto err; - Assert(SizeOfXLogShortPHD <= readOff); + Assert(SizeOfXLogShortPHD <= state->readLen); /* Check that the continuation on next page looks valid */ pageHeader = (XLogPageHeader) state->readBuf; @@ -445,21 +456,14 @@ XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg) /* Append the continuation from this page to the buffer */ pageHeaderSize = XLogPageHeaderSize(pageHeader); - if (readOff < pageHeaderSize) - readOff = ReadPageInternal(state, targetPagePtr, - pageHeaderSize); - - Assert(pageHeaderSize <= readOff); + Assert (pageHeaderSize <= state->readLen); contdata = (char *) state->readBuf + pageHeaderSize; len = XLOG_BLCKSZ - pageHeaderSize; if (pageHeader->xlp_rem_len < len) len = pageHeader->xlp_rem_len; - if (readOff < pageHeaderSize + len) - readOff = ReadPageInternal(state, targetPagePtr, - pageHeaderSize + len); - + Assert (pageHeaderSize + len <= state->readLen); memcpy(buffer, (char *) contdata, len); buffer += len; gotlen += len; @@ -489,9 +493,15 @@ XLogReadRecord(XLogReaderState *state, XLogRecPtr RecPtr, char **errormsg) else { /* Wait for the record data to become available */ - readOff = ReadPageInternal(state, targetPagePtr, - Min(targetRecOff + total_len, XLOG_BLCKSZ)); - if (readOff < 0) + while (XLogNeedData(state, targetPagePtr, + Min(targetRecOff + total_len, XLOG_BLCKSZ), true)) + { + if (!state->read_page(state, state->readPagePtr, state->readLen, + state->ReadRecPtr, state->readBuf)) + break; + } + + if (!state->page_verified) goto err; /* Record does not cross a page boundary */ @@ -534,109 +544,158 @@ err: } /* - * Read a single xlog page including at least [pageptr, reqLen] of valid data - * via the read_page() callback. + * Checks that an xlog page loaded in state->readBuf is including at least + * [pageptr, reqLen] and the page is valid. header_inclusive indicates that + * reqLen is calculated including page header length. * - * Returns -1 if the required page cannot be read for some reason; errormsg_buf - * is set in that case (unless the error occurs in the read_page callback). + * Returns false if the buffer already contains the requested data, or found + * error. state->page_verified is set to true for the former and false for the + * latter. * - * We fetch the page from a reader-local cache if we know we have the required - * data and if there hasn't been any error since caching the data. + * Otherwise returns true and requests data loaded onto state->readBuf by + * state->readPagePtr and state->readLen. The caller shall call this function + * again after filling the buffer at least with that portion of data and set + * state->readLen to the length of actually loaded data. + * + * If header_inclusive is false, corrects reqLen internally by adding the + * actual page header length and may request caller for new data. */ -static int -ReadPageInternal(XLogReaderState *state, XLogRecPtr pageptr, int reqLen) +static bool +XLogNeedData(XLogReaderState *state, XLogRecPtr pageptr, int reqLen, + bool header_inclusive) { - int readLen; uint32 targetPageOff; XLogSegNo targetSegNo; - XLogPageHeader hdr; - - Assert((pageptr % XLOG_BLCKSZ) == 0); - - XLByteToSeg(pageptr, targetSegNo, state->segcxt.ws_segsize); - targetPageOff = XLogSegmentOffset(pageptr, state->segcxt.ws_segsize); + uint32 addLen = 0; /* check whether we have all the requested data already */ - if (targetSegNo == state->seg.ws_segno && - targetPageOff == state->seg.ws_off && reqLen <= state->readLen) - return state->readLen; + if (state->page_verified && pageptr == state->readPagePtr) + { + if (!header_inclusive) + { + /* + * calculate additional length for page header so that the total + * length doesn't exceed the block size. + */ + uint32 pageHeaderSize = + XLogPageHeaderSize((XLogPageHeader) state->readBuf); + + addLen = pageHeaderSize; + if (reqLen + pageHeaderSize <= XLOG_BLCKSZ) + addLen = pageHeaderSize; + else + addLen = XLOG_BLCKSZ - reqLen; + } + + if (reqLen + addLen <= state->readLen) + return false; + } + + if (!state->page_verified && + !XLogRecPtrIsInvalid(state->readPagePtr) && state->readLen >= 0) + { + uint32 pageHeaderSize; + + /* just loaded new data so needs to verify page header */ + + /* The caller must have loaded at least page header */ + Assert (state->readLen >= SizeOfXLogShortPHD); + + /* + * We have enough data to check the header length. Recheck the loaded + * length if it is a long header if any. + */ + pageHeaderSize = XLogPageHeaderSize((XLogPageHeader) state->readBuf); + + /* Request more data if we don't have the full header. */ + if (state->readLen < pageHeaderSize) + { + state->readLen = pageHeaderSize; + return true; + } + + /* Now that we know we have the full header, validate it. */ + if (!XLogReaderValidatePageHeader(state, state->readPagePtr, + (char *) state->readBuf)) + { + /* That's bad. Force reading the page again. */ + XLogReaderInvalReadState(state); + + return false; + } + + state->page_verified = true; + + XLByteToSeg(state->readPagePtr, state->seg.ws_segno, + state->segcxt.ws_segsize); + + /* + * The loaded page may not be the one caller is supposing to read when + * we are verifying the first page of new segment. In that case, skip + * further verification and immediately load the target page. + */ + if (pageptr == state->readPagePtr) + { + + /* + * calculate additional length for page header keeping the total + * length within the block size. + */ + if (!header_inclusive) + { + addLen = pageHeaderSize; + if (reqLen + pageHeaderSize <= XLOG_BLCKSZ) + addLen = pageHeaderSize; + else + addLen = XLOG_BLCKSZ - reqLen; + + Assert(addLen >= 0); + } + + /* Return if we already have it. */ + if (reqLen + addLen <= state->readLen) + return false; + } + } + + /* Data is not in our buffer, request the caller for it. */ + XLByteToSeg(pageptr, targetSegNo, state->segcxt.ws_segsize); + targetPageOff = XLogSegmentOffset(pageptr, state->segcxt.ws_segsize); + Assert((pageptr % XLOG_BLCKSZ) == 0); + + /* + * Every time we request to load new data of a page to the caller, even if + * we looked at a part of it before, we need to do verification on the next + * invocation as the caller might now be rereading data from a different + * source. + */ + state->page_verified = false; /* - * Data is not in our buffer. - * - * Every time we actually read the page, even if we looked at parts of it - * before, we need to do verification as the read_page callback might now - * be rereading data from a different source. - * * Whenever switching to a new WAL segment, we read the first page of the * file and validate its header, even if that's not where the target * record is. This is so that we can check the additional identification * info that is present in the first page's "long" header. + * Don't do this if the caller requested the first page in the segment. */ if (targetSegNo != state->seg.ws_segno && targetPageOff != 0) { - XLogRecPtr targetSegmentPtr = pageptr - targetPageOff; - - readLen = state->read_page(state, targetSegmentPtr, XLOG_BLCKSZ, - state->currRecPtr, - state->readBuf); - if (readLen < 0) - goto err; - - /* we can be sure to have enough WAL available, we scrolled back */ - Assert(readLen == XLOG_BLCKSZ); - - if (!XLogReaderValidatePageHeader(state, targetSegmentPtr, - state->readBuf)) - goto err; + /* + * Then we'll see that the targetSegNo now matches the ws_segno, and + * will not come back here, but will request the actual target page. + */ + state->readPagePtr = pageptr - targetPageOff; + state->readLen = XLOG_BLCKSZ; + return true; } /* - * First, read the requested data length, but at least a short page header - * so that we can validate it. + * Request the caller to load the page. We need at least a short page + * header so that we can validate it. */ - readLen = state->read_page(state, pageptr, Max(reqLen, SizeOfXLogShortPHD), - state->currRecPtr, - state->readBuf); - if (readLen < 0) - goto err; - - Assert(readLen <= XLOG_BLCKSZ); - - /* Do we have enough data to check the header length? */ - if (readLen <= SizeOfXLogShortPHD) - goto err; - - Assert(readLen >= reqLen); - - hdr = (XLogPageHeader) state->readBuf; - - /* still not enough */ - if (readLen < XLogPageHeaderSize(hdr)) - { - readLen = state->read_page(state, pageptr, XLogPageHeaderSize(hdr), - state->currRecPtr, - state->readBuf); - if (readLen < 0) - goto err; - } - - /* - * Now that we know we have the full header, validate it. - */ - if (!XLogReaderValidatePageHeader(state, pageptr, (char *) hdr)) - goto err; - - /* update read state information */ - state->seg.ws_segno = targetSegNo; - state->seg.ws_off = targetPageOff; - state->readLen = readLen; - - return readLen; - -err: - XLogReaderInvalReadState(state); - return -1; + state->readPagePtr = pageptr; + state->readLen = Max(reqLen + addLen, SizeOfXLogShortPHD); + return true; } /* @@ -645,9 +704,7 @@ err: static void XLogReaderInvalReadState(XLogReaderState *state) { - state->seg.ws_segno = 0; - state->seg.ws_off = 0; - state->readLen = 0; + state->readPagePtr = InvalidXLogRecPtr; } /* @@ -925,7 +982,6 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr) XLogRecPtr targetPagePtr; int targetRecOff; uint32 pageHeaderSize; - int readLen; /* * Compute targetRecOff. It should typically be equal or greater than @@ -933,7 +989,7 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr) * that, except when caller has explicitly specified the offset that * falls somewhere there or when we are skipping multi-page * continuation record. It doesn't matter though because - * ReadPageInternal() is prepared to handle that and will read at + * CheckPage() is prepared to handle that and will read at * least short page-header worth of data */ targetRecOff = tmpRecPtr % XLOG_BLCKSZ; @@ -941,19 +997,23 @@ XLogFindNextRecord(XLogReaderState *state, XLogRecPtr RecPtr) /* scroll back to page boundary */ targetPagePtr = tmpRecPtr - targetRecOff; - /* Read the page containing the record */ - readLen = ReadPageInternal(state, targetPagePtr, targetRecOff); - if (readLen < 0) + while(XLogNeedData(state, targetPagePtr, targetRecOff, + targetRecOff != 0)) + { + if (!state->read_page(state, state->readPagePtr, state->readLen, + state->ReadRecPtr, state->readBuf)) + break; + } + + if (!state->page_verified) goto err; header = (XLogPageHeader) state->readBuf; pageHeaderSize = XLogPageHeaderSize(header); - /* make sure we have enough data for the page header */ - readLen = ReadPageInternal(state, targetPagePtr, pageHeaderSize); - if (readLen < 0) - goto err; + /* we should have read the page header */ + Assert (state->readLen >= pageHeaderSize); /* skip over potential continuation data */ if (header->xlp_info & XLP_FIRST_IS_CONTRECORD) diff --git a/src/backend/access/transam/xlogutils.c b/src/backend/access/transam/xlogutils.c index 5f1e5ba75d..a19726a96e 100644 --- a/src/backend/access/transam/xlogutils.c +++ b/src/backend/access/transam/xlogutils.c @@ -803,7 +803,7 @@ void XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wantLength) { const XLogRecPtr lastReadPage = state->seg.ws_segno * - state->segcxt.ws_segsize + state->seg.ws_off; + state->segcxt.ws_segsize + state->readLen; Assert(wantPage != InvalidXLogRecPtr && wantPage % XLOG_BLCKSZ == 0); Assert(wantLength <= XLOG_BLCKSZ); @@ -907,7 +907,7 @@ XLogReadDetermineTimeline(XLogReaderState *state, XLogRecPtr wantPage, uint32 wa * exists for normal backends, so we have to do a check/sleep/repeat style of * loop for now. */ -int +bool read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page) { @@ -1007,7 +1007,8 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, else if (targetPagePtr + reqLen > read_upto) { /* not enough data there */ - return -1; + state->readLen = -1; + return false; } else { @@ -1024,5 +1025,6 @@ read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, XLOG_BLCKSZ); /* number of valid bytes in the buffer */ - return count; + state->readLen = count; + return true; } diff --git a/src/backend/replication/logical/logicalfuncs.c b/src/backend/replication/logical/logicalfuncs.c index d1cf80d441..310cd9d8cf 100644 --- a/src/backend/replication/logical/logicalfuncs.c +++ b/src/backend/replication/logical/logicalfuncs.c @@ -114,7 +114,7 @@ check_permissions(void) (errmsg("must be superuser or replication role to use replication slots")))); } -int +bool logical_read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page) { diff --git a/src/backend/replication/walsender.c b/src/backend/replication/walsender.c index eb4a98cc91..0809ceaeb8 100644 --- a/src/backend/replication/walsender.c +++ b/src/backend/replication/walsender.c @@ -760,7 +760,7 @@ StartReplication(StartReplicationCmd *cmd) * which has to do a plain sleep/busy loop, because the walsender's latch gets * set every time WAL is flushed. */ -static int +static bool logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page) { @@ -778,7 +778,10 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req /* fail if not (implies we are going to shut down) */ if (flushptr < targetPagePtr + reqLen) - return -1; + { + state->readLen = -1; + return false; + } if (targetPagePtr + XLOG_BLCKSZ <= flushptr) count = XLOG_BLCKSZ; /* more than one block available */ @@ -788,7 +791,8 @@ logical_read_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int req /* now actually read the data, we know it's there */ XLogRead(sendCxt, cur_page, targetPagePtr, XLOG_BLCKSZ); - return count; + state->readLen = count; + return true; } /* diff --git a/src/bin/pg_rewind/parsexlog.c b/src/bin/pg_rewind/parsexlog.c index 264a8f4db5..8aecd1adc7 100644 --- a/src/bin/pg_rewind/parsexlog.c +++ b/src/bin/pg_rewind/parsexlog.c @@ -46,7 +46,7 @@ typedef struct XLogPageReadPrivate int tliIndex; } XLogPageReadPrivate; -static int SimpleXLogPageRead(XLogReaderState *xlogreader, +static bool SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *readBuf); @@ -230,7 +230,7 @@ findLastCheckpoint(const char *datadir, XLogRecPtr forkptr, int tliIndex, } /* XLogReader callback function, to read a WAL page */ -static int +static bool SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *readBuf) { @@ -285,7 +285,8 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, if (xlogreadfd < 0) { pg_log_error("could not open file \"%s\": %m", xlogfpath); - return -1; + xlogreader->readLen = -1; + return false; } } @@ -298,7 +299,8 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, if (lseek(xlogreadfd, (off_t) targetPageOff, SEEK_SET) < 0) { pg_log_error("could not seek in file \"%s\": %m", xlogfpath); - return -1; + xlogreader->readLen = -1; + return false; } @@ -311,13 +313,15 @@ SimpleXLogPageRead(XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, pg_log_error("could not read file \"%s\": read %d of %zu", xlogfpath, r, (Size) XLOG_BLCKSZ); - return -1; + xlogreader->readLen = -1; + return false; } Assert(targetSegNo == xlogreadsegno); xlogreader->seg.ws_tli = targetHistory[private->tliIndex].tli; - return XLOG_BLCKSZ; + xlogreader->readLen = XLOG_BLCKSZ; + return true; } /* diff --git a/src/bin/pg_waldump/pg_waldump.c b/src/bin/pg_waldump/pg_waldump.c index b79208cd73..6e424bd8e1 100644 --- a/src/bin/pg_waldump/pg_waldump.c +++ b/src/bin/pg_waldump/pg_waldump.c @@ -406,7 +406,7 @@ XLogDumpXLogRead(const char *directory, TimeLineID timeline_id, /* * XLogReader read_page callback */ -static int +static bool XLogDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetPtr, char *readBuff) { @@ -422,14 +422,16 @@ XLogDumpReadPage(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, else { private->endptr_reached = true; - return -1; + state->readLen = -1; + return false; } } XLogDumpXLogRead(state->segcxt.ws_dir, private->timeline, targetPagePtr, readBuff, count); - return count; + state->readLen = count; + return true; } /* diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.h index 1bbee386e8..8b747d465f 100644 --- a/src/include/access/xlogreader.h +++ b/src/include/access/xlogreader.h @@ -50,7 +50,7 @@ typedef struct WALSegmentContext typedef struct XLogReaderState XLogReaderState; /* Function type definition for the read_page callback */ -typedef int (*XLogPageReadCB) (XLogReaderState *xlogreader, +typedef bool (*XLogPageReadCB) (XLogReaderState *xlogreader, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, @@ -132,6 +132,20 @@ struct XLogReaderState XLogRecPtr ReadRecPtr; /* start of last record read */ XLogRecPtr EndRecPtr; /* end+1 of last record read */ + /* ---------------------------------------- + * Communication with page reader + * readBuf is XLOG_BLCKSZ bytes, valid up to at least readLen bytes. + * ---------------------------------------- + */ + /* variables to communicate with page reader */ + XLogRecPtr readPagePtr; /* page pointer to read */ + int32 readLen; /* bytes requested to reader, or actual bytes + * read by reader, which must be larger than + * the request, or -1 on error */ + TimeLineID readPageTLI; /* TLI for data currently in readBuf */ + char *readBuf; /* buffer to store data */ + bool page_verified; /* is the page on the buffer verified? */ + /* ---------------------------------------- * Decoded representation of current record @@ -158,13 +172,6 @@ struct XLogReaderState * ---------------------------------------- */ - /* - * Buffer for currently read page (XLOG_BLCKSZ bytes, valid up to at least - * readLen bytes) - */ - char *readBuf; - uint32 readLen; - /* last read XLOG position for data currently in readBuf */ WALSegmentContext segcxt; WALOpenSegment seg; diff --git a/src/include/access/xlogutils.h b/src/include/access/xlogutils.h index 2df98e45b2..47b65463f9 100644 --- a/src/include/access/xlogutils.h +++ b/src/include/access/xlogutils.h @@ -47,7 +47,7 @@ extern Buffer XLogReadBufferExtended(RelFileNode rnode, ForkNumber forknum, extern Relation CreateFakeRelcacheEntry(RelFileNode rnode); extern void FreeFakeRelcacheEntry(Relation fakerel); -extern int read_local_xlog_page(XLogReaderState *state, +extern bool read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page); diff --git a/src/include/replication/logicalfuncs.h b/src/include/replication/logicalfuncs.h index 012096f183..54291221c3 100644 --- a/src/include/replication/logicalfuncs.h +++ b/src/include/replication/logicalfuncs.h @@ -11,7 +11,7 @@ #include "replication/logical.h" -extern int logical_read_local_xlog_page(XLogReaderState *state, +extern bool logical_read_local_xlog_page(XLogReaderState *state, XLogRecPtr targetPagePtr, int reqLen, XLogRecPtr targetRecPtr, char *cur_page); diff --git a/src/test/recovery/t/011_crash_recovery.pl b/src/test/recovery/t/011_crash_recovery.pl index 526a3481fb..c78912571e 100644 --- a/src/test/recovery/t/011_crash_recovery.pl +++ b/src/test/recovery/t/011_crash_recovery.pl @@ -55,6 +55,7 @@ is($node->safe_psql('postgres', qq[SELECT txid_status('$xid');]), # Crash and restart the postmaster $node->stop('immediate'); +print "HOGEEEEEEEEEEEE\n"; $node->start; # Make sure we really got a new xid -- 2.16.3 ----Next_Part(Fri_Sep_27_12_07_26_2019_308)-- Content-Type: Text/X-Patch; charset=us-ascii Content-Transfer-Encoding: 7bit Content-Disposition: inline; filename="v8-0002-Move-page-reader-out-of-XLogReadRecord.patch" ^ permalink raw reply [nested|flat] 18+ messages in thread
* [PATCH v9 3/4] add support for syncfs in frontend support functions @ 2023-08-31 14:59 Nathan Bossart <[email protected]> 0 siblings, 0 replies; 18+ messages in thread From: Nathan Bossart @ 2023-08-31 14:59 UTC (permalink / raw) --- src/bin/initdb/initdb.c | 5 +- src/bin/pg_basebackup/pg_basebackup.c | 5 +- src/bin/pg_checksums/pg_checksums.c | 3 +- src/bin/pg_dump/pg_backup.h | 4 +- src/bin/pg_dump/pg_backup_archiver.c | 14 ++- src/bin/pg_dump/pg_backup_archiver.h | 1 + src/bin/pg_dump/pg_backup_directory.c | 2 +- src/bin/pg_dump/pg_dump.c | 3 +- src/bin/pg_rewind/file_ops.c | 2 +- src/bin/pg_rewind/pg_rewind.c | 1 + src/bin/pg_rewind/pg_rewind.h | 2 + src/common/file_utils.c | 170 +++++++++++++++++++++----- src/include/common/file_utils.h | 5 +- 13 files changed, 172 insertions(+), 45 deletions(-) diff --git a/src/bin/initdb/initdb.c b/src/bin/initdb/initdb.c index 905b979947..bbea1e412b 100644 --- a/src/bin/initdb/initdb.c +++ b/src/bin/initdb/initdb.c @@ -165,6 +165,7 @@ static bool show_setting = false; static bool data_checksums = false; static char *xlog_dir = NULL; static int wal_segment_size_mb = (DEFAULT_XLOG_SEG_SIZE) / (1024 * 1024); +static DataDirSyncMethod sync_method = DATA_DIR_SYNC_METHOD_FSYNC; /* internal vars */ @@ -3333,7 +3334,7 @@ main(int argc, char *argv[]) fputs(_("syncing data to disk ... "), stdout); fflush(stdout); - fsync_pgdata(pg_data, PG_VERSION_NUM); + fsync_pgdata(pg_data, PG_VERSION_NUM, sync_method); check_ok(); return 0; } @@ -3396,7 +3397,7 @@ main(int argc, char *argv[]) { fputs(_("syncing data to disk ... "), stdout); fflush(stdout); - fsync_pgdata(pg_data, PG_VERSION_NUM); + fsync_pgdata(pg_data, PG_VERSION_NUM, sync_method); check_ok(); } else diff --git a/src/bin/pg_basebackup/pg_basebackup.c b/src/bin/pg_basebackup/pg_basebackup.c index 1dc8efe0cb..1a6eacf6d5 100644 --- a/src/bin/pg_basebackup/pg_basebackup.c +++ b/src/bin/pg_basebackup/pg_basebackup.c @@ -148,6 +148,7 @@ static bool verify_checksums = true; static bool manifest = true; static bool manifest_force_encode = false; static char *manifest_checksums = NULL; +static DataDirSyncMethod sync_method = DATA_DIR_SYNC_METHOD_FSYNC; static bool success = false; static bool made_new_pgdata = false; @@ -2199,11 +2200,11 @@ BaseBackup(char *compression_algorithm, char *compression_detail, if (format == 't') { if (strcmp(basedir, "-") != 0) - (void) fsync_dir_recurse(basedir); + (void) fsync_dir_recurse(basedir, sync_method); } else { - (void) fsync_pgdata(basedir, serverVersion); + (void) fsync_pgdata(basedir, serverVersion, sync_method); } } diff --git a/src/bin/pg_checksums/pg_checksums.c b/src/bin/pg_checksums/pg_checksums.c index 9011a19b4e..123450f483 100644 --- a/src/bin/pg_checksums/pg_checksums.c +++ b/src/bin/pg_checksums/pg_checksums.c @@ -44,6 +44,7 @@ static char *only_filenode = NULL; static bool do_sync = true; static bool verbose = false; static bool showprogress = false; +static DataDirSyncMethod sync_method = DATA_DIR_SYNC_METHOD_FSYNC; typedef enum { @@ -623,7 +624,7 @@ main(int argc, char *argv[]) if (do_sync) { pg_log_info("syncing data directory"); - fsync_pgdata(DataDir, PG_VERSION_NUM); + fsync_pgdata(DataDir, PG_VERSION_NUM, sync_method); } pg_log_info("updating control file"); diff --git a/src/bin/pg_dump/pg_backup.h b/src/bin/pg_dump/pg_backup.h index aba780ef4b..3a57cdd97d 100644 --- a/src/bin/pg_dump/pg_backup.h +++ b/src/bin/pg_dump/pg_backup.h @@ -24,6 +24,7 @@ #define PG_BACKUP_H #include "common/compression.h" +#include "common/file_utils.h" #include "fe_utils/simple_list.h" #include "libpq-fe.h" @@ -307,7 +308,8 @@ extern Archive *OpenArchive(const char *FileSpec, const ArchiveFormat fmt); extern Archive *CreateArchive(const char *FileSpec, const ArchiveFormat fmt, const pg_compress_specification compression_spec, bool dosync, ArchiveMode mode, - SetupWorkerPtrType setupDumpWorker); + SetupWorkerPtrType setupDumpWorker, + DataDirSyncMethod sync_method); /* The --list option */ extern void PrintTOCSummary(Archive *AHX); diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c index 39ebcfec32..4d83381d84 100644 --- a/src/bin/pg_dump/pg_backup_archiver.c +++ b/src/bin/pg_dump/pg_backup_archiver.c @@ -66,7 +66,8 @@ typedef struct _parallelReadyList static ArchiveHandle *_allocAH(const char *FileSpec, const ArchiveFormat fmt, const pg_compress_specification compression_spec, bool dosync, ArchiveMode mode, - SetupWorkerPtrType setupWorkerPtr); + SetupWorkerPtrType setupWorkerPtr, + DataDirSyncMethod sync_method); static void _getObjectDescription(PQExpBuffer buf, const TocEntry *te); static void _printTocEntry(ArchiveHandle *AH, TocEntry *te, bool isData); static char *sanitize_line(const char *str, bool want_hyphen); @@ -238,11 +239,12 @@ Archive * CreateArchive(const char *FileSpec, const ArchiveFormat fmt, const pg_compress_specification compression_spec, bool dosync, ArchiveMode mode, - SetupWorkerPtrType setupDumpWorker) + SetupWorkerPtrType setupDumpWorker, + DataDirSyncMethod sync_method) { ArchiveHandle *AH = _allocAH(FileSpec, fmt, compression_spec, - dosync, mode, setupDumpWorker); + dosync, mode, setupDumpWorker, sync_method); return (Archive *) AH; } @@ -257,7 +259,8 @@ OpenArchive(const char *FileSpec, const ArchiveFormat fmt) compression_spec.algorithm = PG_COMPRESSION_NONE; AH = _allocAH(FileSpec, fmt, compression_spec, true, - archModeRead, setupRestoreWorker); + archModeRead, setupRestoreWorker, + DATA_DIR_SYNC_METHOD_FSYNC); return (Archive *) AH; } @@ -2233,7 +2236,7 @@ static ArchiveHandle * _allocAH(const char *FileSpec, const ArchiveFormat fmt, const pg_compress_specification compression_spec, bool dosync, ArchiveMode mode, - SetupWorkerPtrType setupWorkerPtr) + SetupWorkerPtrType setupWorkerPtr, DataDirSyncMethod sync_method) { ArchiveHandle *AH; CompressFileHandle *CFH; @@ -2287,6 +2290,7 @@ _allocAH(const char *FileSpec, const ArchiveFormat fmt, AH->mode = mode; AH->compression_spec = compression_spec; AH->dosync = dosync; + AH->sync_method = sync_method; memset(&(AH->sqlparse), 0, sizeof(AH->sqlparse)); diff --git a/src/bin/pg_dump/pg_backup_archiver.h b/src/bin/pg_dump/pg_backup_archiver.h index 18b38c17ab..b07673933d 100644 --- a/src/bin/pg_dump/pg_backup_archiver.h +++ b/src/bin/pg_dump/pg_backup_archiver.h @@ -312,6 +312,7 @@ struct _archiveHandle pg_compress_specification compression_spec; /* Requested specification for * compression */ bool dosync; /* data requested to be synced on sight */ + DataDirSyncMethod sync_method; ArchiveMode mode; /* File mode - r or w */ void *formatData; /* Header data specific to file format */ diff --git a/src/bin/pg_dump/pg_backup_directory.c b/src/bin/pg_dump/pg_backup_directory.c index 7f2ac7c7fd..6faa3a511f 100644 --- a/src/bin/pg_dump/pg_backup_directory.c +++ b/src/bin/pg_dump/pg_backup_directory.c @@ -613,7 +613,7 @@ _CloseArchive(ArchiveHandle *AH) * individually. Just recurse once through all the files generated. */ if (AH->dosync) - fsync_dir_recurse(ctx->directory); + fsync_dir_recurse(ctx->directory, AH->sync_method); } AH->FH = NULL; } diff --git a/src/bin/pg_dump/pg_dump.c b/src/bin/pg_dump/pg_dump.c index 65f64c282d..39a468b131 100644 --- a/src/bin/pg_dump/pg_dump.c +++ b/src/bin/pg_dump/pg_dump.c @@ -357,6 +357,7 @@ main(int argc, char **argv) char *compression_algorithm_str = "none"; char *error_detail = NULL; bool user_compression_defined = false; + DataDirSyncMethod sync_method = DATA_DIR_SYNC_METHOD_FSYNC; static DumpOptions dopt; @@ -777,7 +778,7 @@ main(int argc, char **argv) /* Open the output file */ fout = CreateArchive(filename, archiveFormat, compression_spec, - dosync, archiveMode, setupDumpWorker); + dosync, archiveMode, setupDumpWorker, sync_method); /* Make dump options accessible right away */ SetArchiveOptions(fout, &dopt, NULL); diff --git a/src/bin/pg_rewind/file_ops.c b/src/bin/pg_rewind/file_ops.c index 25996b4da4..451fb1856e 100644 --- a/src/bin/pg_rewind/file_ops.c +++ b/src/bin/pg_rewind/file_ops.c @@ -296,7 +296,7 @@ sync_target_dir(void) if (!do_sync || dry_run) return; - fsync_pgdata(datadir_target, PG_VERSION_NUM); + fsync_pgdata(datadir_target, PG_VERSION_NUM, sync_method); } diff --git a/src/bin/pg_rewind/pg_rewind.c b/src/bin/pg_rewind/pg_rewind.c index 7f69f02441..bdfacf3263 100644 --- a/src/bin/pg_rewind/pg_rewind.c +++ b/src/bin/pg_rewind/pg_rewind.c @@ -74,6 +74,7 @@ bool showprogress = false; bool dry_run = false; bool do_sync = true; bool restore_wal = false; +DataDirSyncMethod sync_method = DATA_DIR_SYNC_METHOD_FSYNC; /* Target history */ TimeLineHistoryEntry *targetHistory; diff --git a/src/bin/pg_rewind/pg_rewind.h b/src/bin/pg_rewind/pg_rewind.h index ef8bdc1fbb..05729adfef 100644 --- a/src/bin/pg_rewind/pg_rewind.h +++ b/src/bin/pg_rewind/pg_rewind.h @@ -13,6 +13,7 @@ #include "access/timeline.h" #include "common/logging.h" +#include "common/file_utils.h" #include "datapagemap.h" #include "libpq-fe.h" #include "storage/block.h" @@ -24,6 +25,7 @@ extern bool showprogress; extern bool dry_run; extern bool do_sync; extern int WalSegSz; +extern DataDirSyncMethod sync_method; /* Target history */ extern TimeLineHistoryEntry *targetHistory; diff --git a/src/common/file_utils.c b/src/common/file_utils.c index 74833c4acb..05c73c0bb7 100644 --- a/src/common/file_utils.c +++ b/src/common/file_utils.c @@ -51,6 +51,31 @@ static void walkdir(const char *path, int (*action) (const char *fname, bool isdir), bool process_symlinks); +#ifdef HAVE_SYNCFS +static void +do_syncfs(const char *path) +{ + int fd; + + fd = open(path, O_RDONLY, 0); + + if (fd < 0) + { + pg_log_error("could not open file \"%s\": %m", path); + return; + } + + if (syncfs(fd) < 0) + { + pg_log_error("could not synchronize file system for file \"%s\": %m", path); + (void) close(fd); + exit(EXIT_FAILURE); + } + + (void) close(fd); +} +#endif + /* * Issue fsync recursively on PGDATA and all its contents. * @@ -63,7 +88,8 @@ static void walkdir(const char *path, */ void fsync_pgdata(const char *pg_data, - int serverVersion) + int serverVersion, + DataDirSyncMethod sync_method) { bool xlog_is_symlink; char pg_wal[MAXPGPATH]; @@ -89,30 +115,93 @@ fsync_pgdata(const char *pg_data, xlog_is_symlink = true; } - /* - * If possible, hint to the kernel that we're soon going to fsync the data - * directory and its contents. - */ + switch (sync_method) + { + case DATA_DIR_SYNC_METHOD_SYNCFS: + { +#ifndef HAVE_SYNCFS + pg_log_error("this build does not support sync method \"%s\"", + "syncfs"); + exit(EXIT_FAILURE); +#else + DIR *dir; + struct dirent *de; + + /* + * On Linux, we don't have to open every single file one by + * one. We can use syncfs() to sync whole filesystems. We + * only expect filesystem boundaries to exist where we + * tolerate symlinks, namely pg_wal and the tablespaces, so we + * call syncfs() for each of those directories. + */ + + /* Sync the top level pgdata directory. */ + do_syncfs(pg_data); + + /* If any tablespaces are configured, sync each of those. */ + dir = opendir(pg_tblspc); + if (dir == NULL) + pg_log_error("could not open directory \"%s\": %m", + pg_tblspc); + else + { + while (errno = 0, (de = readdir(dir)) != NULL) + { + char subpath[MAXPGPATH * 2]; + + if (strcmp(de->d_name, ".") == 0 || + strcmp(de->d_name, "..") == 0) + continue; + + snprintf(subpath, sizeof(subpath), "%s/%s", + pg_tblspc, de->d_name); + do_syncfs(subpath); + } + + if (errno) + pg_log_error("could not read directory \"%s\": %m", + pg_tblspc); + + (void) closedir(dir); + } + + /* If pg_wal is a symlink, process that too. */ + if (xlog_is_symlink) + do_syncfs(pg_wal); +#endif /* HAVE_SYNCFS */ + } + break; + + case DATA_DIR_SYNC_METHOD_FSYNC: + { + /* + * If possible, hint to the kernel that we're soon going to + * fsync the data directory and its contents. + */ #ifdef PG_FLUSH_DATA_WORKS - walkdir(pg_data, pre_sync_fname, false); - if (xlog_is_symlink) - walkdir(pg_wal, pre_sync_fname, false); - walkdir(pg_tblspc, pre_sync_fname, true); + walkdir(pg_data, pre_sync_fname, false); + if (xlog_is_symlink) + walkdir(pg_wal, pre_sync_fname, false); + walkdir(pg_tblspc, pre_sync_fname, true); #endif - /* - * Now we do the fsync()s in the same order. - * - * The main call ignores symlinks, so in addition to specially processing - * pg_wal if it's a symlink, pg_tblspc has to be visited separately with - * process_symlinks = true. Note that if there are any plain directories - * in pg_tblspc, they'll get fsync'd twice. That's not an expected case - * so we don't worry about optimizing it. - */ - walkdir(pg_data, fsync_fname, false); - if (xlog_is_symlink) - walkdir(pg_wal, fsync_fname, false); - walkdir(pg_tblspc, fsync_fname, true); + /* + * Now we do the fsync()s in the same order. + * + * The main call ignores symlinks, so in addition to specially + * processing pg_wal if it's a symlink, pg_tblspc has to be + * visited separately with process_symlinks = true. Note that + * if there are any plain directories in pg_tblspc, they'll + * get fsync'd twice. That's not an expected case so we don't + * worry about optimizing it. + */ + walkdir(pg_data, fsync_fname, false); + if (xlog_is_symlink) + walkdir(pg_wal, fsync_fname, false); + walkdir(pg_tblspc, fsync_fname, true); + } + break; + } } /* @@ -121,17 +210,40 @@ fsync_pgdata(const char *pg_data, * This is a convenient wrapper on top of walkdir(). */ void -fsync_dir_recurse(const char *dir) +fsync_dir_recurse(const char *dir, DataDirSyncMethod sync_method) { - /* - * If possible, hint to the kernel that we're soon going to fsync the data - * directory and its contents. - */ + switch (sync_method) + { + case DATA_DIR_SYNC_METHOD_SYNCFS: + { +#ifndef HAVE_SYNCFS + pg_log_error("this build does not support sync method \"%s\"", + "syncfs"); + exit(EXIT_FAILURE); +#else + /* + * On Linux, we don't have to open every single file one by + * one. We can use syncfs() to sync the whole filesystem. + */ + do_syncfs(dir); +#endif /* HAVE_SYNCFS */ + } + break; + + case DATA_DIR_SYNC_METHOD_FSYNC: + { + /* + * If possible, hint to the kernel that we're soon going to + * fsync the data directory and its contents. + */ #ifdef PG_FLUSH_DATA_WORKS - walkdir(dir, pre_sync_fname, false); + walkdir(dir, pre_sync_fname, false); #endif - walkdir(dir, fsync_fname, false); + walkdir(dir, fsync_fname, false); + } + break; + } } /* diff --git a/src/include/common/file_utils.h b/src/include/common/file_utils.h index 7da21f15e6..cae6159bf6 100644 --- a/src/include/common/file_utils.h +++ b/src/include/common/file_utils.h @@ -34,8 +34,9 @@ struct iovec; /* avoid including port/pg_iovec.h here */ #ifdef FRONTEND extern int fsync_fname(const char *fname, bool isdir); -extern void fsync_pgdata(const char *pg_data, int serverVersion); -extern void fsync_dir_recurse(const char *dir); +extern void fsync_pgdata(const char *pg_data, int serverVersion, + DataDirSyncMethod sync_method); +extern void fsync_dir_recurse(const char *dir, DataDirSyncMethod sync_method); extern int durable_rename(const char *oldfile, const char *newfile); extern int fsync_parent_path(const char *fname); #endif -- 2.25.1 --2fHTh5uZTiUOsy+g Content-Type: text/x-diff; charset=us-ascii Content-Disposition: attachment; filename="v9-0004-allow-syncfs-in-frontend-utilities.patch" ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-08 14:38 Bharath Rupireddy <[email protected]> 0 siblings, 1 reply; 18+ messages in thread From: Bharath Rupireddy @ 2024-03-08 14:38 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Nathan Bossart <[email protected]>; Bertrand Drouvot <[email protected]>; PostgreSQL Hackers <[email protected]> On Wed, Mar 6, 2024 at 4:28 PM Amit Kapila <[email protected]> wrote: > > IIUC, the current conflict_reason is primarily used to determine > logical slots on standby that got invalidated due to recovery time > conflict. On the primary, it will also show logical slots that got > invalidated due to the corresponding WAL got removed. Is that > understanding correct? That's right. > If so, we are already sort of overloading this > column. However, now adding more invalidation reasons that won't > happen during recovery conflict handling will change entirely the > purpose (as per the name we use) of this variable. I think > invalidation_reason could depict this column correctly but OTOH I > guess it would lose its original meaning/purpose. Hm. I get the concern. Are you okay with having inavlidation_reason separately for both logical and physical slots? In such a case, logical slots that got invalidated on the standby will have duplicate info in conflict_reason and invalidation_reason, is this fine? Another idea is to make 'conflict_reason text' as a 'conflicting boolean' again (revert 007693f2a3), and have 'invalidation_reason text' for both logical and physical slots. So, whenever 'conflicting' is true, one can look at invalidation_reason for the reason for conflict. How does this sound? -- Bharath Rupireddy PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-11 05:55 Amit Kapila <[email protected]> parent: Bharath Rupireddy <[email protected]> 0 siblings, 1 reply; 18+ messages in thread From: Amit Kapila @ 2024-03-11 05:55 UTC (permalink / raw) To: Bharath Rupireddy <[email protected]>; +Cc: Nathan Bossart <[email protected]>; Bertrand Drouvot <[email protected]>; PostgreSQL Hackers <[email protected]> On Fri, Mar 8, 2024 at 8:08 PM Bharath Rupireddy <[email protected]> wrote: > > On Wed, Mar 6, 2024 at 4:28 PM Amit Kapila <[email protected]> wrote: > > > > IIUC, the current conflict_reason is primarily used to determine > > logical slots on standby that got invalidated due to recovery time > > conflict. On the primary, it will also show logical slots that got > > invalidated due to the corresponding WAL got removed. Is that > > understanding correct? > > That's right. > > > If so, we are already sort of overloading this > > column. However, now adding more invalidation reasons that won't > > happen during recovery conflict handling will change entirely the > > purpose (as per the name we use) of this variable. I think > > invalidation_reason could depict this column correctly but OTOH I > > guess it would lose its original meaning/purpose. > > Hm. I get the concern. Are you okay with having inavlidation_reason > separately for both logical and physical slots? In such a case, > logical slots that got invalidated on the standby will have duplicate > info in conflict_reason and invalidation_reason, is this fine? > If we have duplicate information in two columns that could be confusing for users. BTW, isn't the recovery conflict occur only because of rows_removed and wal_level_insufficient reasons? The wal_removed or the new reasons you are proposing can't happen because of recovery conflict. Am, I missing something here? > Another idea is to make 'conflict_reason text' as a 'conflicting > boolean' again (revert 007693f2a3), and have 'invalidation_reason > text' for both logical and physical slots. So, whenever 'conflicting' > is true, one can look at invalidation_reason for the reason for > conflict. How does this sound? > So, does this mean that conflicting will only be true for some of the reasons (say wal_level_insufficient, rows_removed, wal_removed) and logical slots but not for others? I think that will also not eliminate the duplicate information as user could have deduced that from single column -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-12 15:25 Bharath Rupireddy <[email protected]> parent: Amit Kapila <[email protected]> 0 siblings, 1 reply; 18+ messages in thread From: Bharath Rupireddy @ 2024-03-12 15:25 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Nathan Bossart <[email protected]>; Bertrand Drouvot <[email protected]>; PostgreSQL Hackers <[email protected]> On Mon, Mar 11, 2024 at 11:26 AM Amit Kapila <[email protected]> wrote: > > > Hm. I get the concern. Are you okay with having inavlidation_reason > > separately for both logical and physical slots? In such a case, > > logical slots that got invalidated on the standby will have duplicate > > info in conflict_reason and invalidation_reason, is this fine? > > > > If we have duplicate information in two columns that could be > confusing for users. BTW, isn't the recovery conflict occur only > because of rows_removed and wal_level_insufficient reasons? The > wal_removed or the new reasons you are proposing can't happen because > of recovery conflict. Am, I missing something here? My understanding aligns with yours that the rows_removed and wal_level_insufficient invalidations can occur only upon recovery conflict. FWIW, a test named 'synchronized slot has been invalidated' in 040_standby_failover_slots_sync.pl inappropriately uses conflict_reason = 'wal_removed' logical slot on standby. As per the above understanding, it's inappropriate to use conflict_reason here because wal_removed invalidation doesn't conflict with recovery. > > Another idea is to make 'conflict_reason text' as a 'conflicting > > boolean' again (revert 007693f2a3), and have 'invalidation_reason > > text' for both logical and physical slots. So, whenever 'conflicting' > > is true, one can look at invalidation_reason for the reason for > > conflict. How does this sound? > > > > So, does this mean that conflicting will only be true for some of the > reasons (say wal_level_insufficient, rows_removed, wal_removed) and > logical slots but not for others? I think that will also not eliminate > the duplicate information as user could have deduced that from single > column. So, how about we turn conflict_reason to only report the reasons that actually cause conflict with recovery for logical slots, something like below, and then have invalidation_cause as a generic column for all sorts of invalidation reasons for both logical and physical slots? ReplicationSlotInvalidationCause cause = slot_contents.data.invalidated; if (slot_contents.data.database == InvalidOid || cause == RS_INVAL_NONE || cause != RS_INVAL_HORIZON || cause != RS_INVAL_WAL_LEVEL) { nulls[i++] = true; } else { Assert(cause == RS_INVAL_HORIZON || cause == RS_INVAL_WAL_LEVEL); values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]); } -- Bharath Rupireddy PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-13 03:51 Amit Kapila <[email protected]> parent: Bharath Rupireddy <[email protected]> 0 siblings, 1 reply; 18+ messages in thread From: Amit Kapila @ 2024-03-13 03:51 UTC (permalink / raw) To: Bharath Rupireddy <[email protected]>; +Cc: Nathan Bossart <[email protected]>; Bertrand Drouvot <[email protected]>; PostgreSQL Hackers <[email protected]> On Tue, Mar 12, 2024 at 8:55 PM Bharath Rupireddy <[email protected]> wrote: > > On Mon, Mar 11, 2024 at 11:26 AM Amit Kapila <[email protected]> wrote: > > > > > Hm. I get the concern. Are you okay with having inavlidation_reason > > > separately for both logical and physical slots? In such a case, > > > logical slots that got invalidated on the standby will have duplicate > > > info in conflict_reason and invalidation_reason, is this fine? > > > > > > > If we have duplicate information in two columns that could be > > confusing for users. BTW, isn't the recovery conflict occur only > > because of rows_removed and wal_level_insufficient reasons? The > > wal_removed or the new reasons you are proposing can't happen because > > of recovery conflict. Am, I missing something here? > > My understanding aligns with yours that the rows_removed and > wal_level_insufficient invalidations can occur only upon recovery > conflict. > > FWIW, a test named 'synchronized slot has been invalidated' in > 040_standby_failover_slots_sync.pl inappropriately uses > conflict_reason = 'wal_removed' logical slot on standby. As per the > above understanding, it's inappropriate to use conflict_reason here > because wal_removed invalidation doesn't conflict with recovery. > > > > Another idea is to make 'conflict_reason text' as a 'conflicting > > > boolean' again (revert 007693f2a3), and have 'invalidation_reason > > > text' for both logical and physical slots. So, whenever 'conflicting' > > > is true, one can look at invalidation_reason for the reason for > > > conflict. How does this sound? > > > > > > > So, does this mean that conflicting will only be true for some of the > > reasons (say wal_level_insufficient, rows_removed, wal_removed) and > > logical slots but not for others? I think that will also not eliminate > > the duplicate information as user could have deduced that from single > > column. > > So, how about we turn conflict_reason to only report the reasons that > actually cause conflict with recovery for logical slots, something > like below, and then have invalidation_cause as a generic column for > all sorts of invalidation reasons for both logical and physical slots? > If our above understanding is correct then coflict_reason will be a subset of invalidation_reason. If so, whatever way we arrange this information, there will be some sort of duplicity unless we just have one column 'invalidation_reason' and update the docs to interpret it correctly for conflicts. -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-13 15:54 Bharath Rupireddy <[email protected]> parent: Amit Kapila <[email protected]> 0 siblings, 1 reply; 18+ messages in thread From: Bharath Rupireddy @ 2024-03-13 15:54 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Nathan Bossart <[email protected]>; Bertrand Drouvot <[email protected]>; PostgreSQL Hackers <[email protected]> On Wed, Mar 13, 2024 at 9:21 AM Amit Kapila <[email protected]> wrote: > > > So, how about we turn conflict_reason to only report the reasons that > > actually cause conflict with recovery for logical slots, something > > like below, and then have invalidation_cause as a generic column for > > all sorts of invalidation reasons for both logical and physical slots? > > If our above understanding is correct then coflict_reason will be a > subset of invalidation_reason. If so, whatever way we arrange this > information, there will be some sort of duplicity unless we just have > one column 'invalidation_reason' and update the docs to interpret it > correctly for conflicts. Yes, there will be some sort of duplicity if we emit conflict_reason as a text field. However, I still think the better way is to turn conflict_reason text to conflict boolean and set it to true only on rows_removed and wal_level_insufficient invalidations. When conflict boolean is true, one (including all the tests that we've added recently) can look for invalidation_reason text field for the reason. This sounds reasonable to me as opposed to we just mentioning in the docs that "if invalidation_reason is rows_removed or wal_level_insufficient it's the reason for conflict with recovery". Thoughts? -- Bharath Rupireddy PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-14 06:54 Amit Kapila <[email protected]> parent: Bharath Rupireddy <[email protected]> 0 siblings, 3 replies; 18+ messages in thread From: Amit Kapila @ 2024-03-14 06:54 UTC (permalink / raw) To: Bharath Rupireddy <[email protected]>; +Cc: Nathan Bossart <[email protected]>; Bertrand Drouvot <[email protected]>; PostgreSQL Hackers <[email protected]> On Wed, Mar 13, 2024 at 9:24 PM Bharath Rupireddy <[email protected]> wrote: > > On Wed, Mar 13, 2024 at 9:21 AM Amit Kapila <[email protected]> wrote: > > > > > So, how about we turn conflict_reason to only report the reasons that > > > actually cause conflict with recovery for logical slots, something > > > like below, and then have invalidation_cause as a generic column for > > > all sorts of invalidation reasons for both logical and physical slots? > > > > If our above understanding is correct then coflict_reason will be a > > subset of invalidation_reason. If so, whatever way we arrange this > > information, there will be some sort of duplicity unless we just have > > one column 'invalidation_reason' and update the docs to interpret it > > correctly for conflicts. > > Yes, there will be some sort of duplicity if we emit conflict_reason > as a text field. However, I still think the better way is to turn > conflict_reason text to conflict boolean and set it to true only on > rows_removed and wal_level_insufficient invalidations. When conflict > boolean is true, one (including all the tests that we've added > recently) can look for invalidation_reason text field for the reason. > This sounds reasonable to me as opposed to we just mentioning in the > docs that "if invalidation_reason is rows_removed or > wal_level_insufficient it's the reason for conflict with recovery". > Fair point. I think we can go either way. Bertrand, Nathan, and others, do you have an opinion on this matter? -- With Regards, Amit Kapila. ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-14 14:27 Bharath Rupireddy <[email protected]> parent: Amit Kapila <[email protected]> 2 siblings, 2 replies; 18+ messages in thread From: Bharath Rupireddy @ 2024-03-14 14:27 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Nathan Bossart <[email protected]>; Bertrand Drouvot <[email protected]>; PostgreSQL Hackers <[email protected]> On Thu, Mar 14, 2024 at 12:24 PM Amit Kapila <[email protected]> wrote: > > On Wed, Mar 13, 2024 at 9:24 PM Bharath Rupireddy > > > > Yes, there will be some sort of duplicity if we emit conflict_reason > > as a text field. However, I still think the better way is to turn > > conflict_reason text to conflict boolean and set it to true only on > > rows_removed and wal_level_insufficient invalidations. When conflict > > boolean is true, one (including all the tests that we've added > > recently) can look for invalidation_reason text field for the reason. > > This sounds reasonable to me as opposed to we just mentioning in the > > docs that "if invalidation_reason is rows_removed or > > wal_level_insufficient it's the reason for conflict with recovery". > > > Fair point. I think we can go either way. Bertrand, Nathan, and > others, do you have an opinion on this matter? While we wait to hear from others on this, I'm attaching the v9 patch set implementing the above idea (check 0001 patch). Please have a look. I'll come back to the other review comments soon. -- Bharath Rupireddy PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com Attachments: [application/x-patch] v9-0001-Track-invalidation_reason-in-pg_replication_slots.patch (19.8K, ../../CALj2ACVRS2LZbycpM8rZT8QynR05AtvttOHfFOhczjyqU7zZSA@mail.gmail.com/2-v9-0001-Track-invalidation_reason-in-pg_replication_slots.patch) download | inline diff: From 18855c08cd8bcbaf41aba10048f0ea23a246e546 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Thu, 14 Mar 2024 12:48:52 +0000 Subject: [PATCH v9 1/4] Track invalidation_reason in pg_replication_slots Up until now, reason for replication slot invalidation is not tracked in pg_replication_slots. A recent commit 007693f2a added conflict_reason to show the reasons for slot invalidation, but only for logical slots. This commit adds a new column to show invalidation reasons for both physical and logical slots. And, this commit also turns conflict_reason text column to conflicting boolean column (effectively reverting commit 007693f2a). One now can look at the new invalidation_reason column for logical slots conflict with recovery. --- doc/src/sgml/ref/pgupgrade.sgml | 4 +- doc/src/sgml/system-views.sgml | 63 +++++++++++-------- src/backend/catalog/system_views.sql | 5 +- src/backend/replication/logical/slotsync.c | 2 +- src/backend/replication/slot.c | 8 +-- src/backend/replication/slotfuncs.c | 25 +++++--- src/bin/pg_upgrade/info.c | 4 +- src/include/catalog/pg_proc.dat | 6 +- src/include/replication/slot.h | 2 +- .../t/035_standby_logical_decoding.pl | 35 ++++++----- .../t/040_standby_failover_slots_sync.pl | 4 +- src/test/regress/expected/rules.out | 7 ++- 12 files changed, 95 insertions(+), 70 deletions(-) diff --git a/doc/src/sgml/ref/pgupgrade.sgml b/doc/src/sgml/ref/pgupgrade.sgml index 58c6c2df8b..8de52bf752 100644 --- a/doc/src/sgml/ref/pgupgrade.sgml +++ b/doc/src/sgml/ref/pgupgrade.sgml @@ -453,8 +453,8 @@ make prefix=/usr/local/pgsql.new install <para> All slots on the old cluster must be usable, i.e., there are no slots whose - <link linkend="view-pg-replication-slots">pg_replication_slots</link>.<structfield>conflict_reason</structfield> - is not <literal>NULL</literal>. + <link linkend="view-pg-replication-slots">pg_replication_slots</link>.<structfield>conflicting</structfield> + is not <literal>true</literal>. </para> </listitem> <listitem> diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml index be90edd0e2..f3fb5ba1b0 100644 --- a/doc/src/sgml/system-views.sgml +++ b/doc/src/sgml/system-views.sgml @@ -2525,34 +2525,13 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx <row> <entry role="catalog_table_entry"><para role="column_definition"> - <structfield>conflict_reason</structfield> <type>text</type> + <structfield>conflicting</structfield> <type>bool</type> </para> <para> - The reason for the logical slot's conflict with recovery. It is always - NULL for physical slots, as well as for logical slots which are not - invalidated. The non-NULL values indicate that the slot is marked - as invalidated. Possible values are: - <itemizedlist spacing="compact"> - <listitem> - <para> - <literal>wal_removed</literal> means that the required WAL has been - removed. - </para> - </listitem> - <listitem> - <para> - <literal>rows_removed</literal> means that the required rows have - been removed. - </para> - </listitem> - <listitem> - <para> - <literal>wal_level_insufficient</literal> means that the - primary doesn't have a <xref linkend="guc-wal-level"/> sufficient to - perform logical decoding. - </para> - </listitem> - </itemizedlist> + True if this logical slot conflicted with recovery (and so is now + invalidated). When this column is true, check + <structfield>invalidation_reason</structfield> column for the conflict + reason. </para></entry> </row> @@ -2581,6 +2560,38 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx </para></entry> </row> + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>invalidation_reason</structfield> <type>text</type> + </para> + <para> + The reason for the slot's invalidation. <literal>NULL</literal> if the + slot is currently actively being used. The non-NULL values indicate that + the slot is marked as invalidated. Possible values are: + <itemizedlist spacing="compact"> + <listitem> + <para> + <literal>wal_removed</literal> means that the required WAL has been + removed. + </para> + </listitem> + <listitem> + <para> + <literal>rows_removed</literal> means that the required rows have + been removed. + </para> + </listitem> + <listitem> + <para> + <literal>wal_level_insufficient</literal> means that the + primary doesn't have a <xref linkend="guc-wal-level"/> sufficient to + perform logical decoding. + </para> + </listitem> + </itemizedlist> + </para></entry> + </row> + </tbody> </tgroup> </table> diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 04227a72d1..cd22dad959 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1023,9 +1023,10 @@ CREATE VIEW pg_replication_slots AS L.wal_status, L.safe_wal_size, L.two_phase, - L.conflict_reason, + L.conflicting, L.failover, - L.synced + L.synced, + L.invalidation_reason FROM pg_get_replication_slots() AS L LEFT JOIN pg_database D ON (L.datoid = D.oid); diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index 5074c8409f..260632cfdd 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -668,7 +668,7 @@ synchronize_slots(WalReceiverConn *wrconn) bool started_tx = false; const char *query = "SELECT slot_name, plugin, confirmed_flush_lsn," " restart_lsn, catalog_xmin, two_phase, failover," - " database, conflict_reason" + " database, invalidation_reason" " FROM pg_catalog.pg_replication_slots" " WHERE failover and NOT temporary"; diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 91ca397857..4f1a17f6ce 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -2356,21 +2356,21 @@ RestoreSlotFromDisk(const char *name) } /* - * Maps a conflict reason for a replication slot to + * Maps a invalidation reason for a replication slot to * ReplicationSlotInvalidationCause. */ ReplicationSlotInvalidationCause -GetSlotInvalidationCause(const char *conflict_reason) +GetSlotInvalidationCause(const char *invalidation_reason) { ReplicationSlotInvalidationCause cause; ReplicationSlotInvalidationCause result = RS_INVAL_NONE; bool found PG_USED_FOR_ASSERTS_ONLY = false; - Assert(conflict_reason); + Assert(invalidation_reason); for (cause = RS_INVAL_NONE; cause <= RS_INVAL_MAX_CAUSES; cause++) { - if (strcmp(SlotInvalidationCauses[cause], conflict_reason) == 0) + if (strcmp(SlotInvalidationCauses[cause], invalidation_reason) == 0) { found = true; result = cause; diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c index ad79e1fccd..b5a638edea 100644 --- a/src/backend/replication/slotfuncs.c +++ b/src/backend/replication/slotfuncs.c @@ -239,7 +239,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS) Datum pg_get_replication_slots(PG_FUNCTION_ARGS) { -#define PG_GET_REPLICATION_SLOTS_COLS 17 +#define PG_GET_REPLICATION_SLOTS_COLS 18 ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; XLogRecPtr currlsn; int slotno; @@ -263,6 +263,7 @@ pg_get_replication_slots(PG_FUNCTION_ARGS) bool nulls[PG_GET_REPLICATION_SLOTS_COLS]; WALAvailability walstate; int i; + ReplicationSlotInvalidationCause cause; if (!slot->in_use) continue; @@ -409,22 +410,32 @@ pg_get_replication_slots(PG_FUNCTION_ARGS) values[i++] = BoolGetDatum(slot_contents.data.two_phase); - if (slot_contents.data.database == InvalidOid) + cause = slot_contents.data.invalidated; + + if (SlotIsPhysical(&slot_contents)) nulls[i++] = true; else { - ReplicationSlotInvalidationCause cause = slot_contents.data.invalidated; - - if (cause == RS_INVAL_NONE) - nulls[i++] = true; + /* + * rows_removed and wal_level_insufficient are only two reasons + * for the logical slot's conflict with recovery. + */ + if (cause == RS_INVAL_HORIZON || + cause == RS_INVAL_WAL_LEVEL) + values[i++] = BoolGetDatum(true); else - values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]); + values[i++] = BoolGetDatum(false); } values[i++] = BoolGetDatum(slot_contents.data.failover); values[i++] = BoolGetDatum(slot_contents.data.synced); + if (cause == RS_INVAL_NONE) + nulls[i++] = true; + else + values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]); + Assert(i == PG_GET_REPLICATION_SLOTS_COLS); tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, diff --git a/src/bin/pg_upgrade/info.c b/src/bin/pg_upgrade/info.c index b5b8d11602..34a157f792 100644 --- a/src/bin/pg_upgrade/info.c +++ b/src/bin/pg_upgrade/info.c @@ -676,13 +676,13 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check) * removed. */ res = executeQueryOrDie(conn, "SELECT slot_name, plugin, two_phase, failover, " - "%s as caught_up, conflict_reason IS NOT NULL as invalid " + "%s as caught_up, invalidation_reason IS NOT NULL as invalid " "FROM pg_catalog.pg_replication_slots " "WHERE slot_type = 'logical' AND " "database = current_database() AND " "temporary IS FALSE;", live_check ? "FALSE" : - "(CASE WHEN conflict_reason IS NOT NULL THEN FALSE " + "(CASE WHEN conflicting THEN FALSE " "ELSE (SELECT pg_catalog.binary_upgrade_logical_slot_has_caught_up(slot_name)) " "END)"); diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 4af5c2e847..e5dc1cbdb3 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -11120,9 +11120,9 @@ proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f', proretset => 't', provolatile => 's', prorettype => 'record', proargtypes => '', - proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,text,bool,bool}', - proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}', - proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflict_reason,failover,synced}', + proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,bool,bool,text}', + proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}', + proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,failover,synced,invalidation_reason}', prosrc => 'pg_get_replication_slots' }, { oid => '3786', descr => 'set up a logical replication slot', proname => 'pg_create_logical_replication_slot', provolatile => 'v', diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 425effad21..7f25a083ee 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -273,7 +273,7 @@ extern void CheckPointReplicationSlots(bool is_shutdown); extern void CheckSlotRequirements(void); extern void CheckSlotPermissions(void); extern ReplicationSlotInvalidationCause - GetSlotInvalidationCause(const char *conflict_reason); + GetSlotInvalidationCause(const char *invalidation_reason); extern bool SlotExistsInStandbySlotNames(const char *slot_name); extern bool StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel); diff --git a/src/test/recovery/t/035_standby_logical_decoding.pl b/src/test/recovery/t/035_standby_logical_decoding.pl index 88b03048c4..addff6a1a5 100644 --- a/src/test/recovery/t/035_standby_logical_decoding.pl +++ b/src/test/recovery/t/035_standby_logical_decoding.pl @@ -168,7 +168,7 @@ sub change_hot_standby_feedback_and_wait_for_xmins } } -# Check conflict_reason in pg_replication_slots. +# Check reason for conflict in pg_replication_slots. sub check_slots_conflict_reason { my ($slot_prefix, $reason) = @_; @@ -178,15 +178,15 @@ sub check_slots_conflict_reason $res = $node_standby->safe_psql( 'postgres', qq( - select conflict_reason from pg_replication_slots where slot_name = '$active_slot';)); + select invalidation_reason from pg_replication_slots where slot_name = '$active_slot' and conflicting;)); - is($res, "$reason", "$active_slot conflict_reason is $reason"); + is($res, "$reason", "$active_slot reason for conflict is $reason"); $res = $node_standby->safe_psql( 'postgres', qq( - select conflict_reason from pg_replication_slots where slot_name = '$inactive_slot';)); + select invalidation_reason from pg_replication_slots where slot_name = '$inactive_slot' and conflicting;)); - is($res, "$reason", "$inactive_slot conflict_reason is $reason"); + is($res, "$reason", "$inactive_slot reason for conflict is $reason"); } # Drop the slots, re-create them, change hot_standby_feedback, @@ -293,13 +293,13 @@ $node_primary->safe_psql('testdb', qq[SELECT * FROM pg_create_physical_replication_slot('$primary_slotname');] ); -# Check conflict_reason is NULL for physical slot +# Check conflicting is NULL for physical slot $res = $node_primary->safe_psql( 'postgres', qq[ - SELECT conflict_reason is null FROM pg_replication_slots where slot_name = '$primary_slotname';] + SELECT conflicting is null FROM pg_replication_slots where slot_name = '$primary_slotname';] ); -is($res, 't', "Physical slot reports conflict_reason as NULL"); +is($res, 't', "Physical slot reports conflicting as NULL"); my $backup_name = 'b1'; $node_primary->backup($backup_name); @@ -524,7 +524,7 @@ $node_primary->wait_for_replay_catchup($node_standby); # Check invalidation in the logfile and in pg_stat_database_conflicts check_for_invalidation('vacuum_full_', 1, 'with vacuum FULL on pg_class'); -# Verify conflict_reason is 'rows_removed' in pg_replication_slots +# Verify reason for conflict is 'rows_removed' in pg_replication_slots check_slots_conflict_reason('vacuum_full_', 'rows_removed'); # Ensure that replication slot stats are not removed after invalidation. @@ -551,7 +551,7 @@ change_hot_standby_feedback_and_wait_for_xmins(1, 1); ################################################## $node_standby->restart; -# Verify conflict_reason is retained across a restart. +# Verify reason for conflict is retained across a restart. check_slots_conflict_reason('vacuum_full_', 'rows_removed'); ################################################## @@ -560,7 +560,8 @@ check_slots_conflict_reason('vacuum_full_', 'rows_removed'); # Get the restart_lsn from an invalidated slot my $restart_lsn = $node_standby->safe_psql('postgres', - "SELECT restart_lsn from pg_replication_slots WHERE slot_name = 'vacuum_full_activeslot' and conflict_reason is not null;" + "SELECT restart_lsn FROM pg_replication_slots + WHERE slot_name = 'vacuum_full_activeslot' AND conflicting;" ); chomp($restart_lsn); @@ -611,7 +612,7 @@ $node_primary->wait_for_replay_catchup($node_standby); # Check invalidation in the logfile and in pg_stat_database_conflicts check_for_invalidation('row_removal_', $logstart, 'with vacuum on pg_class'); -# Verify conflict_reason is 'rows_removed' in pg_replication_slots +# Verify reason for conflict is 'rows_removed' in pg_replication_slots check_slots_conflict_reason('row_removal_', 'rows_removed'); $handle = @@ -647,7 +648,7 @@ $node_primary->wait_for_replay_catchup($node_standby); check_for_invalidation('shared_row_removal_', $logstart, 'with vacuum on pg_authid'); -# Verify conflict_reason is 'rows_removed' in pg_replication_slots +# Verify reason for conflict is 'rows_removed' in pg_replication_slots check_slots_conflict_reason('shared_row_removal_', 'rows_removed'); $handle = make_slot_active($node_standby, 'shared_row_removal_', 0, \$stdout, @@ -700,8 +701,8 @@ ok( $node_standby->poll_query_until( is( $node_standby->safe_psql( 'postgres', q[select bool_or(conflicting) from - (select conflict_reason is not NULL as conflicting - from pg_replication_slots WHERE slot_type = 'logical')]), + (select conflicting from pg_replication_slots + where slot_type = 'logical')]), 'f', 'Logical slots are reported as non conflicting'); @@ -739,7 +740,7 @@ $node_primary->wait_for_replay_catchup($node_standby); # Check invalidation in the logfile and in pg_stat_database_conflicts check_for_invalidation('pruning_', $logstart, 'with on-access pruning'); -# Verify conflict_reason is 'rows_removed' in pg_replication_slots +# Verify reason for conflict is 'rows_removed' in pg_replication_slots check_slots_conflict_reason('pruning_', 'rows_removed'); $handle = make_slot_active($node_standby, 'pruning_', 0, \$stdout, \$stderr); @@ -783,7 +784,7 @@ $node_primary->wait_for_replay_catchup($node_standby); # Check invalidation in the logfile and in pg_stat_database_conflicts check_for_invalidation('wal_level_', $logstart, 'due to wal_level'); -# Verify conflict_reason is 'wal_level_insufficient' in pg_replication_slots +# Verify reason for conflict is 'wal_level_insufficient' in pg_replication_slots check_slots_conflict_reason('wal_level_', 'wal_level_insufficient'); $handle = diff --git a/src/test/recovery/t/040_standby_failover_slots_sync.pl b/src/test/recovery/t/040_standby_failover_slots_sync.pl index 0ea1f3d323..f47bfd78eb 100644 --- a/src/test/recovery/t/040_standby_failover_slots_sync.pl +++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl @@ -228,7 +228,7 @@ $standby1->safe_psql('postgres', "CHECKPOINT"); # Check if the synced slot is invalidated is( $standby1->safe_psql( 'postgres', - q{SELECT conflict_reason = 'wal_removed' FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';} + q{SELECT invalidation_reason = 'wal_removed' FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';} ), "t", 'synchronized slot has been invalidated'); @@ -274,7 +274,7 @@ $standby1->wait_for_log(qr/dropped replication slot "lsub1_slot" of dbid [0-9]+/ # flagged as 'synced' is( $standby1->safe_psql( 'postgres', - q{SELECT conflict_reason IS NULL AND synced AND NOT temporary FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';} + q{SELECT invalidation_reason IS NULL AND synced AND NOT temporary FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';} ), "t", 'logical slot is re-synced'); diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 0cd2c64fca..055bec068d 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1473,10 +1473,11 @@ pg_replication_slots| SELECT l.slot_name, l.wal_status, l.safe_wal_size, l.two_phase, - l.conflict_reason, + l.conflicting, l.failover, - l.synced - FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflict_reason, failover, synced) + l.synced, + l.invalidation_reason + FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, failover, synced, invalidation_reason) LEFT JOIN pg_database d ON ((l.datoid = d.oid))); pg_roles| SELECT pg_authid.rolname, pg_authid.rolsuper, -- 2.34.1 [application/x-patch] v9-0002-Add-XID-age-based-replication-slot-invalidation.patch (12.8K, ../../CALj2ACVRS2LZbycpM8rZT8QynR05AtvttOHfFOhczjyqU7zZSA@mail.gmail.com/3-v9-0002-Add-XID-age-based-replication-slot-invalidation.patch) download | inline diff: From 4d4248000adfd9096c652bdf0a654ac2203d57a0 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Thu, 14 Mar 2024 12:51:48 +0000 Subject: [PATCH v9 2/4] Add XID age based replication slot invalidation Currently postgres has the ability to invalidate inactive replication slots based on the amount of WAL (set via max_slot_wal_keep_size GUC) that will be needed for the slots in case they become active. However, choosing a default value for max_slot_wal_keep_size is tricky. Because the amount of WAL a customer generates, and their allocated storage will vary greatly in production, making it difficult to pin down a one-size-fits-all value. It is often easy for developers to set an XID age (age of slot's xmin or catalog_xmin) of say 1 or 1.5 billion, after which the slots get invalidated. To achieve the above, postgres uses replication slot xmin (the oldest transaction that this slot needs the database to retain) or catalog_xmin (the oldest transaction affecting the system catalogs that this slot needs the database to retain), and a new GUC max_slot_xid_age. The checkpointer then looks at all replication slots invalidating the slots based on the age set. --- doc/src/sgml/config.sgml | 21 ++++ src/backend/access/transam/xlog.c | 10 ++ src/backend/replication/slot.c | 44 ++++++- src/backend/utils/misc/guc_tables.c | 10 ++ src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/replication/slot.h | 3 + src/test/recovery/meson.build | 1 + src/test/recovery/t/050_invalidate_slots.pl | 108 ++++++++++++++++++ 8 files changed, 197 insertions(+), 1 deletion(-) create mode 100644 src/test/recovery/t/050_invalidate_slots.pl diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 65a6e6c408..6dd54ffcb7 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -4544,6 +4544,27 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"' # Windows </listitem> </varlistentry> + <varlistentry id="guc-max-slot-xid-age" xreflabel="max_slot_xid_age"> + <term><varname>max_slot_xid_age</varname> (<type>integer</type>) + <indexterm> + <primary><varname>max_slot_xid_age</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + Invalidate replication slots whose <literal>xmin</literal> (the oldest + transaction that this slot needs the database to retain) or + <literal>catalog_xmin</literal> (the oldest transaction affecting the + system catalogs that this slot needs the database to retain) has reached + the age specified by this setting. A value of zero (which is default) + disables this feature. Users can set this value anywhere from zero to + two billion. This parameter can only be set in the + <filename>postgresql.conf</filename> file or on the server command + line. + </para> + </listitem> + </varlistentry> + <varlistentry id="guc-track-commit-timestamp" xreflabel="track_commit_timestamp"> <term><varname>track_commit_timestamp</varname> (<type>boolean</type>) <indexterm> diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 20a5f86209..36ae2ac6a4 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -7147,6 +7147,11 @@ CreateCheckPoint(int flags) if (PriorRedoPtr != InvalidXLogRecPtr) UpdateCheckPointDistanceEstimate(RedoRecPtr - PriorRedoPtr); + /* Invalidate replication slots based on xmin or catalog_xmin age */ + if (max_slot_xid_age > 0) + InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0, + InvalidOid, InvalidTransactionId); + /* * Delete old log files, those no longer needed for last checkpoint to * prevent the disk holding the xlog from growing full. @@ -7597,6 +7602,11 @@ CreateRestartPoint(int flags) */ XLByteToSeg(RedoRecPtr, _logSegNo, wal_segment_size); + /* Invalidate replication slots based on xmin or catalog_xmin age */ + if (max_slot_xid_age > 0) + InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0, + InvalidOid, InvalidTransactionId); + /* * Retreat _logSegNo using the current end of xlog replayed or received, * whichever is later. diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 4f1a17f6ce..2a1885da24 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -107,10 +107,11 @@ const char *const SlotInvalidationCauses[] = { [RS_INVAL_WAL_REMOVED] = "wal_removed", [RS_INVAL_HORIZON] = "rows_removed", [RS_INVAL_WAL_LEVEL] = "wal_level_insufficient", + [RS_INVAL_XID_AGE] = "xid_aged", }; /* Maximum number of invalidation causes */ -#define RS_INVAL_MAX_CAUSES RS_INVAL_WAL_LEVEL +#define RS_INVAL_MAX_CAUSES RS_INVAL_XID_AGE StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1), "array length mismatch"); @@ -140,6 +141,7 @@ ReplicationSlot *MyReplicationSlot = NULL; /* GUC variables */ int max_replication_slots = 10; /* the maximum number of replication * slots */ +int max_slot_xid_age = 0; /* * This GUC lists streaming replication standby server slot names that @@ -1483,6 +1485,9 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, case RS_INVAL_WAL_LEVEL: appendStringInfoString(&err_detail, _("Logical decoding on standby requires wal_level >= logical on the primary server.")); break; + case RS_INVAL_XID_AGE: + appendStringInfoString(&err_detail, _("The replication slot's xmin or catalog_xmin reached the age specified by max_slot_xid_age.")); + break; case RS_INVAL_NONE: pg_unreachable(); } @@ -1599,6 +1604,42 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, if (SlotIsLogical(s)) conflict = cause; break; + case RS_INVAL_XID_AGE: + { + TransactionId xid_cur = ReadNextTransactionId(); + TransactionId xid_limit; + TransactionId xid_slot; + + if (TransactionIdIsNormal(s->data.xmin)) + { + xid_slot = s->data.xmin; + + xid_limit = xid_slot + max_slot_xid_age; + if (xid_limit < FirstNormalTransactionId) + xid_limit += FirstNormalTransactionId; + + if (TransactionIdFollowsOrEquals(xid_cur, xid_limit)) + { + conflict = cause; + break; + } + } + if (TransactionIdIsNormal(s->data.catalog_xmin)) + { + xid_slot = s->data.catalog_xmin; + + xid_limit = xid_slot + max_slot_xid_age; + if (xid_limit < FirstNormalTransactionId) + xid_limit += FirstNormalTransactionId; + + if (TransactionIdFollowsOrEquals(xid_cur, xid_limit)) + { + conflict = cause; + break; + } + } + } + break; case RS_INVAL_NONE: pg_unreachable(); } @@ -1752,6 +1793,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, * - RS_INVAL_HORIZON: requires a snapshot <= the given horizon in the given * db; dboid may be InvalidOid for shared relations * - RS_INVAL_WAL_LEVEL: is logical + * - RS_INVAL_XID_AGE: slot's xmin or catalog_xmin has reached the age * * NB - this runs as part of checkpoint, so avoid raising errors if possible. */ diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 57d9de4dd9..6b5375909d 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -2954,6 +2954,16 @@ struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + { + {"max_slot_xid_age", PGC_SIGHUP, REPLICATION_SENDING, + gettext_noop("Age of the transaction ID at which a replication slot gets invalidated."), + gettext_noop("The transaction is the oldest transaction (including the one affecting the system catalogs) that a replication slot needs the database to retain.") + }, + &max_slot_xid_age, + 0, 0, 2000000000, + NULL, NULL, NULL + }, + { {"commit_delay", PGC_SUSET, WAL_SETTINGS, gettext_noop("Sets the delay in microseconds between transaction commit and " diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 2244ee52f7..b4c928b826 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -334,6 +334,7 @@ #wal_sender_timeout = 60s # in milliseconds; 0 disables #track_commit_timestamp = off # collect timestamp of transaction commit # (change requires restart) +#max_slot_xid_age = 0 # - Primary Server - diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 7f25a083ee..614ba0e30b 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -53,6 +53,8 @@ typedef enum ReplicationSlotInvalidationCause RS_INVAL_HORIZON, /* wal_level insufficient for slot */ RS_INVAL_WAL_LEVEL, + /* slot's xmin or catalog_xmin has reached the age */ + RS_INVAL_XID_AGE, } ReplicationSlotInvalidationCause; extern PGDLLIMPORT const char *const SlotInvalidationCauses[]; @@ -227,6 +229,7 @@ extern PGDLLIMPORT ReplicationSlot *MyReplicationSlot; /* GUCs */ extern PGDLLIMPORT int max_replication_slots; extern PGDLLIMPORT char *standby_slot_names; +extern PGDLLIMPORT int max_slot_xid_age; /* shmem initialization functions */ extern Size ReplicationSlotsShmemSize(void); diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index c67249500e..d698c3ec73 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -50,6 +50,7 @@ tests += { 't/039_end_of_wal.pl', 't/040_standby_failover_slots_sync.pl', 't/041_checkpoint_at_promote.pl', + 't/050_invalidate_slots.pl', ], }, } diff --git a/src/test/recovery/t/050_invalidate_slots.pl b/src/test/recovery/t/050_invalidate_slots.pl new file mode 100644 index 0000000000..2f482b56e8 --- /dev/null +++ b/src/test/recovery/t/050_invalidate_slots.pl @@ -0,0 +1,108 @@ +# Copyright (c) 2024, PostgreSQL Global Development Group + +# Test for replication slots invalidation +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Utils; +use PostgreSQL::Test::Cluster; +use Test::More; +use Time::HiRes qw(usleep); + +# Initialize primary node, setting wal-segsize to 1MB +my $primary = PostgreSQL::Test::Cluster->new('primary'); +$primary->init(allows_streaming => 1, extra => ['--wal-segsize=1']); +$primary->append_conf( + 'postgresql.conf', q{ +checkpoint_timeout = 1h +}); +$primary->start; +$primary->safe_psql( + 'postgres', qq[ + SELECT pg_create_physical_replication_slot('sb1_slot'); +]); + +# Take backup +my $backup_name = 'my_backup'; +$primary->backup($backup_name); + +# Create a standby linking to the primary using the replication slot +my $standby1 = PostgreSQL::Test::Cluster->new('standby1'); +$standby1->init_from_backup($primary, $backup_name, has_streaming => 1); + +# Enable hs_feedback. The slot should gain an xmin. We set the status interval +# so we'll see the results promptly. +$standby1->append_conf( + 'postgresql.conf', q{ +primary_slot_name = 'sb1_slot' +hot_standby_feedback = on +wal_receiver_status_interval = 1 +}); +$standby1->start; + +# Create some content on primary to move xmin +$primary->safe_psql('postgres', + "CREATE TABLE tab_int AS SELECT generate_series(1,10) AS a"); + +# Wait until standby has replayed enough data +$primary->wait_for_catchup($standby1); + +$primary->poll_query_until( + 'postgres', qq[ + SELECT xmin IS NOT NULL + FROM pg_catalog.pg_replication_slots + WHERE slot_name = 'sb1_slot'; +]) or die "Timed out waiting for slot xmin to advance"; + +$primary->safe_psql( + 'postgres', qq[ + ALTER SYSTEM SET max_slot_xid_age = 500; +]); +$primary->reload; + +# Stop standby to make the replication slot's xmin on primary to age +$standby1->stop; + +my $logstart = -s $primary->logfile; + +# Do some work to advance xmin +$primary->safe_psql( + 'postgres', q{ +do $$ +begin + for i in 10000..11000 loop + -- use an exception block so that each iteration eats an XID + begin + insert into tab_int values (i); + exception + when division_by_zero then null; + end; + end loop; +end$$; +}); + +my $invalidated = 0; +for (my $i = 0; $i < 10 * $PostgreSQL::Test::Utils::timeout_default; $i++) +{ + $primary->safe_psql('postgres', "CHECKPOINT"); + if ($primary->log_contains( + 'invalidating obsolete replication slot "sb1_slot"', $logstart)) + { + $invalidated = 1; + last; + } + usleep(100_000); +} +ok($invalidated, 'check that slot sb1_slot invalidation has been logged'); + +# Wait for the inactive replication slots to be invalidated. +$primary->poll_query_until( + 'postgres', qq[ + SELECT COUNT(slot_name) = 1 FROM pg_replication_slots + WHERE slot_name = 'sb1_slot' AND + invalidation_reason = 'xid_aged'; +]) + or die + "Timed out while waiting for replication slot sb1_slot to be invalidated"; + +done_testing(); -- 2.34.1 [application/x-patch] v9-0003-Track-inactive-replication-slot-information.patch (10.0K, ../../CALj2ACVRS2LZbycpM8rZT8QynR05AtvttOHfFOhczjyqU7zZSA@mail.gmail.com/4-v9-0003-Track-inactive-replication-slot-information.patch) download | inline diff: From 8349200b0bad1c8cda3e3f96034e7b63e3054d97 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Thu, 14 Mar 2024 14:04:00 +0000 Subject: [PATCH v9 3/4] Track inactive replication slot information Currently postgres doesn't track metrics like the time at which the slot became inactive, and the total number of times the slot became inactive in its lifetime. This commit adds two new metrics last_inactive_at of type timestamptz and inactive_count of type numeric to ReplicationSlotPersistentData. Whenever a slot becomes inactive, the current timestamp and inactive count are persisted to disk. These metrics are useful in the following ways: - To improve replication slot monitoring tools. For instance, one can build a monitoring tool that signals a) when replication slots is lying inactive for a day or so using last_inactive_at metric, b) when a replication slot is becoming inactive too frequently using last_inactive_at metric. - To implement timeout-based inactive replication slot management capability in postgres. Increases SLOT_VERSION due to the added two new metrics. --- doc/src/sgml/system-views.sgml | 20 +++++++++++++ src/backend/catalog/system_views.sql | 4 ++- src/backend/replication/slot.c | 43 ++++++++++++++++++++++------ src/backend/replication/slotfuncs.c | 15 +++++++++- src/include/catalog/pg_proc.dat | 6 ++-- src/include/replication/slot.h | 6 ++++ src/test/regress/expected/rules.out | 6 ++-- 7 files changed, 84 insertions(+), 16 deletions(-) diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml index f3fb5ba1b0..59cd1b5211 100644 --- a/doc/src/sgml/system-views.sgml +++ b/doc/src/sgml/system-views.sgml @@ -2750,6 +2750,26 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx ID of role </para></entry> </row> + + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>last_inactive_at</structfield> <type>timestamptz</type> + </para> + <para> + The time at which the slot became inactive. + <literal>NULL</literal> if the slot is currently actively being + used. + </para></entry> + </row> + + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>inactive_count</structfield> <type>numeric</type> + </para> + <para> + The total number of times the slot became inactive in its lifetime. + </para></entry> + </row> </tbody> </tgroup> </table> diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index cd22dad959..de9f1d5506 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1026,7 +1026,9 @@ CREATE VIEW pg_replication_slots AS L.conflicting, L.failover, L.synced, - L.invalidation_reason + L.invalidation_reason, + L.last_inactive_at, + L.inactive_count FROM pg_get_replication_slots() AS L LEFT JOIN pg_database D ON (L.datoid = D.oid); diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 2a1885da24..e606218673 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -130,7 +130,7 @@ StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1), sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize #define SLOT_MAGIC 0x1051CA1 /* format identifier */ -#define SLOT_VERSION 5 /* version for new files */ +#define SLOT_VERSION 6 /* version for new files */ /* Control array for replication slot management */ ReplicationSlotCtlData *ReplicationSlotCtl = NULL; @@ -400,6 +400,8 @@ ReplicationSlotCreate(const char *name, bool db_specific, slot->data.two_phase_at = InvalidXLogRecPtr; slot->data.failover = failover; slot->data.synced = synced; + slot->data.last_inactive_at = 0; + slot->data.inactive_count = 0; /* and then data only present in shared memory */ slot->just_dirtied = false; @@ -626,6 +628,17 @@ retry: if (am_walsender) { + if (s->data.persistency == RS_PERSISTENT) + { + SpinLockAcquire(&s->mutex); + s->data.last_inactive_at = 0; + SpinLockRelease(&s->mutex); + + /* Write this slot to disk */ + ReplicationSlotMarkDirty(); + ReplicationSlotSave(); + } + ereport(log_replication_commands ? LOG : DEBUG1, SlotIsLogical(s) ? errmsg("acquired logical replication slot \"%s\"", @@ -693,16 +706,20 @@ ReplicationSlotRelease(void) ConditionVariableBroadcast(&slot->active_cv); } - MyReplicationSlot = NULL; - - /* might not have been set when we've been a plain slot */ - LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE); - MyProc->statusFlags &= ~PROC_IN_LOGICAL_DECODING; - ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags; - LWLockRelease(ProcArrayLock); - if (am_walsender) { + if (slot->data.persistency == RS_PERSISTENT) + { + SpinLockAcquire(&slot->mutex); + slot->data.last_inactive_at = GetCurrentTimestamp(); + slot->data.inactive_count++; + SpinLockRelease(&slot->mutex); + + /* Write this slot to disk */ + ReplicationSlotMarkDirty(); + ReplicationSlotSave(); + } + ereport(log_replication_commands ? LOG : DEBUG1, is_logical ? errmsg("released logical replication slot \"%s\"", @@ -712,6 +729,14 @@ ReplicationSlotRelease(void) pfree(slotname); } + + MyReplicationSlot = NULL; + + /* might not have been set when we've been a plain slot */ + LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE); + MyProc->statusFlags &= ~PROC_IN_LOGICAL_DECODING; + ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags; + LWLockRelease(ProcArrayLock); } /* diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c index b5a638edea..4c7a120df1 100644 --- a/src/backend/replication/slotfuncs.c +++ b/src/backend/replication/slotfuncs.c @@ -239,7 +239,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS) Datum pg_get_replication_slots(PG_FUNCTION_ARGS) { -#define PG_GET_REPLICATION_SLOTS_COLS 18 +#define PG_GET_REPLICATION_SLOTS_COLS 20 ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; XLogRecPtr currlsn; int slotno; @@ -264,6 +264,7 @@ pg_get_replication_slots(PG_FUNCTION_ARGS) WALAvailability walstate; int i; ReplicationSlotInvalidationCause cause; + char buf[256]; if (!slot->in_use) continue; @@ -436,6 +437,18 @@ pg_get_replication_slots(PG_FUNCTION_ARGS) else values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]); + if (slot_contents.data.last_inactive_at > 0) + values[i++] = TimestampTzGetDatum(slot_contents.data.last_inactive_at); + else + nulls[i++] = true; + + /* Convert to numeric. */ + snprintf(buf, sizeof buf, UINT64_FORMAT, slot_contents.data.inactive_count); + values[i++] = DirectFunctionCall3(numeric_in, + CStringGetDatum(buf), + ObjectIdGetDatum(0), + Int32GetDatum(-1)); + Assert(i == PG_GET_REPLICATION_SLOTS_COLS); tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index e5dc1cbdb3..b26b53b714 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -11120,9 +11120,9 @@ proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f', proretset => 't', provolatile => 's', prorettype => 'record', proargtypes => '', - proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,bool,bool,text}', - proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}', - proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,failover,synced,invalidation_reason}', + proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,bool,bool,text,timestamptz,numeric}', + proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}', + proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,failover,synced,invalidation_reason,last_inactive_at,inactive_count}', prosrc => 'pg_get_replication_slots' }, { oid => '3786', descr => 'set up a logical replication slot', proname => 'pg_create_logical_replication_slot', provolatile => 'v', diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 614ba0e30b..780767a819 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -129,6 +129,12 @@ typedef struct ReplicationSlotPersistentData * for logical slots on the primary server. */ bool failover; + + /* When did this slot become inactive last time? */ + TimestampTz last_inactive_at; + + /* How many times the slot has been inactive? */ + uint64 inactive_count; } ReplicationSlotPersistentData; /* diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 055bec068d..c0bdfe76d8 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1476,8 +1476,10 @@ pg_replication_slots| SELECT l.slot_name, l.conflicting, l.failover, l.synced, - l.invalidation_reason - FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, failover, synced, invalidation_reason) + l.invalidation_reason, + l.last_inactive_at, + l.inactive_count + FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, failover, synced, invalidation_reason, last_inactive_at, inactive_count) LEFT JOIN pg_database d ON ((l.datoid = d.oid))); pg_roles| SELECT pg_authid.rolname, pg_authid.rolsuper, -- 2.34.1 [application/x-patch] v9-0004-Add-inactive_timeout-based-replication-slot-inval.patch (11.3K, ../../CALj2ACVRS2LZbycpM8rZT8QynR05AtvttOHfFOhczjyqU7zZSA@mail.gmail.com/5-v9-0004-Add-inactive_timeout-based-replication-slot-inval.patch) download | inline diff: From ff5007e261039bf951a93babdf407faeeed2bdeb Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Thu, 14 Mar 2024 14:06:41 +0000 Subject: [PATCH v9 4/4] Add inactive_timeout based replication slot invalidation Currently postgres has the ability to invalidate inactive replication slots based on the amount of WAL (set via max_slot_wal_keep_size GUC) that will be needed for the slots in case they become active. However, choosing a default value for max_slot_wal_keep_size is tricky. Because the amount of WAL a customer generates, and their allocated storage will vary greatly in production, making it difficult to pin down a one-size-fits-all value. It is often easy for developers to set a timeout of say 1 or 2 or 3 days, after which the inactive slots get dropped. To achieve the above, postgres uses replication slot metric inactive_at (the time at which the slot became inactive), and a new GUC inactive_replication_slot_timeout. The checkpointer then looks at all replication slots invalidating the inactive slots based on the timeout set. --- doc/src/sgml/config.sgml | 18 +++++ src/backend/access/transam/xlog.c | 10 +++ src/backend/replication/slot.c | 22 +++++- src/backend/utils/misc/guc_tables.c | 12 +++ src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/replication/slot.h | 3 + src/test/recovery/t/050_invalidate_slots.pl | 79 +++++++++++++++++++ 7 files changed, 144 insertions(+), 1 deletion(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 6dd54ffcb7..4b0b60a1ac 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -4565,6 +4565,24 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"' # Windows </listitem> </varlistentry> + <varlistentry id="guc-inactive-replication-slot-timeout" xreflabel="inactive_replication_slot_timeout"> + <term><varname>inactive_replication_slot_timeout</varname> (<type>integer</type>) + <indexterm> + <primary><varname>inactive_replication_slot_timeout</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + Invalidate replication slots that are inactive for longer than this + amount of time at the next checkpoint. If this value is specified + without units, it is taken as seconds. A value of zero (which is + default) disables the timeout mechanism. This parameter can only be + set in the <filename>postgresql.conf</filename> file or on the server + command line. + </para> + </listitem> + </varlistentry> + <varlistentry id="guc-track-commit-timestamp" xreflabel="track_commit_timestamp"> <term><varname>track_commit_timestamp</varname> (<type>boolean</type>) <indexterm> diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 36ae2ac6a4..166c3ed794 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -7152,6 +7152,11 @@ CreateCheckPoint(int flags) InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0, InvalidOid, InvalidTransactionId); + /* Invalidate inactive replication slots based on timeout */ + if (inactive_replication_slot_timeout > 0) + InvalidateObsoleteReplicationSlots(RS_INVAL_INACTIVE_TIMEOUT, 0, + InvalidOid, InvalidTransactionId); + /* * Delete old log files, those no longer needed for last checkpoint to * prevent the disk holding the xlog from growing full. @@ -7607,6 +7612,11 @@ CreateRestartPoint(int flags) InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0, InvalidOid, InvalidTransactionId); + /* Invalidate inactive replication slots based on timeout */ + if (inactive_replication_slot_timeout > 0) + InvalidateObsoleteReplicationSlots(RS_INVAL_INACTIVE_TIMEOUT, 0, + InvalidOid, InvalidTransactionId); + /* * Retreat _logSegNo using the current end of xlog replayed or received, * whichever is later. diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index e606218673..37498d3d98 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -108,10 +108,11 @@ const char *const SlotInvalidationCauses[] = { [RS_INVAL_HORIZON] = "rows_removed", [RS_INVAL_WAL_LEVEL] = "wal_level_insufficient", [RS_INVAL_XID_AGE] = "xid_aged", + [RS_INVAL_INACTIVE_TIMEOUT] = "inactive_timeout", }; /* Maximum number of invalidation causes */ -#define RS_INVAL_MAX_CAUSES RS_INVAL_XID_AGE +#define RS_INVAL_MAX_CAUSES RS_INVAL_INACTIVE_TIMEOUT StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1), "array length mismatch"); @@ -142,6 +143,7 @@ ReplicationSlot *MyReplicationSlot = NULL; int max_replication_slots = 10; /* the maximum number of replication * slots */ int max_slot_xid_age = 0; +int inactive_replication_slot_timeout = 0; /* * This GUC lists streaming replication standby server slot names that @@ -1513,6 +1515,9 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, case RS_INVAL_XID_AGE: appendStringInfoString(&err_detail, _("The replication slot's xmin or catalog_xmin reached the age specified by max_slot_xid_age.")); break; + case RS_INVAL_INACTIVE_TIMEOUT: + appendStringInfoString(&err_detail, _("The slot has been inactive for more than the time specified by inactive_replication_slot_timeout.")); + break; case RS_INVAL_NONE: pg_unreachable(); } @@ -1665,6 +1670,20 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, } } break; + case RS_INVAL_INACTIVE_TIMEOUT: + if (s->data.last_inactive_at > 0) + { + TimestampTz now; + + Assert(s->data.persistency == RS_PERSISTENT); + Assert(s->active_pid == 0); + + now = GetCurrentTimestamp(); + if (TimestampDifferenceExceeds(s->data.last_inactive_at, now, + inactive_replication_slot_timeout * 1000)) + conflict = cause; + } + break; case RS_INVAL_NONE: pg_unreachable(); } @@ -1819,6 +1838,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, * db; dboid may be InvalidOid for shared relations * - RS_INVAL_WAL_LEVEL: is logical * - RS_INVAL_XID_AGE: slot's xmin or catalog_xmin has reached the age + * - RS_INVAL_INACTIVE_TIMEOUT: inactive slot timeout occurs * * NB - this runs as part of checkpoint, so avoid raising errors if possible. */ diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 6b5375909d..6caf40d51e 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -2964,6 +2964,18 @@ struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + { + {"inactive_replication_slot_timeout", PGC_SIGHUP, REPLICATION_SENDING, + gettext_noop("Sets the amount of time to wait before invalidating an " + "inactive replication slot."), + NULL, + GUC_UNIT_S + }, + &inactive_replication_slot_timeout, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, + { {"commit_delay", PGC_SUSET, WAL_SETTINGS, gettext_noop("Sets the delay in microseconds between transaction commit and " diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index b4c928b826..7f2a3e41f1 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -261,6 +261,7 @@ #recovery_prefetch = try # prefetch pages referenced in the WAL? #wal_decode_buffer_size = 512kB # lookahead window used for prefetching # (change requires restart) +#inactive_replication_slot_timeout = 0 # in seconds; 0 disables # - Archiving - diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 780767a819..8378d7c913 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -55,6 +55,8 @@ typedef enum ReplicationSlotInvalidationCause RS_INVAL_WAL_LEVEL, /* slot's xmin or catalog_xmin has reached the age */ RS_INVAL_XID_AGE, + /* inactive slot timeout has occurred */ + RS_INVAL_INACTIVE_TIMEOUT, } ReplicationSlotInvalidationCause; extern PGDLLIMPORT const char *const SlotInvalidationCauses[]; @@ -236,6 +238,7 @@ extern PGDLLIMPORT ReplicationSlot *MyReplicationSlot; extern PGDLLIMPORT int max_replication_slots; extern PGDLLIMPORT char *standby_slot_names; extern PGDLLIMPORT int max_slot_xid_age; +extern PGDLLIMPORT int inactive_replication_slot_timeout; /* shmem initialization functions */ extern Size ReplicationSlotsShmemSize(void); diff --git a/src/test/recovery/t/050_invalidate_slots.pl b/src/test/recovery/t/050_invalidate_slots.pl index 2f482b56e8..4c66dd4a4e 100644 --- a/src/test/recovery/t/050_invalidate_slots.pl +++ b/src/test/recovery/t/050_invalidate_slots.pl @@ -105,4 +105,83 @@ $primary->poll_query_until( or die "Timed out while waiting for replication slot sb1_slot to be invalidated"; +$primary->safe_psql( + 'postgres', qq[ + SELECT pg_create_physical_replication_slot('sb2_slot'); +]); + +$primary->safe_psql( + 'postgres', qq[ + ALTER SYSTEM SET max_slot_xid_age = 0; +]); +$primary->reload; + +# Create a standby linking to the primary using the replication slot +my $standby2 = PostgreSQL::Test::Cluster->new('standby2'); +$standby2->init_from_backup($primary, $backup_name, has_streaming => 1); +$standby2->append_conf( + 'postgresql.conf', q{ +primary_slot_name = 'sb2_slot' +}); +$standby2->start; + +# Wait until standby has replayed enough data +$primary->wait_for_catchup($standby2); + +# The inactive replication slot info should be null when the slot is active +my $result = $primary->safe_psql( + 'postgres', qq[ + SELECT last_inactive_at IS NULL, inactive_count = 0 AS OK + FROM pg_replication_slots WHERE slot_name = 'sb2_slot'; +]); +is($result, "t|t", + 'check the inactive replication slot info for an active slot'); + +# Set timeout so that the next checkpoint will invalidate the inactive +# replication slot. +$primary->safe_psql( + 'postgres', qq[ + ALTER SYSTEM SET inactive_replication_slot_timeout TO '1s'; +]); +$primary->reload; + +$logstart = -s $primary->logfile; + +# Stop standby to make the replication slot on primary inactive +$standby2->stop; + +# Wait for the inactive replication slot info to be updated +$primary->poll_query_until( + 'postgres', qq[ + SELECT COUNT(slot_name) = 1 FROM pg_replication_slots + WHERE last_inactive_at IS NOT NULL AND + inactive_count = 1 AND slot_name = 'sb2_slot'; +]) + or die + "Timed out while waiting for inactive replication slot info to be updated"; + +$invalidated = 0; +for (my $i = 0; $i < 10 * $PostgreSQL::Test::Utils::timeout_default; $i++) +{ + $primary->safe_psql('postgres', "CHECKPOINT"); + if ($primary->log_contains( + 'invalidating obsolete replication slot "sb2_slot"', $logstart)) + { + $invalidated = 1; + last; + } + usleep(100_000); +} +ok($invalidated, 'check that slot sb2_slot invalidation has been logged'); + +# Wait for the inactive replication slots to be invalidated. +$primary->poll_query_until( + 'postgres', qq[ + SELECT COUNT(slot_name) = 1 FROM pg_replication_slots + WHERE slot_name = 'sb2_slot' AND + invalidation_reason = 'inactive_timeout'; +]) + or die + "Timed out while waiting for inactive replication slot sb2_slot to be invalidated"; + done_testing(); -- 2.34.1 ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-15 04:44 shveta malik <[email protected]> parent: Bharath Rupireddy <[email protected]> 1 sibling, 1 reply; 18+ messages in thread From: shveta malik @ 2024-03-15 04:44 UTC (permalink / raw) To: Bharath Rupireddy <[email protected]>; +Cc: Amit Kapila <[email protected]>; Nathan Bossart <[email protected]>; Bertrand Drouvot <[email protected]>; PostgreSQL Hackers <[email protected]>; shveta malik <[email protected]> On Thu, Mar 14, 2024 at 7:58 PM Bharath Rupireddy <[email protected]> wrote: > > On Thu, Mar 14, 2024 at 12:24 PM Amit Kapila <[email protected]> wrote: > > > > On Wed, Mar 13, 2024 at 9:24 PM Bharath Rupireddy > > > > > > Yes, there will be some sort of duplicity if we emit conflict_reason > > > as a text field. However, I still think the better way is to turn > > > conflict_reason text to conflict boolean and set it to true only on > > > rows_removed and wal_level_insufficient invalidations. When conflict > > > boolean is true, one (including all the tests that we've added > > > recently) can look for invalidation_reason text field for the reason. > > > This sounds reasonable to me as opposed to we just mentioning in the > > > docs that "if invalidation_reason is rows_removed or > > > wal_level_insufficient it's the reason for conflict with recovery". +1 on maintaining both conflicting and invalidation_reason > > Fair point. I think we can go either way. Bertrand, Nathan, and > > others, do you have an opinion on this matter? > > While we wait to hear from others on this, I'm attaching the v9 patch > set implementing the above idea (check 0001 patch). Please have a > look. I'll come back to the other review comments soon. Thanks for the patch. JFYI, patch09 does not apply to HEAD, some recent commit caused the conflict. Some trivial comments on patch001 (yet to review other patches) 1) info.c: - "%s as caught_up, conflict_reason IS NOT NULL as invalid " + "%s as caught_up, invalidation_reason IS NOT NULL as invalid " Can we revert back to 'conflicting as invalid' since it is a query for logical slots only. 2) 040_standby_failover_slots_sync.pl: - q{SELECT conflict_reason IS NULL AND synced AND NOT temporary FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';} + q{SELECT invalidation_reason IS NULL AND synced AND NOT temporary FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';} Here too, can we have 'NOT conflicting' instead of ' invalidation_reason IS NULL' as it is a logical slot test. thanks Shveta ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-15 07:19 shveta malik <[email protected]> parent: Bharath Rupireddy <[email protected]> 1 sibling, 1 reply; 18+ messages in thread From: shveta malik @ 2024-03-15 07:19 UTC (permalink / raw) To: Bharath Rupireddy <[email protected]>; +Cc: Amit Kapila <[email protected]>; Nathan Bossart <[email protected]>; Bertrand Drouvot <[email protected]>; PostgreSQL Hackers <[email protected]>; shveta malik <[email protected]> On Thu, Mar 14, 2024 at 7:58 PM Bharath Rupireddy <[email protected]> wrote: > > While we wait to hear from others on this, I'm attaching the v9 patch > set implementing the above idea (check 0001 patch). Please have a > look. I'll come back to the other review comments soon. > patch002: 1) I would like to understand the purpose of 'inactive_count'? Is it only for users for monitoring purposes? We are not using it anywhere internally. I shutdown the instance 5 times and found that 'inactive_count' became 5 for all the slots created on that instance. Is this intentional? I mean we can not really use them if the instance is down. I felt it should increment the inactive_count only if during the span of instance, they were actually inactive i.e. no streaming or replication happening through them. 2) slot.c: + case RS_INVAL_XID_AGE: + { + if (TransactionIdIsNormal(s->data.xmin)) + { + .......... + } + if (TransactionIdIsNormal(s->data.catalog_xmin)) + { + .......... + } + } Can we optimize this code? It has duplicate code for processing s->data.catalog_xmin and s->data.xmin. Can we create a sub-function for this purpose and call it twice here? thanks Shveta ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-15 12:05 Bharath Rupireddy <[email protected]> parent: shveta malik <[email protected]> 0 siblings, 0 replies; 18+ messages in thread From: Bharath Rupireddy @ 2024-03-15 12:05 UTC (permalink / raw) To: shveta malik <[email protected]>; +Cc: Amit Kapila <[email protected]>; Nathan Bossart <[email protected]>; Bertrand Drouvot <[email protected]>; PostgreSQL Hackers <[email protected]> On Fri, Mar 15, 2024 at 10:15 AM shveta malik <[email protected]> wrote: > > > > > wal_level_insufficient it's the reason for conflict with recovery". > > +1 on maintaining both conflicting and invalidation_reason Thanks. > Thanks for the patch. JFYI, patch09 does not apply to HEAD, some > recent commit caused the conflict. Yep, the conflict is in src/test/recovery/meson.build and is because of e6927270cd18d535b77cbe79c55c6584351524be. > Some trivial comments on patch001 (yet to review other patches) Thanks for looking into this. > 1) > info.c: > > - "%s as caught_up, conflict_reason IS NOT NULL as invalid " > + "%s as caught_up, invalidation_reason IS NOT NULL as invalid " > > Can we revert back to 'conflicting as invalid' since it is a query for > logical slots only. I guess, no. There the intention is to check for invalid logical slots not just for the conflicting ones. The logical slots can get invalidated due to other reasons as well. > 2) > 040_standby_failover_slots_sync.pl: > > - q{SELECT conflict_reason IS NULL AND synced AND NOT temporary FROM > pg_replication_slots WHERE slot_name = 'lsub1_slot';} > + q{SELECT invalidation_reason IS NULL AND synced AND NOT temporary > FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';} > > Here too, can we have 'NOT conflicting' instead of ' > invalidation_reason IS NULL' as it is a logical slot test. I guess no. The tests are ensuring the slot on the standby isn't invalidated. In general, one needs to use the 'conflicting' column from pg_replication_slots when the intention is to look for reasons for conflicts, otherwise use the 'invalidation_reason' column for invalidations. Please see the attached v10 patch set after resolving the merge conflict and fixing an indentation warning in the TAP test file. -- Bharath Rupireddy PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com Attachments: [application/x-patch] v10-0001-Track-invalidation_reason-in-pg_replication_slot.patch (19.8K, ../../CALj2ACU=Q5JG1VQFWFQe0VjPkOu2kt-6gpq7z1Ed9+JX_BR-tQ@mail.gmail.com/2-v10-0001-Track-invalidation_reason-in-pg_replication_slot.patch) download | inline diff: From 41290be4eb1562cf10313e3eda19fcbbf392088f Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Fri, 15 Mar 2024 05:47:44 +0000 Subject: [PATCH v10 1/4] Track invalidation_reason in pg_replication_slots Up until now, reason for replication slot invalidation is not tracked in pg_replication_slots. A recent commit 007693f2a added conflict_reason to show the reasons for slot invalidation, but only for logical slots. This commit adds a new column to show invalidation reasons for both physical and logical slots. And, this commit also turns conflict_reason text column to conflicting boolean column (effectively reverting commit 007693f2a). One now can look at the new invalidation_reason column for logical slots conflict with recovery. --- doc/src/sgml/ref/pgupgrade.sgml | 4 +- doc/src/sgml/system-views.sgml | 63 +++++++++++-------- src/backend/catalog/system_views.sql | 5 +- src/backend/replication/logical/slotsync.c | 2 +- src/backend/replication/slot.c | 8 +-- src/backend/replication/slotfuncs.c | 25 +++++--- src/bin/pg_upgrade/info.c | 4 +- src/include/catalog/pg_proc.dat | 6 +- src/include/replication/slot.h | 2 +- .../t/035_standby_logical_decoding.pl | 35 ++++++----- .../t/040_standby_failover_slots_sync.pl | 4 +- src/test/regress/expected/rules.out | 7 ++- 12 files changed, 95 insertions(+), 70 deletions(-) diff --git a/doc/src/sgml/ref/pgupgrade.sgml b/doc/src/sgml/ref/pgupgrade.sgml index 58c6c2df8b..8de52bf752 100644 --- a/doc/src/sgml/ref/pgupgrade.sgml +++ b/doc/src/sgml/ref/pgupgrade.sgml @@ -453,8 +453,8 @@ make prefix=/usr/local/pgsql.new install <para> All slots on the old cluster must be usable, i.e., there are no slots whose - <link linkend="view-pg-replication-slots">pg_replication_slots</link>.<structfield>conflict_reason</structfield> - is not <literal>NULL</literal>. + <link linkend="view-pg-replication-slots">pg_replication_slots</link>.<structfield>conflicting</structfield> + is not <literal>true</literal>. </para> </listitem> <listitem> diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml index be90edd0e2..f3fb5ba1b0 100644 --- a/doc/src/sgml/system-views.sgml +++ b/doc/src/sgml/system-views.sgml @@ -2525,34 +2525,13 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx <row> <entry role="catalog_table_entry"><para role="column_definition"> - <structfield>conflict_reason</structfield> <type>text</type> + <structfield>conflicting</structfield> <type>bool</type> </para> <para> - The reason for the logical slot's conflict with recovery. It is always - NULL for physical slots, as well as for logical slots which are not - invalidated. The non-NULL values indicate that the slot is marked - as invalidated. Possible values are: - <itemizedlist spacing="compact"> - <listitem> - <para> - <literal>wal_removed</literal> means that the required WAL has been - removed. - </para> - </listitem> - <listitem> - <para> - <literal>rows_removed</literal> means that the required rows have - been removed. - </para> - </listitem> - <listitem> - <para> - <literal>wal_level_insufficient</literal> means that the - primary doesn't have a <xref linkend="guc-wal-level"/> sufficient to - perform logical decoding. - </para> - </listitem> - </itemizedlist> + True if this logical slot conflicted with recovery (and so is now + invalidated). When this column is true, check + <structfield>invalidation_reason</structfield> column for the conflict + reason. </para></entry> </row> @@ -2581,6 +2560,38 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx </para></entry> </row> + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>invalidation_reason</structfield> <type>text</type> + </para> + <para> + The reason for the slot's invalidation. <literal>NULL</literal> if the + slot is currently actively being used. The non-NULL values indicate that + the slot is marked as invalidated. Possible values are: + <itemizedlist spacing="compact"> + <listitem> + <para> + <literal>wal_removed</literal> means that the required WAL has been + removed. + </para> + </listitem> + <listitem> + <para> + <literal>rows_removed</literal> means that the required rows have + been removed. + </para> + </listitem> + <listitem> + <para> + <literal>wal_level_insufficient</literal> means that the + primary doesn't have a <xref linkend="guc-wal-level"/> sufficient to + perform logical decoding. + </para> + </listitem> + </itemizedlist> + </para></entry> + </row> + </tbody> </tgroup> </table> diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 04227a72d1..cd22dad959 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1023,9 +1023,10 @@ CREATE VIEW pg_replication_slots AS L.wal_status, L.safe_wal_size, L.two_phase, - L.conflict_reason, + L.conflicting, L.failover, - L.synced + L.synced, + L.invalidation_reason FROM pg_get_replication_slots() AS L LEFT JOIN pg_database D ON (L.datoid = D.oid); diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index 5074c8409f..260632cfdd 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -668,7 +668,7 @@ synchronize_slots(WalReceiverConn *wrconn) bool started_tx = false; const char *query = "SELECT slot_name, plugin, confirmed_flush_lsn," " restart_lsn, catalog_xmin, two_phase, failover," - " database, conflict_reason" + " database, invalidation_reason" " FROM pg_catalog.pg_replication_slots" " WHERE failover and NOT temporary"; diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 91ca397857..4f1a17f6ce 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -2356,21 +2356,21 @@ RestoreSlotFromDisk(const char *name) } /* - * Maps a conflict reason for a replication slot to + * Maps a invalidation reason for a replication slot to * ReplicationSlotInvalidationCause. */ ReplicationSlotInvalidationCause -GetSlotInvalidationCause(const char *conflict_reason) +GetSlotInvalidationCause(const char *invalidation_reason) { ReplicationSlotInvalidationCause cause; ReplicationSlotInvalidationCause result = RS_INVAL_NONE; bool found PG_USED_FOR_ASSERTS_ONLY = false; - Assert(conflict_reason); + Assert(invalidation_reason); for (cause = RS_INVAL_NONE; cause <= RS_INVAL_MAX_CAUSES; cause++) { - if (strcmp(SlotInvalidationCauses[cause], conflict_reason) == 0) + if (strcmp(SlotInvalidationCauses[cause], invalidation_reason) == 0) { found = true; result = cause; diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c index ad79e1fccd..b5a638edea 100644 --- a/src/backend/replication/slotfuncs.c +++ b/src/backend/replication/slotfuncs.c @@ -239,7 +239,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS) Datum pg_get_replication_slots(PG_FUNCTION_ARGS) { -#define PG_GET_REPLICATION_SLOTS_COLS 17 +#define PG_GET_REPLICATION_SLOTS_COLS 18 ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; XLogRecPtr currlsn; int slotno; @@ -263,6 +263,7 @@ pg_get_replication_slots(PG_FUNCTION_ARGS) bool nulls[PG_GET_REPLICATION_SLOTS_COLS]; WALAvailability walstate; int i; + ReplicationSlotInvalidationCause cause; if (!slot->in_use) continue; @@ -409,22 +410,32 @@ pg_get_replication_slots(PG_FUNCTION_ARGS) values[i++] = BoolGetDatum(slot_contents.data.two_phase); - if (slot_contents.data.database == InvalidOid) + cause = slot_contents.data.invalidated; + + if (SlotIsPhysical(&slot_contents)) nulls[i++] = true; else { - ReplicationSlotInvalidationCause cause = slot_contents.data.invalidated; - - if (cause == RS_INVAL_NONE) - nulls[i++] = true; + /* + * rows_removed and wal_level_insufficient are only two reasons + * for the logical slot's conflict with recovery. + */ + if (cause == RS_INVAL_HORIZON || + cause == RS_INVAL_WAL_LEVEL) + values[i++] = BoolGetDatum(true); else - values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]); + values[i++] = BoolGetDatum(false); } values[i++] = BoolGetDatum(slot_contents.data.failover); values[i++] = BoolGetDatum(slot_contents.data.synced); + if (cause == RS_INVAL_NONE) + nulls[i++] = true; + else + values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]); + Assert(i == PG_GET_REPLICATION_SLOTS_COLS); tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, diff --git a/src/bin/pg_upgrade/info.c b/src/bin/pg_upgrade/info.c index b5b8d11602..34a157f792 100644 --- a/src/bin/pg_upgrade/info.c +++ b/src/bin/pg_upgrade/info.c @@ -676,13 +676,13 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check) * removed. */ res = executeQueryOrDie(conn, "SELECT slot_name, plugin, two_phase, failover, " - "%s as caught_up, conflict_reason IS NOT NULL as invalid " + "%s as caught_up, invalidation_reason IS NOT NULL as invalid " "FROM pg_catalog.pg_replication_slots " "WHERE slot_type = 'logical' AND " "database = current_database() AND " "temporary IS FALSE;", live_check ? "FALSE" : - "(CASE WHEN conflict_reason IS NOT NULL THEN FALSE " + "(CASE WHEN conflicting THEN FALSE " "ELSE (SELECT pg_catalog.binary_upgrade_logical_slot_has_caught_up(slot_name)) " "END)"); diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 700f7daf7b..63fd0b4cd7 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -11123,9 +11123,9 @@ proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f', proretset => 't', provolatile => 's', prorettype => 'record', proargtypes => '', - proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,text,bool,bool}', - proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}', - proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflict_reason,failover,synced}', + proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,bool,bool,text}', + proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}', + proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,failover,synced,invalidation_reason}', prosrc => 'pg_get_replication_slots' }, { oid => '3786', descr => 'set up a logical replication slot', proname => 'pg_create_logical_replication_slot', provolatile => 'v', diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 425effad21..7f25a083ee 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -273,7 +273,7 @@ extern void CheckPointReplicationSlots(bool is_shutdown); extern void CheckSlotRequirements(void); extern void CheckSlotPermissions(void); extern ReplicationSlotInvalidationCause - GetSlotInvalidationCause(const char *conflict_reason); + GetSlotInvalidationCause(const char *invalidation_reason); extern bool SlotExistsInStandbySlotNames(const char *slot_name); extern bool StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel); diff --git a/src/test/recovery/t/035_standby_logical_decoding.pl b/src/test/recovery/t/035_standby_logical_decoding.pl index 88b03048c4..8d6740c734 100644 --- a/src/test/recovery/t/035_standby_logical_decoding.pl +++ b/src/test/recovery/t/035_standby_logical_decoding.pl @@ -168,7 +168,7 @@ sub change_hot_standby_feedback_and_wait_for_xmins } } -# Check conflict_reason in pg_replication_slots. +# Check reason for conflict in pg_replication_slots. sub check_slots_conflict_reason { my ($slot_prefix, $reason) = @_; @@ -178,15 +178,15 @@ sub check_slots_conflict_reason $res = $node_standby->safe_psql( 'postgres', qq( - select conflict_reason from pg_replication_slots where slot_name = '$active_slot';)); + select invalidation_reason from pg_replication_slots where slot_name = '$active_slot' and conflicting;)); - is($res, "$reason", "$active_slot conflict_reason is $reason"); + is($res, "$reason", "$active_slot reason for conflict is $reason"); $res = $node_standby->safe_psql( 'postgres', qq( - select conflict_reason from pg_replication_slots where slot_name = '$inactive_slot';)); + select invalidation_reason from pg_replication_slots where slot_name = '$inactive_slot' and conflicting;)); - is($res, "$reason", "$inactive_slot conflict_reason is $reason"); + is($res, "$reason", "$inactive_slot reason for conflict is $reason"); } # Drop the slots, re-create them, change hot_standby_feedback, @@ -293,13 +293,13 @@ $node_primary->safe_psql('testdb', qq[SELECT * FROM pg_create_physical_replication_slot('$primary_slotname');] ); -# Check conflict_reason is NULL for physical slot +# Check conflicting is NULL for physical slot $res = $node_primary->safe_psql( 'postgres', qq[ - SELECT conflict_reason is null FROM pg_replication_slots where slot_name = '$primary_slotname';] + SELECT conflicting is null FROM pg_replication_slots where slot_name = '$primary_slotname';] ); -is($res, 't', "Physical slot reports conflict_reason as NULL"); +is($res, 't', "Physical slot reports conflicting as NULL"); my $backup_name = 'b1'; $node_primary->backup($backup_name); @@ -524,7 +524,7 @@ $node_primary->wait_for_replay_catchup($node_standby); # Check invalidation in the logfile and in pg_stat_database_conflicts check_for_invalidation('vacuum_full_', 1, 'with vacuum FULL on pg_class'); -# Verify conflict_reason is 'rows_removed' in pg_replication_slots +# Verify reason for conflict is 'rows_removed' in pg_replication_slots check_slots_conflict_reason('vacuum_full_', 'rows_removed'); # Ensure that replication slot stats are not removed after invalidation. @@ -551,7 +551,7 @@ change_hot_standby_feedback_and_wait_for_xmins(1, 1); ################################################## $node_standby->restart; -# Verify conflict_reason is retained across a restart. +# Verify reason for conflict is retained across a restart. check_slots_conflict_reason('vacuum_full_', 'rows_removed'); ################################################## @@ -560,7 +560,8 @@ check_slots_conflict_reason('vacuum_full_', 'rows_removed'); # Get the restart_lsn from an invalidated slot my $restart_lsn = $node_standby->safe_psql('postgres', - "SELECT restart_lsn from pg_replication_slots WHERE slot_name = 'vacuum_full_activeslot' and conflict_reason is not null;" + "SELECT restart_lsn FROM pg_replication_slots + WHERE slot_name = 'vacuum_full_activeslot' AND conflicting;" ); chomp($restart_lsn); @@ -611,7 +612,7 @@ $node_primary->wait_for_replay_catchup($node_standby); # Check invalidation in the logfile and in pg_stat_database_conflicts check_for_invalidation('row_removal_', $logstart, 'with vacuum on pg_class'); -# Verify conflict_reason is 'rows_removed' in pg_replication_slots +# Verify reason for conflict is 'rows_removed' in pg_replication_slots check_slots_conflict_reason('row_removal_', 'rows_removed'); $handle = @@ -647,7 +648,7 @@ $node_primary->wait_for_replay_catchup($node_standby); check_for_invalidation('shared_row_removal_', $logstart, 'with vacuum on pg_authid'); -# Verify conflict_reason is 'rows_removed' in pg_replication_slots +# Verify reason for conflict is 'rows_removed' in pg_replication_slots check_slots_conflict_reason('shared_row_removal_', 'rows_removed'); $handle = make_slot_active($node_standby, 'shared_row_removal_', 0, \$stdout, @@ -700,8 +701,8 @@ ok( $node_standby->poll_query_until( is( $node_standby->safe_psql( 'postgres', q[select bool_or(conflicting) from - (select conflict_reason is not NULL as conflicting - from pg_replication_slots WHERE slot_type = 'logical')]), + (select conflicting from pg_replication_slots + where slot_type = 'logical')]), 'f', 'Logical slots are reported as non conflicting'); @@ -739,7 +740,7 @@ $node_primary->wait_for_replay_catchup($node_standby); # Check invalidation in the logfile and in pg_stat_database_conflicts check_for_invalidation('pruning_', $logstart, 'with on-access pruning'); -# Verify conflict_reason is 'rows_removed' in pg_replication_slots +# Verify reason for conflict is 'rows_removed' in pg_replication_slots check_slots_conflict_reason('pruning_', 'rows_removed'); $handle = make_slot_active($node_standby, 'pruning_', 0, \$stdout, \$stderr); @@ -783,7 +784,7 @@ $node_primary->wait_for_replay_catchup($node_standby); # Check invalidation in the logfile and in pg_stat_database_conflicts check_for_invalidation('wal_level_', $logstart, 'due to wal_level'); -# Verify conflict_reason is 'wal_level_insufficient' in pg_replication_slots +# Verify reason for conflict is 'wal_level_insufficient' in pg_replication_slots check_slots_conflict_reason('wal_level_', 'wal_level_insufficient'); $handle = diff --git a/src/test/recovery/t/040_standby_failover_slots_sync.pl b/src/test/recovery/t/040_standby_failover_slots_sync.pl index 0ea1f3d323..f47bfd78eb 100644 --- a/src/test/recovery/t/040_standby_failover_slots_sync.pl +++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl @@ -228,7 +228,7 @@ $standby1->safe_psql('postgres', "CHECKPOINT"); # Check if the synced slot is invalidated is( $standby1->safe_psql( 'postgres', - q{SELECT conflict_reason = 'wal_removed' FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';} + q{SELECT invalidation_reason = 'wal_removed' FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';} ), "t", 'synchronized slot has been invalidated'); @@ -274,7 +274,7 @@ $standby1->wait_for_log(qr/dropped replication slot "lsub1_slot" of dbid [0-9]+/ # flagged as 'synced' is( $standby1->safe_psql( 'postgres', - q{SELECT conflict_reason IS NULL AND synced AND NOT temporary FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';} + q{SELECT invalidation_reason IS NULL AND synced AND NOT temporary FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';} ), "t", 'logical slot is re-synced'); diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 0cd2c64fca..055bec068d 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1473,10 +1473,11 @@ pg_replication_slots| SELECT l.slot_name, l.wal_status, l.safe_wal_size, l.two_phase, - l.conflict_reason, + l.conflicting, l.failover, - l.synced - FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflict_reason, failover, synced) + l.synced, + l.invalidation_reason + FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, failover, synced, invalidation_reason) LEFT JOIN pg_database d ON ((l.datoid = d.oid))); pg_roles| SELECT pg_authid.rolname, pg_authid.rolsuper, -- 2.34.1 [application/x-patch] v10-0002-Add-XID-age-based-replication-slot-invalidation.patch (12.9K, ../../CALj2ACU=Q5JG1VQFWFQe0VjPkOu2kt-6gpq7z1Ed9+JX_BR-tQ@mail.gmail.com/3-v10-0002-Add-XID-age-based-replication-slot-invalidation.patch) download | inline diff: From 98b48e257847299b676e90af25c26f9d4150669a Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Fri, 15 Mar 2024 05:48:01 +0000 Subject: [PATCH v10 2/4] Add XID age based replication slot invalidation Up until now, postgres has the ability to invalidate inactive replication slots based on the amount of WAL (set via max_slot_wal_keep_size GUC) that will be needed for the slots in case they become active. However, choosing a default value for max_slot_wal_keep_size is tricky. Because the amount of WAL a customer generates, and their allocated storage will vary greatly in production, making it difficult to pin down a one-size-fits-all value. It is often easy for developers to set an XID age (age of slot's xmin or catalog_xmin) of say 1 or 1.5 billion, after which the slots get invalidated. To achieve the above, postgres uses replication slot xmin (the oldest transaction that this slot needs the database to retain) or catalog_xmin (the oldest transaction affecting the system catalogs that this slot needs the database to retain), and a new GUC max_slot_xid_age. The checkpointer then looks at all replication slots invalidating the slots based on the age set. --- doc/src/sgml/config.sgml | 21 ++++ src/backend/access/transam/xlog.c | 10 ++ src/backend/replication/slot.c | 44 ++++++- src/backend/utils/misc/guc_tables.c | 10 ++ src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/replication/slot.h | 3 + src/test/recovery/meson.build | 1 + src/test/recovery/t/050_invalidate_slots.pl | 108 ++++++++++++++++++ 8 files changed, 197 insertions(+), 1 deletion(-) create mode 100644 src/test/recovery/t/050_invalidate_slots.pl diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 65a6e6c408..6dd54ffcb7 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -4544,6 +4544,27 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"' # Windows </listitem> </varlistentry> + <varlistentry id="guc-max-slot-xid-age" xreflabel="max_slot_xid_age"> + <term><varname>max_slot_xid_age</varname> (<type>integer</type>) + <indexterm> + <primary><varname>max_slot_xid_age</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + Invalidate replication slots whose <literal>xmin</literal> (the oldest + transaction that this slot needs the database to retain) or + <literal>catalog_xmin</literal> (the oldest transaction affecting the + system catalogs that this slot needs the database to retain) has reached + the age specified by this setting. A value of zero (which is default) + disables this feature. Users can set this value anywhere from zero to + two billion. This parameter can only be set in the + <filename>postgresql.conf</filename> file or on the server command + line. + </para> + </listitem> + </varlistentry> + <varlistentry id="guc-track-commit-timestamp" xreflabel="track_commit_timestamp"> <term><varname>track_commit_timestamp</varname> (<type>boolean</type>) <indexterm> diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 20a5f86209..36ae2ac6a4 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -7147,6 +7147,11 @@ CreateCheckPoint(int flags) if (PriorRedoPtr != InvalidXLogRecPtr) UpdateCheckPointDistanceEstimate(RedoRecPtr - PriorRedoPtr); + /* Invalidate replication slots based on xmin or catalog_xmin age */ + if (max_slot_xid_age > 0) + InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0, + InvalidOid, InvalidTransactionId); + /* * Delete old log files, those no longer needed for last checkpoint to * prevent the disk holding the xlog from growing full. @@ -7597,6 +7602,11 @@ CreateRestartPoint(int flags) */ XLByteToSeg(RedoRecPtr, _logSegNo, wal_segment_size); + /* Invalidate replication slots based on xmin or catalog_xmin age */ + if (max_slot_xid_age > 0) + InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0, + InvalidOid, InvalidTransactionId); + /* * Retreat _logSegNo using the current end of xlog replayed or received, * whichever is later. diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 4f1a17f6ce..2a1885da24 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -107,10 +107,11 @@ const char *const SlotInvalidationCauses[] = { [RS_INVAL_WAL_REMOVED] = "wal_removed", [RS_INVAL_HORIZON] = "rows_removed", [RS_INVAL_WAL_LEVEL] = "wal_level_insufficient", + [RS_INVAL_XID_AGE] = "xid_aged", }; /* Maximum number of invalidation causes */ -#define RS_INVAL_MAX_CAUSES RS_INVAL_WAL_LEVEL +#define RS_INVAL_MAX_CAUSES RS_INVAL_XID_AGE StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1), "array length mismatch"); @@ -140,6 +141,7 @@ ReplicationSlot *MyReplicationSlot = NULL; /* GUC variables */ int max_replication_slots = 10; /* the maximum number of replication * slots */ +int max_slot_xid_age = 0; /* * This GUC lists streaming replication standby server slot names that @@ -1483,6 +1485,9 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, case RS_INVAL_WAL_LEVEL: appendStringInfoString(&err_detail, _("Logical decoding on standby requires wal_level >= logical on the primary server.")); break; + case RS_INVAL_XID_AGE: + appendStringInfoString(&err_detail, _("The replication slot's xmin or catalog_xmin reached the age specified by max_slot_xid_age.")); + break; case RS_INVAL_NONE: pg_unreachable(); } @@ -1599,6 +1604,42 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, if (SlotIsLogical(s)) conflict = cause; break; + case RS_INVAL_XID_AGE: + { + TransactionId xid_cur = ReadNextTransactionId(); + TransactionId xid_limit; + TransactionId xid_slot; + + if (TransactionIdIsNormal(s->data.xmin)) + { + xid_slot = s->data.xmin; + + xid_limit = xid_slot + max_slot_xid_age; + if (xid_limit < FirstNormalTransactionId) + xid_limit += FirstNormalTransactionId; + + if (TransactionIdFollowsOrEquals(xid_cur, xid_limit)) + { + conflict = cause; + break; + } + } + if (TransactionIdIsNormal(s->data.catalog_xmin)) + { + xid_slot = s->data.catalog_xmin; + + xid_limit = xid_slot + max_slot_xid_age; + if (xid_limit < FirstNormalTransactionId) + xid_limit += FirstNormalTransactionId; + + if (TransactionIdFollowsOrEquals(xid_cur, xid_limit)) + { + conflict = cause; + break; + } + } + } + break; case RS_INVAL_NONE: pg_unreachable(); } @@ -1752,6 +1793,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, * - RS_INVAL_HORIZON: requires a snapshot <= the given horizon in the given * db; dboid may be InvalidOid for shared relations * - RS_INVAL_WAL_LEVEL: is logical + * - RS_INVAL_XID_AGE: slot's xmin or catalog_xmin has reached the age * * NB - this runs as part of checkpoint, so avoid raising errors if possible. */ diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 57d9de4dd9..6b5375909d 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -2954,6 +2954,16 @@ struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + { + {"max_slot_xid_age", PGC_SIGHUP, REPLICATION_SENDING, + gettext_noop("Age of the transaction ID at which a replication slot gets invalidated."), + gettext_noop("The transaction is the oldest transaction (including the one affecting the system catalogs) that a replication slot needs the database to retain.") + }, + &max_slot_xid_age, + 0, 0, 2000000000, + NULL, NULL, NULL + }, + { {"commit_delay", PGC_SUSET, WAL_SETTINGS, gettext_noop("Sets the delay in microseconds between transaction commit and " diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 2244ee52f7..b4c928b826 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -334,6 +334,7 @@ #wal_sender_timeout = 60s # in milliseconds; 0 disables #track_commit_timestamp = off # collect timestamp of transaction commit # (change requires restart) +#max_slot_xid_age = 0 # - Primary Server - diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 7f25a083ee..614ba0e30b 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -53,6 +53,8 @@ typedef enum ReplicationSlotInvalidationCause RS_INVAL_HORIZON, /* wal_level insufficient for slot */ RS_INVAL_WAL_LEVEL, + /* slot's xmin or catalog_xmin has reached the age */ + RS_INVAL_XID_AGE, } ReplicationSlotInvalidationCause; extern PGDLLIMPORT const char *const SlotInvalidationCauses[]; @@ -227,6 +229,7 @@ extern PGDLLIMPORT ReplicationSlot *MyReplicationSlot; /* GUCs */ extern PGDLLIMPORT int max_replication_slots; extern PGDLLIMPORT char *standby_slot_names; +extern PGDLLIMPORT int max_slot_xid_age; /* shmem initialization functions */ extern Size ReplicationSlotsShmemSize(void); diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index b1eb77b1ec..708a2a3798 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -51,6 +51,7 @@ tests += { 't/040_standby_failover_slots_sync.pl', 't/041_checkpoint_at_promote.pl', 't/042_low_level_backup.pl', + 't/050_invalidate_slots.pl', ], }, } diff --git a/src/test/recovery/t/050_invalidate_slots.pl b/src/test/recovery/t/050_invalidate_slots.pl new file mode 100644 index 0000000000..2f482b56e8 --- /dev/null +++ b/src/test/recovery/t/050_invalidate_slots.pl @@ -0,0 +1,108 @@ +# Copyright (c) 2024, PostgreSQL Global Development Group + +# Test for replication slots invalidation +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Utils; +use PostgreSQL::Test::Cluster; +use Test::More; +use Time::HiRes qw(usleep); + +# Initialize primary node, setting wal-segsize to 1MB +my $primary = PostgreSQL::Test::Cluster->new('primary'); +$primary->init(allows_streaming => 1, extra => ['--wal-segsize=1']); +$primary->append_conf( + 'postgresql.conf', q{ +checkpoint_timeout = 1h +}); +$primary->start; +$primary->safe_psql( + 'postgres', qq[ + SELECT pg_create_physical_replication_slot('sb1_slot'); +]); + +# Take backup +my $backup_name = 'my_backup'; +$primary->backup($backup_name); + +# Create a standby linking to the primary using the replication slot +my $standby1 = PostgreSQL::Test::Cluster->new('standby1'); +$standby1->init_from_backup($primary, $backup_name, has_streaming => 1); + +# Enable hs_feedback. The slot should gain an xmin. We set the status interval +# so we'll see the results promptly. +$standby1->append_conf( + 'postgresql.conf', q{ +primary_slot_name = 'sb1_slot' +hot_standby_feedback = on +wal_receiver_status_interval = 1 +}); +$standby1->start; + +# Create some content on primary to move xmin +$primary->safe_psql('postgres', + "CREATE TABLE tab_int AS SELECT generate_series(1,10) AS a"); + +# Wait until standby has replayed enough data +$primary->wait_for_catchup($standby1); + +$primary->poll_query_until( + 'postgres', qq[ + SELECT xmin IS NOT NULL + FROM pg_catalog.pg_replication_slots + WHERE slot_name = 'sb1_slot'; +]) or die "Timed out waiting for slot xmin to advance"; + +$primary->safe_psql( + 'postgres', qq[ + ALTER SYSTEM SET max_slot_xid_age = 500; +]); +$primary->reload; + +# Stop standby to make the replication slot's xmin on primary to age +$standby1->stop; + +my $logstart = -s $primary->logfile; + +# Do some work to advance xmin +$primary->safe_psql( + 'postgres', q{ +do $$ +begin + for i in 10000..11000 loop + -- use an exception block so that each iteration eats an XID + begin + insert into tab_int values (i); + exception + when division_by_zero then null; + end; + end loop; +end$$; +}); + +my $invalidated = 0; +for (my $i = 0; $i < 10 * $PostgreSQL::Test::Utils::timeout_default; $i++) +{ + $primary->safe_psql('postgres', "CHECKPOINT"); + if ($primary->log_contains( + 'invalidating obsolete replication slot "sb1_slot"', $logstart)) + { + $invalidated = 1; + last; + } + usleep(100_000); +} +ok($invalidated, 'check that slot sb1_slot invalidation has been logged'); + +# Wait for the inactive replication slots to be invalidated. +$primary->poll_query_until( + 'postgres', qq[ + SELECT COUNT(slot_name) = 1 FROM pg_replication_slots + WHERE slot_name = 'sb1_slot' AND + invalidation_reason = 'xid_aged'; +]) + or die + "Timed out while waiting for replication slot sb1_slot to be invalidated"; + +done_testing(); -- 2.34.1 [application/x-patch] v10-0003-Track-inactive-replication-slot-information.patch (10.0K, ../../CALj2ACU=Q5JG1VQFWFQe0VjPkOu2kt-6gpq7z1Ed9+JX_BR-tQ@mail.gmail.com/4-v10-0003-Track-inactive-replication-slot-information.patch) download | inline diff: From 488702c43d1b6fbb2d7dc56eb5e1409484d8b25f Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Fri, 15 Mar 2024 05:48:33 +0000 Subject: [PATCH v10 3/4] Track inactive replication slot information Up until now, postgres doesn't track metrics like the time at which the slot became inactive, and the total number of times the slot became inactive in its lifetime. This commit adds two new metrics last_inactive_at of type timestamptz and inactive_count of type numeric to ReplicationSlotPersistentData. Whenever a slot becomes inactive, the current timestamp and inactive count are persisted to disk. These metrics are useful in the following ways: - To improve replication slot monitoring tools. For instance, one can build a monitoring tool that signals a) when replication slots is lying inactive for a day or so using last_inactive_at metric, b) when a replication slot is becoming inactive too frequently using last_inactive_at metric. - To implement timeout-based inactive replication slot management capability in postgres. Increases SLOT_VERSION due to the added two new metrics. --- doc/src/sgml/system-views.sgml | 20 +++++++++++++ src/backend/catalog/system_views.sql | 4 ++- src/backend/replication/slot.c | 43 ++++++++++++++++++++++------ src/backend/replication/slotfuncs.c | 15 +++++++++- src/include/catalog/pg_proc.dat | 6 ++-- src/include/replication/slot.h | 6 ++++ src/test/regress/expected/rules.out | 6 ++-- 7 files changed, 84 insertions(+), 16 deletions(-) diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml index f3fb5ba1b0..59cd1b5211 100644 --- a/doc/src/sgml/system-views.sgml +++ b/doc/src/sgml/system-views.sgml @@ -2750,6 +2750,26 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx ID of role </para></entry> </row> + + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>last_inactive_at</structfield> <type>timestamptz</type> + </para> + <para> + The time at which the slot became inactive. + <literal>NULL</literal> if the slot is currently actively being + used. + </para></entry> + </row> + + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>inactive_count</structfield> <type>numeric</type> + </para> + <para> + The total number of times the slot became inactive in its lifetime. + </para></entry> + </row> </tbody> </tgroup> </table> diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index cd22dad959..de9f1d5506 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1026,7 +1026,9 @@ CREATE VIEW pg_replication_slots AS L.conflicting, L.failover, L.synced, - L.invalidation_reason + L.invalidation_reason, + L.last_inactive_at, + L.inactive_count FROM pg_get_replication_slots() AS L LEFT JOIN pg_database D ON (L.datoid = D.oid); diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 2a1885da24..e606218673 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -130,7 +130,7 @@ StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1), sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize #define SLOT_MAGIC 0x1051CA1 /* format identifier */ -#define SLOT_VERSION 5 /* version for new files */ +#define SLOT_VERSION 6 /* version for new files */ /* Control array for replication slot management */ ReplicationSlotCtlData *ReplicationSlotCtl = NULL; @@ -400,6 +400,8 @@ ReplicationSlotCreate(const char *name, bool db_specific, slot->data.two_phase_at = InvalidXLogRecPtr; slot->data.failover = failover; slot->data.synced = synced; + slot->data.last_inactive_at = 0; + slot->data.inactive_count = 0; /* and then data only present in shared memory */ slot->just_dirtied = false; @@ -626,6 +628,17 @@ retry: if (am_walsender) { + if (s->data.persistency == RS_PERSISTENT) + { + SpinLockAcquire(&s->mutex); + s->data.last_inactive_at = 0; + SpinLockRelease(&s->mutex); + + /* Write this slot to disk */ + ReplicationSlotMarkDirty(); + ReplicationSlotSave(); + } + ereport(log_replication_commands ? LOG : DEBUG1, SlotIsLogical(s) ? errmsg("acquired logical replication slot \"%s\"", @@ -693,16 +706,20 @@ ReplicationSlotRelease(void) ConditionVariableBroadcast(&slot->active_cv); } - MyReplicationSlot = NULL; - - /* might not have been set when we've been a plain slot */ - LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE); - MyProc->statusFlags &= ~PROC_IN_LOGICAL_DECODING; - ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags; - LWLockRelease(ProcArrayLock); - if (am_walsender) { + if (slot->data.persistency == RS_PERSISTENT) + { + SpinLockAcquire(&slot->mutex); + slot->data.last_inactive_at = GetCurrentTimestamp(); + slot->data.inactive_count++; + SpinLockRelease(&slot->mutex); + + /* Write this slot to disk */ + ReplicationSlotMarkDirty(); + ReplicationSlotSave(); + } + ereport(log_replication_commands ? LOG : DEBUG1, is_logical ? errmsg("released logical replication slot \"%s\"", @@ -712,6 +729,14 @@ ReplicationSlotRelease(void) pfree(slotname); } + + MyReplicationSlot = NULL; + + /* might not have been set when we've been a plain slot */ + LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE); + MyProc->statusFlags &= ~PROC_IN_LOGICAL_DECODING; + ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags; + LWLockRelease(ProcArrayLock); } /* diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c index b5a638edea..4c7a120df1 100644 --- a/src/backend/replication/slotfuncs.c +++ b/src/backend/replication/slotfuncs.c @@ -239,7 +239,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS) Datum pg_get_replication_slots(PG_FUNCTION_ARGS) { -#define PG_GET_REPLICATION_SLOTS_COLS 18 +#define PG_GET_REPLICATION_SLOTS_COLS 20 ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; XLogRecPtr currlsn; int slotno; @@ -264,6 +264,7 @@ pg_get_replication_slots(PG_FUNCTION_ARGS) WALAvailability walstate; int i; ReplicationSlotInvalidationCause cause; + char buf[256]; if (!slot->in_use) continue; @@ -436,6 +437,18 @@ pg_get_replication_slots(PG_FUNCTION_ARGS) else values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]); + if (slot_contents.data.last_inactive_at > 0) + values[i++] = TimestampTzGetDatum(slot_contents.data.last_inactive_at); + else + nulls[i++] = true; + + /* Convert to numeric. */ + snprintf(buf, sizeof buf, UINT64_FORMAT, slot_contents.data.inactive_count); + values[i++] = DirectFunctionCall3(numeric_in, + CStringGetDatum(buf), + ObjectIdGetDatum(0), + Int32GetDatum(-1)); + Assert(i == PG_GET_REPLICATION_SLOTS_COLS); tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 63fd0b4cd7..c7ab0893eb 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -11123,9 +11123,9 @@ proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f', proretset => 't', provolatile => 's', prorettype => 'record', proargtypes => '', - proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,bool,bool,text}', - proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}', - proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,failover,synced,invalidation_reason}', + proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,bool,bool,text,timestamptz,numeric}', + proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}', + proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,failover,synced,invalidation_reason,last_inactive_at,inactive_count}', prosrc => 'pg_get_replication_slots' }, { oid => '3786', descr => 'set up a logical replication slot', proname => 'pg_create_logical_replication_slot', provolatile => 'v', diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 614ba0e30b..780767a819 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -129,6 +129,12 @@ typedef struct ReplicationSlotPersistentData * for logical slots on the primary server. */ bool failover; + + /* When did this slot become inactive last time? */ + TimestampTz last_inactive_at; + + /* How many times the slot has been inactive? */ + uint64 inactive_count; } ReplicationSlotPersistentData; /* diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 055bec068d..c0bdfe76d8 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1476,8 +1476,10 @@ pg_replication_slots| SELECT l.slot_name, l.conflicting, l.failover, l.synced, - l.invalidation_reason - FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, failover, synced, invalidation_reason) + l.invalidation_reason, + l.last_inactive_at, + l.inactive_count + FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, failover, synced, invalidation_reason, last_inactive_at, inactive_count) LEFT JOIN pg_database d ON ((l.datoid = d.oid))); pg_roles| SELECT pg_authid.rolname, pg_authid.rolsuper, -- 2.34.1 [application/x-patch] v10-0004-Add-inactive_timeout-based-replication-slot-inva.patch (11.3K, ../../CALj2ACU=Q5JG1VQFWFQe0VjPkOu2kt-6gpq7z1Ed9+JX_BR-tQ@mail.gmail.com/5-v10-0004-Add-inactive_timeout-based-replication-slot-inva.patch) download | inline diff: From 559c02f760d998b0b78f68fbbd60fa5d3ec960d9 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Fri, 15 Mar 2024 05:49:02 +0000 Subject: [PATCH v10 4/4] Add inactive_timeout based replication slot invalidation Up until now, postgres has the ability to invalidate inactive replication slots based on the amount of WAL (set via max_slot_wal_keep_size GUC) that will be needed for the slots in case they become active. However, choosing a default value for max_slot_wal_keep_size is tricky. Because the amount of WAL a customer generates, and their allocated storage will vary greatly in production, making it difficult to pin down a one-size-fits-all value. It is often easy for developers to set a timeout of say 1 or 2 or 3 days, after which the inactive slots get dropped. To achieve the above, postgres uses replication slot metric inactive_at (the time at which the slot became inactive), and a new GUC inactive_replication_slot_timeout. The checkpointer then looks at all replication slots invalidating the inactive slots based on the timeout set. --- doc/src/sgml/config.sgml | 18 +++++ src/backend/access/transam/xlog.c | 10 +++ src/backend/replication/slot.c | 22 +++++- src/backend/utils/misc/guc_tables.c | 12 +++ src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/replication/slot.h | 3 + src/test/recovery/t/050_invalidate_slots.pl | 79 +++++++++++++++++++ 7 files changed, 144 insertions(+), 1 deletion(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 6dd54ffcb7..4b0b60a1ac 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -4565,6 +4565,24 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"' # Windows </listitem> </varlistentry> + <varlistentry id="guc-inactive-replication-slot-timeout" xreflabel="inactive_replication_slot_timeout"> + <term><varname>inactive_replication_slot_timeout</varname> (<type>integer</type>) + <indexterm> + <primary><varname>inactive_replication_slot_timeout</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + Invalidate replication slots that are inactive for longer than this + amount of time at the next checkpoint. If this value is specified + without units, it is taken as seconds. A value of zero (which is + default) disables the timeout mechanism. This parameter can only be + set in the <filename>postgresql.conf</filename> file or on the server + command line. + </para> + </listitem> + </varlistentry> + <varlistentry id="guc-track-commit-timestamp" xreflabel="track_commit_timestamp"> <term><varname>track_commit_timestamp</varname> (<type>boolean</type>) <indexterm> diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 36ae2ac6a4..166c3ed794 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -7152,6 +7152,11 @@ CreateCheckPoint(int flags) InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0, InvalidOid, InvalidTransactionId); + /* Invalidate inactive replication slots based on timeout */ + if (inactive_replication_slot_timeout > 0) + InvalidateObsoleteReplicationSlots(RS_INVAL_INACTIVE_TIMEOUT, 0, + InvalidOid, InvalidTransactionId); + /* * Delete old log files, those no longer needed for last checkpoint to * prevent the disk holding the xlog from growing full. @@ -7607,6 +7612,11 @@ CreateRestartPoint(int flags) InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0, InvalidOid, InvalidTransactionId); + /* Invalidate inactive replication slots based on timeout */ + if (inactive_replication_slot_timeout > 0) + InvalidateObsoleteReplicationSlots(RS_INVAL_INACTIVE_TIMEOUT, 0, + InvalidOid, InvalidTransactionId); + /* * Retreat _logSegNo using the current end of xlog replayed or received, * whichever is later. diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index e606218673..37498d3d98 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -108,10 +108,11 @@ const char *const SlotInvalidationCauses[] = { [RS_INVAL_HORIZON] = "rows_removed", [RS_INVAL_WAL_LEVEL] = "wal_level_insufficient", [RS_INVAL_XID_AGE] = "xid_aged", + [RS_INVAL_INACTIVE_TIMEOUT] = "inactive_timeout", }; /* Maximum number of invalidation causes */ -#define RS_INVAL_MAX_CAUSES RS_INVAL_XID_AGE +#define RS_INVAL_MAX_CAUSES RS_INVAL_INACTIVE_TIMEOUT StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1), "array length mismatch"); @@ -142,6 +143,7 @@ ReplicationSlot *MyReplicationSlot = NULL; int max_replication_slots = 10; /* the maximum number of replication * slots */ int max_slot_xid_age = 0; +int inactive_replication_slot_timeout = 0; /* * This GUC lists streaming replication standby server slot names that @@ -1513,6 +1515,9 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, case RS_INVAL_XID_AGE: appendStringInfoString(&err_detail, _("The replication slot's xmin or catalog_xmin reached the age specified by max_slot_xid_age.")); break; + case RS_INVAL_INACTIVE_TIMEOUT: + appendStringInfoString(&err_detail, _("The slot has been inactive for more than the time specified by inactive_replication_slot_timeout.")); + break; case RS_INVAL_NONE: pg_unreachable(); } @@ -1665,6 +1670,20 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, } } break; + case RS_INVAL_INACTIVE_TIMEOUT: + if (s->data.last_inactive_at > 0) + { + TimestampTz now; + + Assert(s->data.persistency == RS_PERSISTENT); + Assert(s->active_pid == 0); + + now = GetCurrentTimestamp(); + if (TimestampDifferenceExceeds(s->data.last_inactive_at, now, + inactive_replication_slot_timeout * 1000)) + conflict = cause; + } + break; case RS_INVAL_NONE: pg_unreachable(); } @@ -1819,6 +1838,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, * db; dboid may be InvalidOid for shared relations * - RS_INVAL_WAL_LEVEL: is logical * - RS_INVAL_XID_AGE: slot's xmin or catalog_xmin has reached the age + * - RS_INVAL_INACTIVE_TIMEOUT: inactive slot timeout occurs * * NB - this runs as part of checkpoint, so avoid raising errors if possible. */ diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 6b5375909d..6caf40d51e 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -2964,6 +2964,18 @@ struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + { + {"inactive_replication_slot_timeout", PGC_SIGHUP, REPLICATION_SENDING, + gettext_noop("Sets the amount of time to wait before invalidating an " + "inactive replication slot."), + NULL, + GUC_UNIT_S + }, + &inactive_replication_slot_timeout, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, + { {"commit_delay", PGC_SUSET, WAL_SETTINGS, gettext_noop("Sets the delay in microseconds between transaction commit and " diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index b4c928b826..7f2a3e41f1 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -261,6 +261,7 @@ #recovery_prefetch = try # prefetch pages referenced in the WAL? #wal_decode_buffer_size = 512kB # lookahead window used for prefetching # (change requires restart) +#inactive_replication_slot_timeout = 0 # in seconds; 0 disables # - Archiving - diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 780767a819..8378d7c913 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -55,6 +55,8 @@ typedef enum ReplicationSlotInvalidationCause RS_INVAL_WAL_LEVEL, /* slot's xmin or catalog_xmin has reached the age */ RS_INVAL_XID_AGE, + /* inactive slot timeout has occurred */ + RS_INVAL_INACTIVE_TIMEOUT, } ReplicationSlotInvalidationCause; extern PGDLLIMPORT const char *const SlotInvalidationCauses[]; @@ -236,6 +238,7 @@ extern PGDLLIMPORT ReplicationSlot *MyReplicationSlot; extern PGDLLIMPORT int max_replication_slots; extern PGDLLIMPORT char *standby_slot_names; extern PGDLLIMPORT int max_slot_xid_age; +extern PGDLLIMPORT int inactive_replication_slot_timeout; /* shmem initialization functions */ extern Size ReplicationSlotsShmemSize(void); diff --git a/src/test/recovery/t/050_invalidate_slots.pl b/src/test/recovery/t/050_invalidate_slots.pl index 2f482b56e8..4c66dd4a4e 100644 --- a/src/test/recovery/t/050_invalidate_slots.pl +++ b/src/test/recovery/t/050_invalidate_slots.pl @@ -105,4 +105,83 @@ $primary->poll_query_until( or die "Timed out while waiting for replication slot sb1_slot to be invalidated"; +$primary->safe_psql( + 'postgres', qq[ + SELECT pg_create_physical_replication_slot('sb2_slot'); +]); + +$primary->safe_psql( + 'postgres', qq[ + ALTER SYSTEM SET max_slot_xid_age = 0; +]); +$primary->reload; + +# Create a standby linking to the primary using the replication slot +my $standby2 = PostgreSQL::Test::Cluster->new('standby2'); +$standby2->init_from_backup($primary, $backup_name, has_streaming => 1); +$standby2->append_conf( + 'postgresql.conf', q{ +primary_slot_name = 'sb2_slot' +}); +$standby2->start; + +# Wait until standby has replayed enough data +$primary->wait_for_catchup($standby2); + +# The inactive replication slot info should be null when the slot is active +my $result = $primary->safe_psql( + 'postgres', qq[ + SELECT last_inactive_at IS NULL, inactive_count = 0 AS OK + FROM pg_replication_slots WHERE slot_name = 'sb2_slot'; +]); +is($result, "t|t", + 'check the inactive replication slot info for an active slot'); + +# Set timeout so that the next checkpoint will invalidate the inactive +# replication slot. +$primary->safe_psql( + 'postgres', qq[ + ALTER SYSTEM SET inactive_replication_slot_timeout TO '1s'; +]); +$primary->reload; + +$logstart = -s $primary->logfile; + +# Stop standby to make the replication slot on primary inactive +$standby2->stop; + +# Wait for the inactive replication slot info to be updated +$primary->poll_query_until( + 'postgres', qq[ + SELECT COUNT(slot_name) = 1 FROM pg_replication_slots + WHERE last_inactive_at IS NOT NULL AND + inactive_count = 1 AND slot_name = 'sb2_slot'; +]) + or die + "Timed out while waiting for inactive replication slot info to be updated"; + +$invalidated = 0; +for (my $i = 0; $i < 10 * $PostgreSQL::Test::Utils::timeout_default; $i++) +{ + $primary->safe_psql('postgres', "CHECKPOINT"); + if ($primary->log_contains( + 'invalidating obsolete replication slot "sb2_slot"', $logstart)) + { + $invalidated = 1; + last; + } + usleep(100_000); +} +ok($invalidated, 'check that slot sb2_slot invalidation has been logged'); + +# Wait for the inactive replication slots to be invalidated. +$primary->poll_query_until( + 'postgres', qq[ + SELECT COUNT(slot_name) = 1 FROM pg_replication_slots + WHERE slot_name = 'sb2_slot' AND + invalidation_reason = 'inactive_timeout'; +]) + or die + "Timed out while waiting for inactive replication slot sb2_slot to be invalidated"; + done_testing(); -- 2.34.1 ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-15 14:28 Nathan Bossart <[email protected]> parent: Amit Kapila <[email protected]> 2 siblings, 0 replies; 18+ messages in thread From: Nathan Bossart @ 2024-03-15 14:28 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Bharath Rupireddy <[email protected]>; Bertrand Drouvot <[email protected]>; PostgreSQL Hackers <[email protected]> On Thu, Mar 14, 2024 at 12:24:00PM +0530, Amit Kapila wrote: > On Wed, Mar 13, 2024 at 9:24 PM Bharath Rupireddy > <[email protected]> wrote: >> On Wed, Mar 13, 2024 at 9:21 AM Amit Kapila <[email protected]> wrote: >> > > So, how about we turn conflict_reason to only report the reasons that >> > > actually cause conflict with recovery for logical slots, something >> > > like below, and then have invalidation_cause as a generic column for >> > > all sorts of invalidation reasons for both logical and physical slots? >> > >> > If our above understanding is correct then coflict_reason will be a >> > subset of invalidation_reason. If so, whatever way we arrange this >> > information, there will be some sort of duplicity unless we just have >> > one column 'invalidation_reason' and update the docs to interpret it >> > correctly for conflicts. >> >> Yes, there will be some sort of duplicity if we emit conflict_reason >> as a text field. However, I still think the better way is to turn >> conflict_reason text to conflict boolean and set it to true only on >> rows_removed and wal_level_insufficient invalidations. When conflict >> boolean is true, one (including all the tests that we've added >> recently) can look for invalidation_reason text field for the reason. >> This sounds reasonable to me as opposed to we just mentioning in the >> docs that "if invalidation_reason is rows_removed or >> wal_level_insufficient it's the reason for conflict with recovery". > > Fair point. I think we can go either way. Bertrand, Nathan, and > others, do you have an opinion on this matter? WFM -- Nathan Bossart Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-15 16:45 Bertrand Drouvot <[email protected]> parent: Amit Kapila <[email protected]> 2 siblings, 0 replies; 18+ messages in thread From: Bertrand Drouvot @ 2024-03-15 16:45 UTC (permalink / raw) To: Amit Kapila <[email protected]>; +Cc: Bharath Rupireddy <[email protected]>; Nathan Bossart <[email protected]>; PostgreSQL Hackers <[email protected]> Hi, On Thu, Mar 14, 2024 at 12:24:00PM +0530, Amit Kapila wrote: > On Wed, Mar 13, 2024 at 9:24 PM Bharath Rupireddy > <[email protected]> wrote: > > > > On Wed, Mar 13, 2024 at 9:21 AM Amit Kapila <[email protected]> wrote: > > > > > > > So, how about we turn conflict_reason to only report the reasons that > > > > actually cause conflict with recovery for logical slots, something > > > > like below, and then have invalidation_cause as a generic column for > > > > all sorts of invalidation reasons for both logical and physical slots? > > > > > > If our above understanding is correct then coflict_reason will be a > > > subset of invalidation_reason. If so, whatever way we arrange this > > > information, there will be some sort of duplicity unless we just have > > > one column 'invalidation_reason' and update the docs to interpret it > > > correctly for conflicts. > > > > Yes, there will be some sort of duplicity if we emit conflict_reason > > as a text field. However, I still think the better way is to turn > > conflict_reason text to conflict boolean and set it to true only on > > rows_removed and wal_level_insufficient invalidations. When conflict > > boolean is true, one (including all the tests that we've added > > recently) can look for invalidation_reason text field for the reason. > > This sounds reasonable to me as opposed to we just mentioning in the > > docs that "if invalidation_reason is rows_removed or > > wal_level_insufficient it's the reason for conflict with recovery". > > > > Fair point. I think we can go either way. Bertrand, Nathan, and > others, do you have an opinion on this matter? Sounds like a good approach to me and one will be able to quickly identify if a conflict occured. Regards, -- Bertrand Drouvot PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-16 03:59 Bharath Rupireddy <[email protected]> parent: shveta malik <[email protected]> 0 siblings, 2 replies; 18+ messages in thread From: Bharath Rupireddy @ 2024-03-16 03:59 UTC (permalink / raw) To: shveta malik <[email protected]>; +Cc: Amit Kapila <[email protected]>; Nathan Bossart <[email protected]>; Drouvot, Bertrand <[email protected]>; PostgreSQL Hackers <[email protected]> On Fri, Mar 15, 2024 at 12:49 PM shveta malik <[email protected]> wrote: > > patch002: > > 1) > I would like to understand the purpose of 'inactive_count'? Is it only > for users for monitoring purposes? We are not using it anywhere > internally. inactive_count metric helps detect unstable replication slots connections that have a lot of disconnections. It's not used for the inactive_timeout based slot invalidation mechanism. > I shutdown the instance 5 times and found that 'inactive_count' became > 5 for all the slots created on that instance. Is this intentional? Yes, it's incremented on shutdown (and for that matter upon every slot release) for all the slots that are tied to walsenders. > I mean we can not really use them if the instance is down. I felt it > should increment the inactive_count only if during the span of > instance, they were actually inactive i.e. no streaming or replication > happening through them. inactive_count is persisted to disk- upon clean shutdown, so, once the slots become active again, one gets to see the metric and deduce some info on disconnections. Having said that, I'm okay to hear from others on the inactive_count metric being added. > 2) > slot.c: > + case RS_INVAL_XID_AGE: > > Can we optimize this code? It has duplicate code for processing > s->data.catalog_xmin and s->data.xmin. Can we create a sub-function > for this purpose and call it twice here? Good idea. Done that way. > 2) > The msg for patch 3 says: > -------------- > a) when replication slots is lying inactive for a day or so using > last_inactive_at metric, > b) when a replication slot is becoming inactive too frequently using > last_inactive_at metric. > -------------- > I think in b, you want to refer to inactive_count instead of last_inactive_at? Right. Changed. > 3) > I do not see invalidation_reason updated for 2 new reasons in system-views.sgml Nice catch. Added them now. I've also responded to Bertrand's comments here. On Wed, Mar 6, 2024 at 3:56 PM Bertrand Drouvot <[email protected]> wrote: > > A few comments: > > 1 === > > + The reason for the slot's invalidation. <literal>NULL</literal> if the > + slot is currently actively being used. > > s/currently actively being used/not invalidated/ ? (I mean it could be valid > and not being used). Changed. > 3 === > > res = executeQueryOrDie(conn, "SELECT slot_name, plugin, two_phase, failover, " > - "%s as caught_up, conflict_reason IS NOT NULL as invalid " > + "%s as caught_up, invalidation_reason IS NOT NULL as invalid " > "FROM pg_catalog.pg_replication_slots " > - "(CASE WHEN conflict_reason IS NOT NULL THEN FALSE " > + "(CASE WHEN invalidation_reason IS NOT NULL THEN FALSE " > > Yeah that's fine because there is logical slot filtering here. Right. And, we really are looking for invalid slots there, so use of invalidation_reason is much more correct than conflicting. > 4 === > > -GetSlotInvalidationCause(const char *conflict_reason) > +GetSlotInvalidationCause(const char *invalidation_reason) > > Should we change the comment "Maps a conflict reason" above this function? Changed. > 5 === > > -# Check conflict_reason is NULL for physical slot > +# Check invalidation_reason is NULL for physical slot > $res = $node_primary->safe_psql( > 'postgres', qq[ > - SELECT conflict_reason is null FROM pg_replication_slots where slot_name = '$primary_slotname';] > + SELECT invalidation_reason is null FROM pg_replication_slots where slot_name = '$primary_slotname';] > ); > > > I don't think this test is needed anymore: it does not make that much sense since > it's done after the primary database initialization and startup. It is now turned into a test verifying 'conflicting boolean' is null for the physical slot. Isn't that okay? > 6 === > > 'Logical slots are reported as non conflicting'); > > What about? > > " > # Verify slots are reported as valid in pg_replication_slots > 'Logical slots are reported as valid'); > " Changed. Please see the attached v11 patch set with all the above review comments addressed. -- Bharath Rupireddy PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com Attachments: [application/octet-stream] v11-0001-Track-invalidation_reason-in-pg_replication_slot.patch (20.2K, ../../CALj2ACWBbSa7ccOGDHNfH1OX7_ArBJm1kNmc=iWKSfqq1NHnuA@mail.gmail.com/2-v11-0001-Track-invalidation_reason-in-pg_replication_slot.patch) download | inline diff: From 483824a8b3248fe08b6bdf22c68bada4f0549212 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Sat, 16 Mar 2024 03:39:35 +0000 Subject: [PATCH v11 1/4] Track invalidation_reason in pg_replication_slots Up until now, reason for replication slot invalidation is not tracked in pg_replication_slots. A recent commit 007693f2a added conflict_reason to show the reasons for slot invalidation, but only for logical slots. This commit adds a new column to show invalidation reasons for both physical and logical slots. And, this commit also turns conflict_reason text column to conflicting boolean column (effectively reverting commit 007693f2a). One now can look at the new invalidation_reason column for logical slots conflict with recovery. --- doc/src/sgml/ref/pgupgrade.sgml | 4 +- doc/src/sgml/system-views.sgml | 63 +++++++++++-------- src/backend/catalog/system_views.sql | 5 +- src/backend/replication/logical/slotsync.c | 2 +- src/backend/replication/slot.c | 8 +-- src/backend/replication/slotfuncs.c | 25 +++++--- src/bin/pg_upgrade/info.c | 4 +- src/include/catalog/pg_proc.dat | 6 +- src/include/replication/slot.h | 2 +- .../t/035_standby_logical_decoding.pl | 39 ++++++------ .../t/040_standby_failover_slots_sync.pl | 4 +- src/test/regress/expected/rules.out | 7 ++- 12 files changed, 97 insertions(+), 72 deletions(-) diff --git a/doc/src/sgml/ref/pgupgrade.sgml b/doc/src/sgml/ref/pgupgrade.sgml index 58c6c2df8b..8de52bf752 100644 --- a/doc/src/sgml/ref/pgupgrade.sgml +++ b/doc/src/sgml/ref/pgupgrade.sgml @@ -453,8 +453,8 @@ make prefix=/usr/local/pgsql.new install <para> All slots on the old cluster must be usable, i.e., there are no slots whose - <link linkend="view-pg-replication-slots">pg_replication_slots</link>.<structfield>conflict_reason</structfield> - is not <literal>NULL</literal>. + <link linkend="view-pg-replication-slots">pg_replication_slots</link>.<structfield>conflicting</structfield> + is not <literal>true</literal>. </para> </listitem> <listitem> diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml index be90edd0e2..e685921847 100644 --- a/doc/src/sgml/system-views.sgml +++ b/doc/src/sgml/system-views.sgml @@ -2525,34 +2525,13 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx <row> <entry role="catalog_table_entry"><para role="column_definition"> - <structfield>conflict_reason</structfield> <type>text</type> + <structfield>conflicting</structfield> <type>bool</type> </para> <para> - The reason for the logical slot's conflict with recovery. It is always - NULL for physical slots, as well as for logical slots which are not - invalidated. The non-NULL values indicate that the slot is marked - as invalidated. Possible values are: - <itemizedlist spacing="compact"> - <listitem> - <para> - <literal>wal_removed</literal> means that the required WAL has been - removed. - </para> - </listitem> - <listitem> - <para> - <literal>rows_removed</literal> means that the required rows have - been removed. - </para> - </listitem> - <listitem> - <para> - <literal>wal_level_insufficient</literal> means that the - primary doesn't have a <xref linkend="guc-wal-level"/> sufficient to - perform logical decoding. - </para> - </listitem> - </itemizedlist> + True if this logical slot conflicted with recovery (and so is now + invalidated). When this column is true, check + <structfield>invalidation_reason</structfield> column for the conflict + reason. </para></entry> </row> @@ -2581,6 +2560,38 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx </para></entry> </row> + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>invalidation_reason</structfield> <type>text</type> + </para> + <para> + The reason for the slot's invalidation. It is set for both logical and + physical slots. <literal>NULL</literal> if the slot is not invalidated. + Possible values are: + <itemizedlist spacing="compact"> + <listitem> + <para> + <literal>wal_removed</literal> means that the required WAL has been + removed. + </para> + </listitem> + <listitem> + <para> + <literal>rows_removed</literal> means that the required rows have + been removed. + </para> + </listitem> + <listitem> + <para> + <literal>wal_level_insufficient</literal> means that the + primary doesn't have a <xref linkend="guc-wal-level"/> sufficient to + perform logical decoding. + </para> + </listitem> + </itemizedlist> + </para></entry> + </row> + </tbody> </tgroup> </table> diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index 04227a72d1..cd22dad959 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1023,9 +1023,10 @@ CREATE VIEW pg_replication_slots AS L.wal_status, L.safe_wal_size, L.two_phase, - L.conflict_reason, + L.conflicting, L.failover, - L.synced + L.synced, + L.invalidation_reason FROM pg_get_replication_slots() AS L LEFT JOIN pg_database D ON (L.datoid = D.oid); diff --git a/src/backend/replication/logical/slotsync.c b/src/backend/replication/logical/slotsync.c index 5074c8409f..260632cfdd 100644 --- a/src/backend/replication/logical/slotsync.c +++ b/src/backend/replication/logical/slotsync.c @@ -668,7 +668,7 @@ synchronize_slots(WalReceiverConn *wrconn) bool started_tx = false; const char *query = "SELECT slot_name, plugin, confirmed_flush_lsn," " restart_lsn, catalog_xmin, two_phase, failover," - " database, conflict_reason" + " database, invalidation_reason" " FROM pg_catalog.pg_replication_slots" " WHERE failover and NOT temporary"; diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 91ca397857..4f1a17f6ce 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -2356,21 +2356,21 @@ RestoreSlotFromDisk(const char *name) } /* - * Maps a conflict reason for a replication slot to + * Maps a invalidation reason for a replication slot to * ReplicationSlotInvalidationCause. */ ReplicationSlotInvalidationCause -GetSlotInvalidationCause(const char *conflict_reason) +GetSlotInvalidationCause(const char *invalidation_reason) { ReplicationSlotInvalidationCause cause; ReplicationSlotInvalidationCause result = RS_INVAL_NONE; bool found PG_USED_FOR_ASSERTS_ONLY = false; - Assert(conflict_reason); + Assert(invalidation_reason); for (cause = RS_INVAL_NONE; cause <= RS_INVAL_MAX_CAUSES; cause++) { - if (strcmp(SlotInvalidationCauses[cause], conflict_reason) == 0) + if (strcmp(SlotInvalidationCauses[cause], invalidation_reason) == 0) { found = true; result = cause; diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c index ad79e1fccd..b5a638edea 100644 --- a/src/backend/replication/slotfuncs.c +++ b/src/backend/replication/slotfuncs.c @@ -239,7 +239,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS) Datum pg_get_replication_slots(PG_FUNCTION_ARGS) { -#define PG_GET_REPLICATION_SLOTS_COLS 17 +#define PG_GET_REPLICATION_SLOTS_COLS 18 ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; XLogRecPtr currlsn; int slotno; @@ -263,6 +263,7 @@ pg_get_replication_slots(PG_FUNCTION_ARGS) bool nulls[PG_GET_REPLICATION_SLOTS_COLS]; WALAvailability walstate; int i; + ReplicationSlotInvalidationCause cause; if (!slot->in_use) continue; @@ -409,22 +410,32 @@ pg_get_replication_slots(PG_FUNCTION_ARGS) values[i++] = BoolGetDatum(slot_contents.data.two_phase); - if (slot_contents.data.database == InvalidOid) + cause = slot_contents.data.invalidated; + + if (SlotIsPhysical(&slot_contents)) nulls[i++] = true; else { - ReplicationSlotInvalidationCause cause = slot_contents.data.invalidated; - - if (cause == RS_INVAL_NONE) - nulls[i++] = true; + /* + * rows_removed and wal_level_insufficient are only two reasons + * for the logical slot's conflict with recovery. + */ + if (cause == RS_INVAL_HORIZON || + cause == RS_INVAL_WAL_LEVEL) + values[i++] = BoolGetDatum(true); else - values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]); + values[i++] = BoolGetDatum(false); } values[i++] = BoolGetDatum(slot_contents.data.failover); values[i++] = BoolGetDatum(slot_contents.data.synced); + if (cause == RS_INVAL_NONE) + nulls[i++] = true; + else + values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]); + Assert(i == PG_GET_REPLICATION_SLOTS_COLS); tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, diff --git a/src/bin/pg_upgrade/info.c b/src/bin/pg_upgrade/info.c index b5b8d11602..34a157f792 100644 --- a/src/bin/pg_upgrade/info.c +++ b/src/bin/pg_upgrade/info.c @@ -676,13 +676,13 @@ get_old_cluster_logical_slot_infos(DbInfo *dbinfo, bool live_check) * removed. */ res = executeQueryOrDie(conn, "SELECT slot_name, plugin, two_phase, failover, " - "%s as caught_up, conflict_reason IS NOT NULL as invalid " + "%s as caught_up, invalidation_reason IS NOT NULL as invalid " "FROM pg_catalog.pg_replication_slots " "WHERE slot_type = 'logical' AND " "database = current_database() AND " "temporary IS FALSE;", live_check ? "FALSE" : - "(CASE WHEN conflict_reason IS NOT NULL THEN FALSE " + "(CASE WHEN conflicting THEN FALSE " "ELSE (SELECT pg_catalog.binary_upgrade_logical_slot_has_caught_up(slot_name)) " "END)"); diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 700f7daf7b..63fd0b4cd7 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -11123,9 +11123,9 @@ proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f', proretset => 't', provolatile => 's', prorettype => 'record', proargtypes => '', - proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,text,bool,bool}', - proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}', - proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflict_reason,failover,synced}', + proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,bool,bool,text}', + proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}', + proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,failover,synced,invalidation_reason}', prosrc => 'pg_get_replication_slots' }, { oid => '3786', descr => 'set up a logical replication slot', proname => 'pg_create_logical_replication_slot', provolatile => 'v', diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 425effad21..7f25a083ee 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -273,7 +273,7 @@ extern void CheckPointReplicationSlots(bool is_shutdown); extern void CheckSlotRequirements(void); extern void CheckSlotPermissions(void); extern ReplicationSlotInvalidationCause - GetSlotInvalidationCause(const char *conflict_reason); + GetSlotInvalidationCause(const char *invalidation_reason); extern bool SlotExistsInStandbySlotNames(const char *slot_name); extern bool StandbySlotsHaveCaughtup(XLogRecPtr wait_for_lsn, int elevel); diff --git a/src/test/recovery/t/035_standby_logical_decoding.pl b/src/test/recovery/t/035_standby_logical_decoding.pl index 88b03048c4..2203841ca1 100644 --- a/src/test/recovery/t/035_standby_logical_decoding.pl +++ b/src/test/recovery/t/035_standby_logical_decoding.pl @@ -168,7 +168,7 @@ sub change_hot_standby_feedback_and_wait_for_xmins } } -# Check conflict_reason in pg_replication_slots. +# Check reason for conflict in pg_replication_slots. sub check_slots_conflict_reason { my ($slot_prefix, $reason) = @_; @@ -178,15 +178,15 @@ sub check_slots_conflict_reason $res = $node_standby->safe_psql( 'postgres', qq( - select conflict_reason from pg_replication_slots where slot_name = '$active_slot';)); + select invalidation_reason from pg_replication_slots where slot_name = '$active_slot' and conflicting;)); - is($res, "$reason", "$active_slot conflict_reason is $reason"); + is($res, "$reason", "$active_slot reason for conflict is $reason"); $res = $node_standby->safe_psql( 'postgres', qq( - select conflict_reason from pg_replication_slots where slot_name = '$inactive_slot';)); + select invalidation_reason from pg_replication_slots where slot_name = '$inactive_slot' and conflicting;)); - is($res, "$reason", "$inactive_slot conflict_reason is $reason"); + is($res, "$reason", "$inactive_slot reason for conflict is $reason"); } # Drop the slots, re-create them, change hot_standby_feedback, @@ -293,13 +293,13 @@ $node_primary->safe_psql('testdb', qq[SELECT * FROM pg_create_physical_replication_slot('$primary_slotname');] ); -# Check conflict_reason is NULL for physical slot +# Check conflicting is NULL for physical slot $res = $node_primary->safe_psql( 'postgres', qq[ - SELECT conflict_reason is null FROM pg_replication_slots where slot_name = '$primary_slotname';] + SELECT conflicting is null FROM pg_replication_slots where slot_name = '$primary_slotname';] ); -is($res, 't', "Physical slot reports conflict_reason as NULL"); +is($res, 't', "Physical slot reports conflicting as NULL"); my $backup_name = 'b1'; $node_primary->backup($backup_name); @@ -524,7 +524,7 @@ $node_primary->wait_for_replay_catchup($node_standby); # Check invalidation in the logfile and in pg_stat_database_conflicts check_for_invalidation('vacuum_full_', 1, 'with vacuum FULL on pg_class'); -# Verify conflict_reason is 'rows_removed' in pg_replication_slots +# Verify reason for conflict is 'rows_removed' in pg_replication_slots check_slots_conflict_reason('vacuum_full_', 'rows_removed'); # Ensure that replication slot stats are not removed after invalidation. @@ -551,7 +551,7 @@ change_hot_standby_feedback_and_wait_for_xmins(1, 1); ################################################## $node_standby->restart; -# Verify conflict_reason is retained across a restart. +# Verify reason for conflict is retained across a restart. check_slots_conflict_reason('vacuum_full_', 'rows_removed'); ################################################## @@ -560,7 +560,8 @@ check_slots_conflict_reason('vacuum_full_', 'rows_removed'); # Get the restart_lsn from an invalidated slot my $restart_lsn = $node_standby->safe_psql('postgres', - "SELECT restart_lsn from pg_replication_slots WHERE slot_name = 'vacuum_full_activeslot' and conflict_reason is not null;" + "SELECT restart_lsn FROM pg_replication_slots + WHERE slot_name = 'vacuum_full_activeslot' AND conflicting;" ); chomp($restart_lsn); @@ -611,7 +612,7 @@ $node_primary->wait_for_replay_catchup($node_standby); # Check invalidation in the logfile and in pg_stat_database_conflicts check_for_invalidation('row_removal_', $logstart, 'with vacuum on pg_class'); -# Verify conflict_reason is 'rows_removed' in pg_replication_slots +# Verify reason for conflict is 'rows_removed' in pg_replication_slots check_slots_conflict_reason('row_removal_', 'rows_removed'); $handle = @@ -647,7 +648,7 @@ $node_primary->wait_for_replay_catchup($node_standby); check_for_invalidation('shared_row_removal_', $logstart, 'with vacuum on pg_authid'); -# Verify conflict_reason is 'rows_removed' in pg_replication_slots +# Verify reason for conflict is 'rows_removed' in pg_replication_slots check_slots_conflict_reason('shared_row_removal_', 'rows_removed'); $handle = make_slot_active($node_standby, 'shared_row_removal_', 0, \$stdout, @@ -696,14 +697,14 @@ ok( $node_standby->poll_query_until( 'confl_active_logicalslot not updated' ) or die "Timed out waiting confl_active_logicalslot to be updated"; -# Verify slots are reported as non conflicting in pg_replication_slots +# Verify slots are reported as valid in pg_replication_slots is( $node_standby->safe_psql( 'postgres', q[select bool_or(conflicting) from - (select conflict_reason is not NULL as conflicting - from pg_replication_slots WHERE slot_type = 'logical')]), + (select conflicting from pg_replication_slots + where slot_type = 'logical')]), 'f', - 'Logical slots are reported as non conflicting'); + 'Logical slots are reported as valid'); # Turn hot_standby_feedback back on change_hot_standby_feedback_and_wait_for_xmins(1, 0); @@ -739,7 +740,7 @@ $node_primary->wait_for_replay_catchup($node_standby); # Check invalidation in the logfile and in pg_stat_database_conflicts check_for_invalidation('pruning_', $logstart, 'with on-access pruning'); -# Verify conflict_reason is 'rows_removed' in pg_replication_slots +# Verify reason for conflict is 'rows_removed' in pg_replication_slots check_slots_conflict_reason('pruning_', 'rows_removed'); $handle = make_slot_active($node_standby, 'pruning_', 0, \$stdout, \$stderr); @@ -783,7 +784,7 @@ $node_primary->wait_for_replay_catchup($node_standby); # Check invalidation in the logfile and in pg_stat_database_conflicts check_for_invalidation('wal_level_', $logstart, 'due to wal_level'); -# Verify conflict_reason is 'wal_level_insufficient' in pg_replication_slots +# Verify reason for conflict is 'wal_level_insufficient' in pg_replication_slots check_slots_conflict_reason('wal_level_', 'wal_level_insufficient'); $handle = diff --git a/src/test/recovery/t/040_standby_failover_slots_sync.pl b/src/test/recovery/t/040_standby_failover_slots_sync.pl index 0ea1f3d323..f47bfd78eb 100644 --- a/src/test/recovery/t/040_standby_failover_slots_sync.pl +++ b/src/test/recovery/t/040_standby_failover_slots_sync.pl @@ -228,7 +228,7 @@ $standby1->safe_psql('postgres', "CHECKPOINT"); # Check if the synced slot is invalidated is( $standby1->safe_psql( 'postgres', - q{SELECT conflict_reason = 'wal_removed' FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';} + q{SELECT invalidation_reason = 'wal_removed' FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';} ), "t", 'synchronized slot has been invalidated'); @@ -274,7 +274,7 @@ $standby1->wait_for_log(qr/dropped replication slot "lsub1_slot" of dbid [0-9]+/ # flagged as 'synced' is( $standby1->safe_psql( 'postgres', - q{SELECT conflict_reason IS NULL AND synced AND NOT temporary FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';} + q{SELECT invalidation_reason IS NULL AND synced AND NOT temporary FROM pg_replication_slots WHERE slot_name = 'lsub1_slot';} ), "t", 'logical slot is re-synced'); diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 0cd2c64fca..055bec068d 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1473,10 +1473,11 @@ pg_replication_slots| SELECT l.slot_name, l.wal_status, l.safe_wal_size, l.two_phase, - l.conflict_reason, + l.conflicting, l.failover, - l.synced - FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflict_reason, failover, synced) + l.synced, + l.invalidation_reason + FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, failover, synced, invalidation_reason) LEFT JOIN pg_database d ON ((l.datoid = d.oid))); pg_roles| SELECT pg_authid.rolname, pg_authid.rolsuper, -- 2.34.1 [application/octet-stream] v11-0002-Add-XID-age-based-replication-slot-invalidation.patch (14.0K, ../../CALj2ACWBbSa7ccOGDHNfH1OX7_ArBJm1kNmc=iWKSfqq1NHnuA@mail.gmail.com/3-v11-0002-Add-XID-age-based-replication-slot-invalidation.patch) download | inline diff: From ea5acdd80b3a93dd8e9ae69628d237e71e9ad575 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Sat, 16 Mar 2024 03:46:28 +0000 Subject: [PATCH v11 2/4] Add XID age based replication slot invalidation Up until now, postgres has the ability to invalidate inactive replication slots based on the amount of WAL (set via max_slot_wal_keep_size GUC) that will be needed for the slots in case they become active. However, choosing a default value for max_slot_wal_keep_size is tricky. Because the amount of WAL a customer generates, and their allocated storage will vary greatly in production, making it difficult to pin down a one-size-fits-all value. It is often easy for developers to set an XID age (age of slot's xmin or catalog_xmin) of say 1 or 1.5 billion, after which the slots get invalidated. To achieve the above, postgres uses replication slot xmin (the oldest transaction that this slot needs the database to retain) or catalog_xmin (the oldest transaction affecting the system catalogs that this slot needs the database to retain), and a new GUC max_slot_xid_age. The checkpointer then looks at all replication slots invalidating the slots based on the age set. --- doc/src/sgml/config.sgml | 21 ++++ doc/src/sgml/system-views.sgml | 8 ++ src/backend/access/transam/xlog.c | 10 ++ src/backend/replication/slot.c | 49 +++++++- src/backend/utils/misc/guc_tables.c | 10 ++ src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/replication/slot.h | 3 + src/test/recovery/meson.build | 1 + src/test/recovery/t/050_invalidate_slots.pl | 108 ++++++++++++++++++ 9 files changed, 210 insertions(+), 1 deletion(-) create mode 100644 src/test/recovery/t/050_invalidate_slots.pl diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 65a6e6c408..6dd54ffcb7 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -4544,6 +4544,27 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"' # Windows </listitem> </varlistentry> + <varlistentry id="guc-max-slot-xid-age" xreflabel="max_slot_xid_age"> + <term><varname>max_slot_xid_age</varname> (<type>integer</type>) + <indexterm> + <primary><varname>max_slot_xid_age</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + Invalidate replication slots whose <literal>xmin</literal> (the oldest + transaction that this slot needs the database to retain) or + <literal>catalog_xmin</literal> (the oldest transaction affecting the + system catalogs that this slot needs the database to retain) has reached + the age specified by this setting. A value of zero (which is default) + disables this feature. Users can set this value anywhere from zero to + two billion. This parameter can only be set in the + <filename>postgresql.conf</filename> file or on the server command + line. + </para> + </listitem> + </varlistentry> + <varlistentry id="guc-track-commit-timestamp" xreflabel="track_commit_timestamp"> <term><varname>track_commit_timestamp</varname> (<type>boolean</type>) <indexterm> diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml index e685921847..56252b12ee 100644 --- a/doc/src/sgml/system-views.sgml +++ b/doc/src/sgml/system-views.sgml @@ -2588,6 +2588,14 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx perform logical decoding. </para> </listitem> + <listitem> + <para> + <literal>xid_aged</literal> means that the slot's + <literal>xmin</literal> or <literal>catalog_xmin</literal> + has reached the age specified by + <xref linkend="guc-max-slot-xid-age"/> parameter. + </para> + </listitem> </itemizedlist> </para></entry> </row> diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 20a5f86209..36ae2ac6a4 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -7147,6 +7147,11 @@ CreateCheckPoint(int flags) if (PriorRedoPtr != InvalidXLogRecPtr) UpdateCheckPointDistanceEstimate(RedoRecPtr - PriorRedoPtr); + /* Invalidate replication slots based on xmin or catalog_xmin age */ + if (max_slot_xid_age > 0) + InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0, + InvalidOid, InvalidTransactionId); + /* * Delete old log files, those no longer needed for last checkpoint to * prevent the disk holding the xlog from growing full. @@ -7597,6 +7602,11 @@ CreateRestartPoint(int flags) */ XLByteToSeg(RedoRecPtr, _logSegNo, wal_segment_size); + /* Invalidate replication slots based on xmin or catalog_xmin age */ + if (max_slot_xid_age > 0) + InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0, + InvalidOid, InvalidTransactionId); + /* * Retreat _logSegNo using the current end of xlog replayed or received, * whichever is later. diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 4f1a17f6ce..dc37586dcc 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -107,10 +107,11 @@ const char *const SlotInvalidationCauses[] = { [RS_INVAL_WAL_REMOVED] = "wal_removed", [RS_INVAL_HORIZON] = "rows_removed", [RS_INVAL_WAL_LEVEL] = "wal_level_insufficient", + [RS_INVAL_XID_AGE] = "xid_aged", }; /* Maximum number of invalidation causes */ -#define RS_INVAL_MAX_CAUSES RS_INVAL_WAL_LEVEL +#define RS_INVAL_MAX_CAUSES RS_INVAL_XID_AGE StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1), "array length mismatch"); @@ -140,6 +141,7 @@ ReplicationSlot *MyReplicationSlot = NULL; /* GUC variables */ int max_replication_slots = 10; /* the maximum number of replication * slots */ +int max_slot_xid_age = 0; /* * This GUC lists streaming replication standby server slot names that @@ -158,6 +160,7 @@ static XLogRecPtr ss_oldest_flush_lsn = InvalidXLogRecPtr; static void ReplicationSlotShmemExit(int code, Datum arg); static void ReplicationSlotDropPtr(ReplicationSlot *slot); +static bool IsSlotXIDAged(TransactionId xmin); /* internal persistency functions */ static void RestoreSlotFromDisk(const char *name); @@ -1483,6 +1486,9 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, case RS_INVAL_WAL_LEVEL: appendStringInfoString(&err_detail, _("Logical decoding on standby requires wal_level >= logical on the primary server.")); break; + case RS_INVAL_XID_AGE: + appendStringInfoString(&err_detail, _("The replication slot's xmin or catalog_xmin reached the age specified by max_slot_xid_age.")); + break; case RS_INVAL_NONE: pg_unreachable(); } @@ -1499,6 +1505,31 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, pfree(err_detail.data); } +/* + * Returns true if slot's passed in xmin/catalog_xmin age is more than + * max_slot_xid_age. + */ +static bool +IsSlotXIDAged(TransactionId xmin) +{ + TransactionId xid_cur; + TransactionId xid_limit; + + if (!TransactionIdIsNormal(xmin)) + return false; + + xid_cur = ReadNextTransactionId(); + xid_limit = xmin + max_slot_xid_age; + + if (xid_limit < FirstNormalTransactionId) + xid_limit += FirstNormalTransactionId; + + if (TransactionIdFollowsOrEquals(xid_cur, xid_limit)) + return true; + + return false; +} + /* * Helper for InvalidateObsoleteReplicationSlots * @@ -1599,6 +1630,21 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, if (SlotIsLogical(s)) conflict = cause; break; + case RS_INVAL_XID_AGE: + { + if (IsSlotXIDAged(s->data.xmin)) + { + conflict = cause; + break; + } + + if (IsSlotXIDAged(s->data.catalog_xmin)) + { + conflict = cause; + break; + } + } + break; case RS_INVAL_NONE: pg_unreachable(); } @@ -1752,6 +1798,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, * - RS_INVAL_HORIZON: requires a snapshot <= the given horizon in the given * db; dboid may be InvalidOid for shared relations * - RS_INVAL_WAL_LEVEL: is logical + * - RS_INVAL_XID_AGE: slot's xmin or catalog_xmin has reached the age * * NB - this runs as part of checkpoint, so avoid raising errors if possible. */ diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 57d9de4dd9..6b5375909d 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -2954,6 +2954,16 @@ struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + { + {"max_slot_xid_age", PGC_SIGHUP, REPLICATION_SENDING, + gettext_noop("Age of the transaction ID at which a replication slot gets invalidated."), + gettext_noop("The transaction is the oldest transaction (including the one affecting the system catalogs) that a replication slot needs the database to retain.") + }, + &max_slot_xid_age, + 0, 0, 2000000000, + NULL, NULL, NULL + }, + { {"commit_delay", PGC_SUSET, WAL_SETTINGS, gettext_noop("Sets the delay in microseconds between transaction commit and " diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index 2244ee52f7..b4c928b826 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -334,6 +334,7 @@ #wal_sender_timeout = 60s # in milliseconds; 0 disables #track_commit_timestamp = off # collect timestamp of transaction commit # (change requires restart) +#max_slot_xid_age = 0 # - Primary Server - diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 7f25a083ee..614ba0e30b 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -53,6 +53,8 @@ typedef enum ReplicationSlotInvalidationCause RS_INVAL_HORIZON, /* wal_level insufficient for slot */ RS_INVAL_WAL_LEVEL, + /* slot's xmin or catalog_xmin has reached the age */ + RS_INVAL_XID_AGE, } ReplicationSlotInvalidationCause; extern PGDLLIMPORT const char *const SlotInvalidationCauses[]; @@ -227,6 +229,7 @@ extern PGDLLIMPORT ReplicationSlot *MyReplicationSlot; /* GUCs */ extern PGDLLIMPORT int max_replication_slots; extern PGDLLIMPORT char *standby_slot_names; +extern PGDLLIMPORT int max_slot_xid_age; /* shmem initialization functions */ extern Size ReplicationSlotsShmemSize(void); diff --git a/src/test/recovery/meson.build b/src/test/recovery/meson.build index b1eb77b1ec..708a2a3798 100644 --- a/src/test/recovery/meson.build +++ b/src/test/recovery/meson.build @@ -51,6 +51,7 @@ tests += { 't/040_standby_failover_slots_sync.pl', 't/041_checkpoint_at_promote.pl', 't/042_low_level_backup.pl', + 't/050_invalidate_slots.pl', ], }, } diff --git a/src/test/recovery/t/050_invalidate_slots.pl b/src/test/recovery/t/050_invalidate_slots.pl new file mode 100644 index 0000000000..2f482b56e8 --- /dev/null +++ b/src/test/recovery/t/050_invalidate_slots.pl @@ -0,0 +1,108 @@ +# Copyright (c) 2024, PostgreSQL Global Development Group + +# Test for replication slots invalidation +use strict; +use warnings FATAL => 'all'; + +use PostgreSQL::Test::Utils; +use PostgreSQL::Test::Cluster; +use Test::More; +use Time::HiRes qw(usleep); + +# Initialize primary node, setting wal-segsize to 1MB +my $primary = PostgreSQL::Test::Cluster->new('primary'); +$primary->init(allows_streaming => 1, extra => ['--wal-segsize=1']); +$primary->append_conf( + 'postgresql.conf', q{ +checkpoint_timeout = 1h +}); +$primary->start; +$primary->safe_psql( + 'postgres', qq[ + SELECT pg_create_physical_replication_slot('sb1_slot'); +]); + +# Take backup +my $backup_name = 'my_backup'; +$primary->backup($backup_name); + +# Create a standby linking to the primary using the replication slot +my $standby1 = PostgreSQL::Test::Cluster->new('standby1'); +$standby1->init_from_backup($primary, $backup_name, has_streaming => 1); + +# Enable hs_feedback. The slot should gain an xmin. We set the status interval +# so we'll see the results promptly. +$standby1->append_conf( + 'postgresql.conf', q{ +primary_slot_name = 'sb1_slot' +hot_standby_feedback = on +wal_receiver_status_interval = 1 +}); +$standby1->start; + +# Create some content on primary to move xmin +$primary->safe_psql('postgres', + "CREATE TABLE tab_int AS SELECT generate_series(1,10) AS a"); + +# Wait until standby has replayed enough data +$primary->wait_for_catchup($standby1); + +$primary->poll_query_until( + 'postgres', qq[ + SELECT xmin IS NOT NULL + FROM pg_catalog.pg_replication_slots + WHERE slot_name = 'sb1_slot'; +]) or die "Timed out waiting for slot xmin to advance"; + +$primary->safe_psql( + 'postgres', qq[ + ALTER SYSTEM SET max_slot_xid_age = 500; +]); +$primary->reload; + +# Stop standby to make the replication slot's xmin on primary to age +$standby1->stop; + +my $logstart = -s $primary->logfile; + +# Do some work to advance xmin +$primary->safe_psql( + 'postgres', q{ +do $$ +begin + for i in 10000..11000 loop + -- use an exception block so that each iteration eats an XID + begin + insert into tab_int values (i); + exception + when division_by_zero then null; + end; + end loop; +end$$; +}); + +my $invalidated = 0; +for (my $i = 0; $i < 10 * $PostgreSQL::Test::Utils::timeout_default; $i++) +{ + $primary->safe_psql('postgres', "CHECKPOINT"); + if ($primary->log_contains( + 'invalidating obsolete replication slot "sb1_slot"', $logstart)) + { + $invalidated = 1; + last; + } + usleep(100_000); +} +ok($invalidated, 'check that slot sb1_slot invalidation has been logged'); + +# Wait for the inactive replication slots to be invalidated. +$primary->poll_query_until( + 'postgres', qq[ + SELECT COUNT(slot_name) = 1 FROM pg_replication_slots + WHERE slot_name = 'sb1_slot' AND + invalidation_reason = 'xid_aged'; +]) + or die + "Timed out while waiting for replication slot sb1_slot to be invalidated"; + +done_testing(); -- 2.34.1 [application/octet-stream] v11-0003-Track-inactive-replication-slot-information.patch (10.0K, ../../CALj2ACWBbSa7ccOGDHNfH1OX7_ArBJm1kNmc=iWKSfqq1NHnuA@mail.gmail.com/4-v11-0003-Track-inactive-replication-slot-information.patch) download | inline diff: From 3f08b8cc6346aabba5226d060823fcdf71e6b1f8 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Sat, 16 Mar 2024 03:47:15 +0000 Subject: [PATCH v11 3/4] Track inactive replication slot information Up until now, postgres doesn't track metrics like the time at which the slot became inactive, and the total number of times the slot became inactive in its lifetime. This commit adds two new metrics last_inactive_at of type timestamptz and inactive_count of type numeric to ReplicationSlotPersistentData. Whenever a slot becomes inactive, the current timestamp and inactive count are persisted to disk. These metrics are useful in the following ways: - To improve replication slot monitoring tools. For instance, one can build a monitoring tool that signals a) when replication slots is lying inactive for a day or so using last_inactive_at metric, b) when a replication slot is becoming inactive too frequently using inactive_count metric. - To implement timeout-based inactive replication slot management capability in postgres. Increases SLOT_VERSION due to the added two new metrics. --- doc/src/sgml/system-views.sgml | 20 +++++++++++++ src/backend/catalog/system_views.sql | 4 ++- src/backend/replication/slot.c | 43 ++++++++++++++++++++++------ src/backend/replication/slotfuncs.c | 15 +++++++++- src/include/catalog/pg_proc.dat | 6 ++-- src/include/replication/slot.h | 6 ++++ src/test/regress/expected/rules.out | 6 ++-- 7 files changed, 84 insertions(+), 16 deletions(-) diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml index 56252b12ee..365c0fd52d 100644 --- a/doc/src/sgml/system-views.sgml +++ b/doc/src/sgml/system-views.sgml @@ -2758,6 +2758,26 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx ID of role </para></entry> </row> + + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>last_inactive_at</structfield> <type>timestamptz</type> + </para> + <para> + The time at which the slot became inactive. + <literal>NULL</literal> if the slot is currently actively being + used. + </para></entry> + </row> + + <row> + <entry role="catalog_table_entry"><para role="column_definition"> + <structfield>inactive_count</structfield> <type>numeric</type> + </para> + <para> + The total number of times the slot became inactive in its lifetime. + </para></entry> + </row> </tbody> </tgroup> </table> diff --git a/src/backend/catalog/system_views.sql b/src/backend/catalog/system_views.sql index cd22dad959..de9f1d5506 100644 --- a/src/backend/catalog/system_views.sql +++ b/src/backend/catalog/system_views.sql @@ -1026,7 +1026,9 @@ CREATE VIEW pg_replication_slots AS L.conflicting, L.failover, L.synced, - L.invalidation_reason + L.invalidation_reason, + L.last_inactive_at, + L.inactive_count FROM pg_get_replication_slots() AS L LEFT JOIN pg_database D ON (L.datoid = D.oid); diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index dc37586dcc..6b6e5141f7 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -130,7 +130,7 @@ StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1), sizeof(ReplicationSlotOnDisk) - ReplicationSlotOnDiskConstantSize #define SLOT_MAGIC 0x1051CA1 /* format identifier */ -#define SLOT_VERSION 5 /* version for new files */ +#define SLOT_VERSION 6 /* version for new files */ /* Control array for replication slot management */ ReplicationSlotCtlData *ReplicationSlotCtl = NULL; @@ -401,6 +401,8 @@ ReplicationSlotCreate(const char *name, bool db_specific, slot->data.two_phase_at = InvalidXLogRecPtr; slot->data.failover = failover; slot->data.synced = synced; + slot->data.last_inactive_at = 0; + slot->data.inactive_count = 0; /* and then data only present in shared memory */ slot->just_dirtied = false; @@ -627,6 +629,17 @@ retry: if (am_walsender) { + if (s->data.persistency == RS_PERSISTENT) + { + SpinLockAcquire(&s->mutex); + s->data.last_inactive_at = 0; + SpinLockRelease(&s->mutex); + + /* Write this slot to disk */ + ReplicationSlotMarkDirty(); + ReplicationSlotSave(); + } + ereport(log_replication_commands ? LOG : DEBUG1, SlotIsLogical(s) ? errmsg("acquired logical replication slot \"%s\"", @@ -694,16 +707,20 @@ ReplicationSlotRelease(void) ConditionVariableBroadcast(&slot->active_cv); } - MyReplicationSlot = NULL; - - /* might not have been set when we've been a plain slot */ - LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE); - MyProc->statusFlags &= ~PROC_IN_LOGICAL_DECODING; - ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags; - LWLockRelease(ProcArrayLock); - if (am_walsender) { + if (slot->data.persistency == RS_PERSISTENT) + { + SpinLockAcquire(&slot->mutex); + slot->data.last_inactive_at = GetCurrentTimestamp(); + slot->data.inactive_count++; + SpinLockRelease(&slot->mutex); + + /* Write this slot to disk */ + ReplicationSlotMarkDirty(); + ReplicationSlotSave(); + } + ereport(log_replication_commands ? LOG : DEBUG1, is_logical ? errmsg("released logical replication slot \"%s\"", @@ -713,6 +730,14 @@ ReplicationSlotRelease(void) pfree(slotname); } + + MyReplicationSlot = NULL; + + /* might not have been set when we've been a plain slot */ + LWLockAcquire(ProcArrayLock, LW_EXCLUSIVE); + MyProc->statusFlags &= ~PROC_IN_LOGICAL_DECODING; + ProcGlobal->statusFlags[MyProc->pgxactoff] = MyProc->statusFlags; + LWLockRelease(ProcArrayLock); } /* diff --git a/src/backend/replication/slotfuncs.c b/src/backend/replication/slotfuncs.c index b5a638edea..4c7a120df1 100644 --- a/src/backend/replication/slotfuncs.c +++ b/src/backend/replication/slotfuncs.c @@ -239,7 +239,7 @@ pg_drop_replication_slot(PG_FUNCTION_ARGS) Datum pg_get_replication_slots(PG_FUNCTION_ARGS) { -#define PG_GET_REPLICATION_SLOTS_COLS 18 +#define PG_GET_REPLICATION_SLOTS_COLS 20 ReturnSetInfo *rsinfo = (ReturnSetInfo *) fcinfo->resultinfo; XLogRecPtr currlsn; int slotno; @@ -264,6 +264,7 @@ pg_get_replication_slots(PG_FUNCTION_ARGS) WALAvailability walstate; int i; ReplicationSlotInvalidationCause cause; + char buf[256]; if (!slot->in_use) continue; @@ -436,6 +437,18 @@ pg_get_replication_slots(PG_FUNCTION_ARGS) else values[i++] = CStringGetTextDatum(SlotInvalidationCauses[cause]); + if (slot_contents.data.last_inactive_at > 0) + values[i++] = TimestampTzGetDatum(slot_contents.data.last_inactive_at); + else + nulls[i++] = true; + + /* Convert to numeric. */ + snprintf(buf, sizeof buf, UINT64_FORMAT, slot_contents.data.inactive_count); + values[i++] = DirectFunctionCall3(numeric_in, + CStringGetDatum(buf), + ObjectIdGetDatum(0), + Int32GetDatum(-1)); + Assert(i == PG_GET_REPLICATION_SLOTS_COLS); tuplestore_putvalues(rsinfo->setResult, rsinfo->setDesc, diff --git a/src/include/catalog/pg_proc.dat b/src/include/catalog/pg_proc.dat index 63fd0b4cd7..c7ab0893eb 100644 --- a/src/include/catalog/pg_proc.dat +++ b/src/include/catalog/pg_proc.dat @@ -11123,9 +11123,9 @@ proname => 'pg_get_replication_slots', prorows => '10', proisstrict => 'f', proretset => 't', provolatile => 's', prorettype => 'record', proargtypes => '', - proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,bool,bool,text}', - proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}', - proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,failover,synced,invalidation_reason}', + proallargtypes => '{name,name,text,oid,bool,bool,int4,xid,xid,pg_lsn,pg_lsn,text,int8,bool,bool,bool,bool,text,timestamptz,numeric}', + proargmodes => '{o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o,o}', + proargnames => '{slot_name,plugin,slot_type,datoid,temporary,active,active_pid,xmin,catalog_xmin,restart_lsn,confirmed_flush_lsn,wal_status,safe_wal_size,two_phase,conflicting,failover,synced,invalidation_reason,last_inactive_at,inactive_count}', prosrc => 'pg_get_replication_slots' }, { oid => '3786', descr => 'set up a logical replication slot', proname => 'pg_create_logical_replication_slot', provolatile => 'v', diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 614ba0e30b..780767a819 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -129,6 +129,12 @@ typedef struct ReplicationSlotPersistentData * for logical slots on the primary server. */ bool failover; + + /* When did this slot become inactive last time? */ + TimestampTz last_inactive_at; + + /* How many times the slot has been inactive? */ + uint64 inactive_count; } ReplicationSlotPersistentData; /* diff --git a/src/test/regress/expected/rules.out b/src/test/regress/expected/rules.out index 055bec068d..c0bdfe76d8 100644 --- a/src/test/regress/expected/rules.out +++ b/src/test/regress/expected/rules.out @@ -1476,8 +1476,10 @@ pg_replication_slots| SELECT l.slot_name, l.conflicting, l.failover, l.synced, - l.invalidation_reason - FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, failover, synced, invalidation_reason) + l.invalidation_reason, + l.last_inactive_at, + l.inactive_count + FROM (pg_get_replication_slots() l(slot_name, plugin, slot_type, datoid, temporary, active, active_pid, xmin, catalog_xmin, restart_lsn, confirmed_flush_lsn, wal_status, safe_wal_size, two_phase, conflicting, failover, synced, invalidation_reason, last_inactive_at, inactive_count) LEFT JOIN pg_database d ON ((l.datoid = d.oid))); pg_roles| SELECT pg_authid.rolname, pg_authid.rolsuper, -- 2.34.1 [application/octet-stream] v11-0004-Add-inactive_timeout-based-replication-slot-inva.patch (12.0K, ../../CALj2ACWBbSa7ccOGDHNfH1OX7_ArBJm1kNmc=iWKSfqq1NHnuA@mail.gmail.com/5-v11-0004-Add-inactive_timeout-based-replication-slot-inva.patch) download | inline diff: From 233febfa7d67b53c1a5094ec494cb69caead5120 Mon Sep 17 00:00:00 2001 From: Bharath Rupireddy <[email protected]> Date: Sat, 16 Mar 2024 03:53:04 +0000 Subject: [PATCH v11 4/4] Add inactive_timeout based replication slot invalidation Up until now, postgres has the ability to invalidate inactive replication slots based on the amount of WAL (set via max_slot_wal_keep_size GUC) that will be needed for the slots in case they become active. However, choosing a default value for max_slot_wal_keep_size is tricky. Because the amount of WAL a customer generates, and their allocated storage will vary greatly in production, making it difficult to pin down a one-size-fits-all value. It is often easy for developers to set a timeout of say 1 or 2 or 3 days, after which the inactive slots get dropped. To achieve the above, postgres uses replication slot metric inactive_at (the time at which the slot became inactive), and a new GUC inactive_replication_slot_timeout. The checkpointer then looks at all replication slots invalidating the inactive slots based on the timeout set. --- doc/src/sgml/config.sgml | 18 +++++ doc/src/sgml/system-views.sgml | 7 ++ src/backend/access/transam/xlog.c | 10 +++ src/backend/replication/slot.c | 22 +++++- src/backend/utils/misc/guc_tables.c | 12 +++ src/backend/utils/misc/postgresql.conf.sample | 1 + src/include/replication/slot.h | 3 + src/test/recovery/t/050_invalidate_slots.pl | 79 +++++++++++++++++++ 8 files changed, 151 insertions(+), 1 deletion(-) diff --git a/doc/src/sgml/config.sgml b/doc/src/sgml/config.sgml index 6dd54ffcb7..4b0b60a1ac 100644 --- a/doc/src/sgml/config.sgml +++ b/doc/src/sgml/config.sgml @@ -4565,6 +4565,24 @@ restore_command = 'copy "C:\\server\\archivedir\\%f" "%p"' # Windows </listitem> </varlistentry> + <varlistentry id="guc-inactive-replication-slot-timeout" xreflabel="inactive_replication_slot_timeout"> + <term><varname>inactive_replication_slot_timeout</varname> (<type>integer</type>) + <indexterm> + <primary><varname>inactive_replication_slot_timeout</varname> configuration parameter</primary> + </indexterm> + </term> + <listitem> + <para> + Invalidate replication slots that are inactive for longer than this + amount of time at the next checkpoint. If this value is specified + without units, it is taken as seconds. A value of zero (which is + default) disables the timeout mechanism. This parameter can only be + set in the <filename>postgresql.conf</filename> file or on the server + command line. + </para> + </listitem> + </varlistentry> + <varlistentry id="guc-track-commit-timestamp" xreflabel="track_commit_timestamp"> <term><varname>track_commit_timestamp</varname> (<type>boolean</type>) <indexterm> diff --git a/doc/src/sgml/system-views.sgml b/doc/src/sgml/system-views.sgml index 365c0fd52d..c18dc5feb5 100644 --- a/doc/src/sgml/system-views.sgml +++ b/doc/src/sgml/system-views.sgml @@ -2596,6 +2596,13 @@ SELECT * FROM pg_locks pl LEFT JOIN pg_prepared_xacts ppx <xref linkend="guc-max-slot-xid-age"/> parameter. </para> </listitem> + <listitem> + <para> + <literal>inactive_timeout</literal> means that the slot has been + inactive for the duration specified by + <xref linkend="guc-inactive-replication-slot-timeout"/> parameter. + </para> + </listitem> </itemizedlist> </para></entry> </row> diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c index 36ae2ac6a4..166c3ed794 100644 --- a/src/backend/access/transam/xlog.c +++ b/src/backend/access/transam/xlog.c @@ -7152,6 +7152,11 @@ CreateCheckPoint(int flags) InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0, InvalidOid, InvalidTransactionId); + /* Invalidate inactive replication slots based on timeout */ + if (inactive_replication_slot_timeout > 0) + InvalidateObsoleteReplicationSlots(RS_INVAL_INACTIVE_TIMEOUT, 0, + InvalidOid, InvalidTransactionId); + /* * Delete old log files, those no longer needed for last checkpoint to * prevent the disk holding the xlog from growing full. @@ -7607,6 +7612,11 @@ CreateRestartPoint(int flags) InvalidateObsoleteReplicationSlots(RS_INVAL_XID_AGE, 0, InvalidOid, InvalidTransactionId); + /* Invalidate inactive replication slots based on timeout */ + if (inactive_replication_slot_timeout > 0) + InvalidateObsoleteReplicationSlots(RS_INVAL_INACTIVE_TIMEOUT, 0, + InvalidOid, InvalidTransactionId); + /* * Retreat _logSegNo using the current end of xlog replayed or received, * whichever is later. diff --git a/src/backend/replication/slot.c b/src/backend/replication/slot.c index 6b6e5141f7..10fe944623 100644 --- a/src/backend/replication/slot.c +++ b/src/backend/replication/slot.c @@ -108,10 +108,11 @@ const char *const SlotInvalidationCauses[] = { [RS_INVAL_HORIZON] = "rows_removed", [RS_INVAL_WAL_LEVEL] = "wal_level_insufficient", [RS_INVAL_XID_AGE] = "xid_aged", + [RS_INVAL_INACTIVE_TIMEOUT] = "inactive_timeout", }; /* Maximum number of invalidation causes */ -#define RS_INVAL_MAX_CAUSES RS_INVAL_XID_AGE +#define RS_INVAL_MAX_CAUSES RS_INVAL_INACTIVE_TIMEOUT StaticAssertDecl(lengthof(SlotInvalidationCauses) == (RS_INVAL_MAX_CAUSES + 1), "array length mismatch"); @@ -142,6 +143,7 @@ ReplicationSlot *MyReplicationSlot = NULL; int max_replication_slots = 10; /* the maximum number of replication * slots */ int max_slot_xid_age = 0; +int inactive_replication_slot_timeout = 0; /* * This GUC lists streaming replication standby server slot names that @@ -1514,6 +1516,9 @@ ReportSlotInvalidation(ReplicationSlotInvalidationCause cause, case RS_INVAL_XID_AGE: appendStringInfoString(&err_detail, _("The replication slot's xmin or catalog_xmin reached the age specified by max_slot_xid_age.")); break; + case RS_INVAL_INACTIVE_TIMEOUT: + appendStringInfoString(&err_detail, _("The slot has been inactive for more than the time specified by inactive_replication_slot_timeout.")); + break; case RS_INVAL_NONE: pg_unreachable(); } @@ -1670,6 +1675,20 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, } } break; + case RS_INVAL_INACTIVE_TIMEOUT: + if (s->data.last_inactive_at > 0) + { + TimestampTz now; + + Assert(s->data.persistency == RS_PERSISTENT); + Assert(s->active_pid == 0); + + now = GetCurrentTimestamp(); + if (TimestampDifferenceExceeds(s->data.last_inactive_at, now, + inactive_replication_slot_timeout * 1000)) + conflict = cause; + } + break; case RS_INVAL_NONE: pg_unreachable(); } @@ -1824,6 +1843,7 @@ InvalidatePossiblyObsoleteSlot(ReplicationSlotInvalidationCause cause, * db; dboid may be InvalidOid for shared relations * - RS_INVAL_WAL_LEVEL: is logical * - RS_INVAL_XID_AGE: slot's xmin or catalog_xmin has reached the age + * - RS_INVAL_INACTIVE_TIMEOUT: inactive slot timeout occurs * * NB - this runs as part of checkpoint, so avoid raising errors if possible. */ diff --git a/src/backend/utils/misc/guc_tables.c b/src/backend/utils/misc/guc_tables.c index 6b5375909d..6caf40d51e 100644 --- a/src/backend/utils/misc/guc_tables.c +++ b/src/backend/utils/misc/guc_tables.c @@ -2964,6 +2964,18 @@ struct config_int ConfigureNamesInt[] = NULL, NULL, NULL }, + { + {"inactive_replication_slot_timeout", PGC_SIGHUP, REPLICATION_SENDING, + gettext_noop("Sets the amount of time to wait before invalidating an " + "inactive replication slot."), + NULL, + GUC_UNIT_S + }, + &inactive_replication_slot_timeout, + 0, 0, INT_MAX, + NULL, NULL, NULL + }, + { {"commit_delay", PGC_SUSET, WAL_SETTINGS, gettext_noop("Sets the delay in microseconds between transaction commit and " diff --git a/src/backend/utils/misc/postgresql.conf.sample b/src/backend/utils/misc/postgresql.conf.sample index b4c928b826..7f2a3e41f1 100644 --- a/src/backend/utils/misc/postgresql.conf.sample +++ b/src/backend/utils/misc/postgresql.conf.sample @@ -261,6 +261,7 @@ #recovery_prefetch = try # prefetch pages referenced in the WAL? #wal_decode_buffer_size = 512kB # lookahead window used for prefetching # (change requires restart) +#inactive_replication_slot_timeout = 0 # in seconds; 0 disables # - Archiving - diff --git a/src/include/replication/slot.h b/src/include/replication/slot.h index 780767a819..8378d7c913 100644 --- a/src/include/replication/slot.h +++ b/src/include/replication/slot.h @@ -55,6 +55,8 @@ typedef enum ReplicationSlotInvalidationCause RS_INVAL_WAL_LEVEL, /* slot's xmin or catalog_xmin has reached the age */ RS_INVAL_XID_AGE, + /* inactive slot timeout has occurred */ + RS_INVAL_INACTIVE_TIMEOUT, } ReplicationSlotInvalidationCause; extern PGDLLIMPORT const char *const SlotInvalidationCauses[]; @@ -236,6 +238,7 @@ extern PGDLLIMPORT ReplicationSlot *MyReplicationSlot; extern PGDLLIMPORT int max_replication_slots; extern PGDLLIMPORT char *standby_slot_names; extern PGDLLIMPORT int max_slot_xid_age; +extern PGDLLIMPORT int inactive_replication_slot_timeout; /* shmem initialization functions */ extern Size ReplicationSlotsShmemSize(void); diff --git a/src/test/recovery/t/050_invalidate_slots.pl b/src/test/recovery/t/050_invalidate_slots.pl index 2f482b56e8..4c66dd4a4e 100644 --- a/src/test/recovery/t/050_invalidate_slots.pl +++ b/src/test/recovery/t/050_invalidate_slots.pl @@ -105,4 +105,83 @@ $primary->poll_query_until( or die "Timed out while waiting for replication slot sb1_slot to be invalidated"; +$primary->safe_psql( + 'postgres', qq[ + SELECT pg_create_physical_replication_slot('sb2_slot'); +]); + +$primary->safe_psql( + 'postgres', qq[ + ALTER SYSTEM SET max_slot_xid_age = 0; +]); +$primary->reload; + +# Create a standby linking to the primary using the replication slot +my $standby2 = PostgreSQL::Test::Cluster->new('standby2'); +$standby2->init_from_backup($primary, $backup_name, has_streaming => 1); +$standby2->append_conf( + 'postgresql.conf', q{ +primary_slot_name = 'sb2_slot' +}); +$standby2->start; + +# Wait until standby has replayed enough data +$primary->wait_for_catchup($standby2); + +# The inactive replication slot info should be null when the slot is active +my $result = $primary->safe_psql( + 'postgres', qq[ + SELECT last_inactive_at IS NULL, inactive_count = 0 AS OK + FROM pg_replication_slots WHERE slot_name = 'sb2_slot'; +]); +is($result, "t|t", + 'check the inactive replication slot info for an active slot'); + +# Set timeout so that the next checkpoint will invalidate the inactive +# replication slot. +$primary->safe_psql( + 'postgres', qq[ + ALTER SYSTEM SET inactive_replication_slot_timeout TO '1s'; +]); +$primary->reload; + +$logstart = -s $primary->logfile; + +# Stop standby to make the replication slot on primary inactive +$standby2->stop; + +# Wait for the inactive replication slot info to be updated +$primary->poll_query_until( + 'postgres', qq[ + SELECT COUNT(slot_name) = 1 FROM pg_replication_slots + WHERE last_inactive_at IS NOT NULL AND + inactive_count = 1 AND slot_name = 'sb2_slot'; +]) + or die + "Timed out while waiting for inactive replication slot info to be updated"; + +$invalidated = 0; +for (my $i = 0; $i < 10 * $PostgreSQL::Test::Utils::timeout_default; $i++) +{ + $primary->safe_psql('postgres', "CHECKPOINT"); + if ($primary->log_contains( + 'invalidating obsolete replication slot "sb2_slot"', $logstart)) + { + $invalidated = 1; + last; + } + usleep(100_000); +} +ok($invalidated, 'check that slot sb2_slot invalidation has been logged'); + +# Wait for the inactive replication slots to be invalidated. +$primary->poll_query_until( + 'postgres', qq[ + SELECT COUNT(slot_name) = 1 FROM pg_replication_slots + WHERE slot_name = 'sb2_slot' AND + invalidation_reason = 'inactive_timeout'; +]) + or die + "Timed out while waiting for inactive replication slot sb2_slot to be invalidated"; + done_testing(); -- 2.34.1 ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-18 09:18 Bertrand Drouvot <[email protected]> parent: Bharath Rupireddy <[email protected]> 1 sibling, 0 replies; 18+ messages in thread From: Bertrand Drouvot @ 2024-03-18 09:18 UTC (permalink / raw) To: Bharath Rupireddy <[email protected]>; +Cc: shveta malik <[email protected]>; Amit Kapila <[email protected]>; Nathan Bossart <[email protected]>; PostgreSQL Hackers <[email protected]> Hi, On Sat, Mar 16, 2024 at 09:29:01AM +0530, Bharath Rupireddy wrote: > I've also responded to Bertrand's comments here. Thanks! > > On Wed, Mar 6, 2024 at 3:56 PM Bertrand Drouvot > <[email protected]> wrote: > > > > 5 === > > > > -# Check conflict_reason is NULL for physical slot > > +# Check invalidation_reason is NULL for physical slot > > $res = $node_primary->safe_psql( > > 'postgres', qq[ > > - SELECT conflict_reason is null FROM pg_replication_slots where slot_name = '$primary_slotname';] > > + SELECT invalidation_reason is null FROM pg_replication_slots where slot_name = '$primary_slotname';] > > ); > > > > > > I don't think this test is needed anymore: it does not make that much sense since > > it's done after the primary database initialization and startup. > > It is now turned into a test verifying 'conflicting boolean' is null > for the physical slot. Isn't that okay? Yeah makes more sense now, thanks! Regards, -- Bertrand Drouvot PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-18 10:12 Bertrand Drouvot <[email protected]> parent: Bharath Rupireddy <[email protected]> 1 sibling, 1 reply; 18+ messages in thread From: Bertrand Drouvot @ 2024-03-18 10:12 UTC (permalink / raw) To: Bharath Rupireddy <[email protected]>; +Cc: shveta malik <[email protected]>; Amit Kapila <[email protected]>; Nathan Bossart <[email protected]>; PostgreSQL Hackers <[email protected]> Hi, On Sat, Mar 16, 2024 at 09:29:01AM +0530, Bharath Rupireddy wrote: > Please see the attached v11 patch set with all the above review > comments addressed. Thanks! Looking at 0001: 1 === + True if this logical slot conflicted with recovery (and so is now + invalidated). When this column is true, check Worth to add back the physical slot mention "Always NULL for physical slots."? 2 === @@ -1023,9 +1023,10 @@ CREATE VIEW pg_replication_slots AS L.wal_status, L.safe_wal_size, L.two_phase, - L.conflict_reason, + L.conflicting, L.failover, - L.synced + L.synced, + L.invalidation_reason What about making invalidation_reason close to conflict_reason? 3 === - * Maps a conflict reason for a replication slot to + * Maps a invalidation reason for a replication slot to s/a invalidation/an invalidation/? 4 === While at it, shouldn't we also rename "conflict" to say "invalidation_cause" in InvalidatePossiblyObsoleteSlot()? 5 === + * rows_removed and wal_level_insufficient are only two reasons s/are only two/are the only two/? Regards, -- Bertrand Drouvot PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 18+ messages in thread
* Re: Introduce XID age and inactive timeout based replication slot invalidation @ 2024-03-20 05:47 Bharath Rupireddy <[email protected]> parent: Bertrand Drouvot <[email protected]> 0 siblings, 0 replies; 18+ messages in thread From: Bharath Rupireddy @ 2024-03-20 05:47 UTC (permalink / raw) To: Bertrand Drouvot <[email protected]>; +Cc: shveta malik <[email protected]>; Amit Kapila <[email protected]>; Nathan Bossart <[email protected]>; PostgreSQL Hackers <[email protected]> On Mon, Mar 18, 2024 at 3:42 PM Bertrand Drouvot <[email protected]> wrote: > > Hi, > > Looking at 0001: Thanks for reviewing. > 1 === > > + True if this logical slot conflicted with recovery (and so is now > + invalidated). When this column is true, check > > Worth to add back the physical slot mention "Always NULL for physical slots."? Will change. > 2 === > > @@ -1023,9 +1023,10 @@ CREATE VIEW pg_replication_slots AS > L.wal_status, > L.safe_wal_size, > L.two_phase, > - L.conflict_reason, > + L.conflicting, > L.failover, > - L.synced > + L.synced, > + L.invalidation_reason > > What about making invalidation_reason close to conflict_reason? Not required I think. One can pick the required columns in the SELECT clause anyways. > 3 === > > - * Maps a conflict reason for a replication slot to > + * Maps a invalidation reason for a replication slot to > > s/a invalidation/an invalidation/? Will change. > 4 === > > While at it, shouldn't we also rename "conflict" to say "invalidation_cause" in > InvalidatePossiblyObsoleteSlot()? That's inline with our understanding about conflict vs invalidation, and keeps the function generic. Will change. > 5 === > > + * rows_removed and wal_level_insufficient are only two reasons > > s/are only two/are the only two/? Will change.. -- Bharath Rupireddy PostgreSQL Contributors Team RDS Open Source Databases Amazon Web Services: https://aws.amazon.com ^ permalink raw reply [nested|flat] 18+ messages in thread
end of thread, other threads:[~2024-03-20 05:47 UTC | newest] Thread overview: 18+ messages (download: mbox mbox.gz follow: Atom feed) -- links below jump to the message on this page -- 2019-09-05 11:21 [PATCH v8 1/4] Move callback-call from ReadPageInternal to XLogReadRecord. Kyotaro Horiguchi <[email protected]> 2023-08-31 14:59 [PATCH v9 3/4] add support for syncfs in frontend support functions Nathan Bossart <[email protected]> 2024-03-08 14:38 Re: Introduce XID age and inactive timeout based replication slot invalidation Bharath Rupireddy <[email protected]> 2024-03-11 05:55 ` Re: Introduce XID age and inactive timeout based replication slot invalidation Amit Kapila <[email protected]> 2024-03-12 15:25 ` Re: Introduce XID age and inactive timeout based replication slot invalidation Bharath Rupireddy <[email protected]> 2024-03-13 03:51 ` Re: Introduce XID age and inactive timeout based replication slot invalidation Amit Kapila <[email protected]> 2024-03-13 15:54 ` Re: Introduce XID age and inactive timeout based replication slot invalidation Bharath Rupireddy <[email protected]> 2024-03-14 06:54 ` Re: Introduce XID age and inactive timeout based replication slot invalidation Amit Kapila <[email protected]> 2024-03-14 14:27 ` Re: Introduce XID age and inactive timeout based replication slot invalidation Bharath Rupireddy <[email protected]> 2024-03-15 04:44 ` Re: Introduce XID age and inactive timeout based replication slot invalidation shveta malik <[email protected]> 2024-03-15 12:05 ` Re: Introduce XID age and inactive timeout based replication slot invalidation Bharath Rupireddy <[email protected]> 2024-03-15 07:19 ` Re: Introduce XID age and inactive timeout based replication slot invalidation shveta malik <[email protected]> 2024-03-16 03:59 ` Re: Introduce XID age and inactive timeout based replication slot invalidation Bharath Rupireddy <[email protected]> 2024-03-18 09:18 ` Re: Introduce XID age and inactive timeout based replication slot invalidation Bertrand Drouvot <[email protected]> 2024-03-18 10:12 ` Re: Introduce XID age and inactive timeout based replication slot invalidation Bertrand Drouvot <[email protected]> 2024-03-20 05:47 ` Re: Introduce XID age and inactive timeout based replication slot invalidation Bharath Rupireddy <[email protected]> 2024-03-15 14:28 ` Re: Introduce XID age and inactive timeout based replication slot invalidation Nathan Bossart <[email protected]> 2024-03-15 16:45 ` Re: Introduce XID age and inactive timeout based replication slot invalidation Bertrand Drouvot <[email protected]>
This inbox is served by agora; see mirroring instructions for how to clone and mirror all data and code used for this inbox