From: Alvaro Herrera <alvherre@alvh.no-ip.org>
To: Kyotaro Horiguchi <horikyota.ntt@gmail.com>
Cc: bossartn@amazon.com
Cc: andres@anarazel.de
Cc: masao.fujii@oss.nttdata.com
Cc: pgsql-hackers@lists.postgresql.org
Cc: mengjuan.cmj@alibaba-inc.com
Cc: Jakub.Wartak@tomtom.com
Subject: Re: prevent immature WAL streaming
Date: Tue, 14 Sep 2021 22:32:04 -0300
Message-ID: <202109150132.73bzmi4bcppw@alvherre.pgsql> (raw)
In-Reply-To: <202109141755.r4ivg6an5pfp@alvherre.pgsql>
On 2021-Sep-14, Alvaro Herrera wrote:
> On 2021-Sep-08, Kyotaro Horiguchi wrote:
>
> > Thanks! As my understanding the new record add the ability to
> > cross-check between a teard-off contrecord and the new record inserted
> > after the teard-off record. I didn't test the version by myself but
> > the previous version implemented the essential machinery and that
> > won't change fundamentally by the new record.
> >
> > So I think the current patch deserves to see the algorithm actually
> > works against the problem.
>
> Here's a version with the new record type. It passes check-world, and
> it seems to work correctly to prevent overwrite of the tail end of a
> segment containing a broken record. This is very much WIP still;
> comments are missing and I haven't tried to implement any sort of
> verification that the record being aborted is the right one.
Here's the attachment I forgot earlier.
--
Álvaro Herrera 39°49'30"S 73°17'W — https://www.EnterpriseDB.com/
"[PostgreSQL] is a great group; in my opinion it is THE best open source
development communities in existence anywhere." (Lamar Owen)
Attachments:
[text/x-diff] v5-0001-Implement-FIRST_IS_ABORTED_CONTRECORD.patch (14.8K, ../202109150132.73bzmi4bcppw@alvherre.pgsql/2-v5-0001-Implement-FIRST_IS_ABORTED_CONTRECORD.patch)
download | inline diff:
From 3de9660b2e570604412d435023dcbd13355fc123 Mon Sep 17 00:00:00 2001
From: Alvaro Herrera <alvherre@alvh.no-ip.org>
Date: Thu, 2 Sep 2021 17:21:46 -0400
Subject: [PATCH v5] Implement FIRST_IS_ABORTED_CONTRECORD
---
src/backend/access/rmgrdesc/xlogdesc.c | 11 +++
src/backend/access/transam/xlog.c | 91 +++++++++++++++++++++++--
src/backend/access/transam/xloginsert.c | 3 +
src/backend/access/transam/xlogreader.c | 39 ++++++++++-
src/include/access/xlog.h | 1 +
src/include/access/xlog_internal.h | 20 +++++-
src/include/access/xlogreader.h | 3 +
src/include/catalog/pg_control.h | 1 +
8 files changed, 159 insertions(+), 10 deletions(-)
diff --git a/src/backend/access/rmgrdesc/xlogdesc.c b/src/backend/access/rmgrdesc/xlogdesc.cindex e6090a9dad..0be382f8a5 100644--- a/src/backend/access/rmgrdesc/xlogdesc.c+++ b/src/backend/access/rmgrdesc/xlogdesc.c@@ -139,6 +139,14 @@ xlog_desc(StringInfo buf, XLogReaderState *record)
xlrec.ThisTimeLineID, xlrec.PrevTimeLineID,
timestamptz_to_str(xlrec.end_time));
}
+ else if (info == XLOG_OVERWRITE_CONTRECORD)+ {+ xl_overwrite_contrecord xlrec;++ memcpy(&xlrec, rec, sizeof(xl_overwrite_contrecord));+ appendStringInfo(buf, "lsn %X/%X",+ LSN_FORMAT_ARGS(xlrec.overwritten_lsn));+ }
}
const char *
@@ -178,6 +186,9 @@ xlog_identify(uint8 info)
case XLOG_END_OF_RECOVERY:
id = "END_OF_RECOVERY";
break;
+ case XLOG_OVERWRITE_CONTRECORD:+ id = "OVERWRITE_CONTRECORD";+ break;
case XLOG_FPI:
id = "FPI";
break;
diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.cindex e51a7a749d..73cfd64125 100644--- a/src/backend/access/transam/xlog.c+++ b/src/backend/access/transam/xlog.c@@ -586,6 +586,8 @@ typedef struct XLogCtlData
XLogRecPtr replicationSlotMinLSN; /* oldest LSN needed by any slot */
XLogSegNo lastRemovedSegNo; /* latest removed/recycled XLOG segment */
+ XLogRecPtr abortedContrecordPtr; /* LSN of incomplete record at end of+ * WAL */
/* Fake LSN counter, for unlogged relations. Protected by ulsn_lck. */
XLogRecPtr unloggedLSN;
@@ -894,6 +896,7 @@ static void checkTimeLineSwitch(XLogRecPtr lsn, TimeLineID newTLI,
TimeLineID prevTLI);
static void LocalSetXLogInsertAllowed(void);
static void CreateEndOfRecoveryRecord(void);
+static XLogRecPtr CreateOverwriteContrecordRecord(XLogRecPtr aborted_lsn);
static void CheckPointGuts(XLogRecPtr checkPointRedo, int flags);
static void KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo);
static XLogRecPtr XLogGetReplicationSlotMinimumLSN(void);
@@ -950,6 +953,7 @@ static void rm_redo_error_callback(void *arg);
static int get_sync_bit(int method);
static void CopyXLogRecordToWAL(int write_len, bool isLogSwitch,
+ bool set_aborted_partial,
XLogRecData *rdata,
XLogRecPtr StartPos, XLogRecPtr EndPos);
static void ReserveXLogInsertLocation(int size, XLogRecPtr *StartPos,
@@ -1120,8 +1124,9 @@ XLogInsertRecord(XLogRecData *rdata,
* All the record data, including the header, is now ready to be
* inserted. Copy the record in the space reserved.
*/
- CopyXLogRecordToWAL(rechdr->xl_tot_len, isLogSwitch, rdata,- StartPos, EndPos);+ CopyXLogRecordToWAL(rechdr->xl_tot_len, isLogSwitch,+ flags & XLOG_SET_ABORTED_PARTIAL,+ rdata, StartPos, EndPos);
/*
* Unless record is flagged as not important, update LSN of last
@@ -1504,7 +1509,8 @@ checkXLogConsistency(XLogReaderState *record)
* area in the WAL.
*/
static void
-CopyXLogRecordToWAL(int write_len, bool isLogSwitch, XLogRecData *rdata,+CopyXLogRecordToWAL(int write_len, bool isLogSwitch,+ bool set_aborted_partial, XLogRecData *rdata,
XLogRecPtr StartPos, XLogRecPtr EndPos)
{
char *currpos;
@@ -1560,6 +1566,10 @@ CopyXLogRecordToWAL(int write_len, bool isLogSwitch, XLogRecData *rdata,
pagehdr->xlp_rem_len = write_len - written;
pagehdr->xlp_info |= XLP_FIRST_IS_CONTRECORD;
+ /* XXX can only happen once in the loop. Verify? */+ if (set_aborted_partial)+ pagehdr->xlp_info |= XLP_FIRST_IS_ABORTED_PARTIAL;+
/* skip over the page header */
if (XLogSegmentOffset(CurrPos, wal_segment_size) == 0)
{
@@ -4394,6 +4404,19 @@ ReadRecord(XLogReaderState *xlogreader, int emode,
EndRecPtr = xlogreader->EndRecPtr;
if (record == NULL)
{
+ /*+ * When WAL ends in an incomplete record, keep track of it. After+ * recovery is done, we'll write a record to indicate downstream+ * WAL readers that that portion is to be ignored.+ */+ if (InRecovery &&+ !XLogRecPtrIsInvalid(xlogreader->abortedContrecordPtr))+ {+ SpinLockAcquire(&XLogCtl->info_lck);+ XLogCtl->abortedContrecordPtr = xlogreader->abortedContrecordPtr;+ SpinLockRelease(&XLogCtl->info_lck);+ }+
if (readFile >= 0)
{
close(readFile);
@@ -6502,6 +6525,7 @@ StartupXLOG(void)
bool haveTblspcMap = false;
XLogRecPtr RecPtr,
checkPointLoc,
+ aborted_lsn,
EndOfLog;
TimeLineID EndOfLogTLI;
TimeLineID PrevTimeLineID;
@@ -7655,8 +7679,9 @@ StartupXLOG(void)
/*
* Kill WAL receiver, if it's still running, before we continue to write
- * the startup checkpoint record. It will trump over the checkpoint and- * subsequent records if it's still alive when we start writing WAL.+ * the startup checkpoint and aborted-contrecord records. It will trump+ * over these records and subsequent ones if it's still alive when we+ * start writing WAL.
*/
XLogShutdownWalRcv();
@@ -7689,8 +7714,12 @@ StartupXLOG(void)
StandbyMode = false;
/*
- * Re-fetch the last valid or last applied record, so we can identify the- * exact endpoint of what we consider the valid portion of WAL.+ * Determine where to start writing WAL next.+ *+ * When recovery ended in an incomplete record, write a WAL record about+ * that and continue after it. In all other cases, re-fetch the last+ * valid or last applied record, so we can identify the exact endpoint of+ * what we consider the valid portion of WAL.
*/
XLogBeginRead(xlogreader, LastRec);
record = ReadRecord(xlogreader, PANIC, false);
@@ -7821,6 +7850,17 @@ StartupXLOG(void)
XLogCtl->ThisTimeLineID = ThisTimeLineID;
XLogCtl->PrevTimeLineID = PrevTimeLineID;
+ /* XXX explain EndOfLog change */+ SpinLockAcquire(&XLogCtl->info_lck);+ if (!XLogRecPtrIsInvalid(XLogCtl->abortedContrecordPtr))+ {+ aborted_lsn = EndOfLog = XLogCtl->abortedContrecordPtr;+ XLogCtl->abortedContrecordPtr = InvalidXLogRecPtr;+ }+ else+ aborted_lsn = InvalidXLogRecPtr;+ SpinLockRelease(&XLogCtl->info_lck);+
/*
* Prepare to write WAL starting at EndOfLog location, and init xlog
* buffer cache using the block containing the last record from the
@@ -7880,6 +7920,8 @@ StartupXLOG(void)
*/
Insert->fullPageWrites = lastFullPageWrites;
LocalSetXLogInsertAllowed();
+ if (!XLogRecPtrIsInvalid(aborted_lsn))+ CreateOverwriteContrecordRecord(aborted_lsn);
UpdateFullPageWrites();
LocalXLogInsertAllowed = -1;
@@ -9365,6 +9407,37 @@ CreateEndOfRecoveryRecord(void)
LocalXLogInsertAllowed = -1; /* return to "check" state */
}
+/* XXX document this */+static XLogRecPtr+CreateOverwriteContrecordRecord(XLogRecPtr aborted_lsn)+{+ xl_overwrite_contrecord xlrec;+ XLogRecPtr recptr;++ /* sanity check */+ if (!RecoveryInProgress())+ elog(ERROR, "can only be used at end of recovery");++ xlrec.overwritten_lsn = aborted_lsn;+ /* XXX include the CRC-so-far? */++ START_CRIT_SECTION();++ XLogBeginInsert();+ XLogRegisterData((char *) &xlrec, sizeof(xl_overwrite_contrecord));++ /* this record overwrites the tail of the incomplete record */+ XLogSetRecordFlags(XLOG_SET_ABORTED_PARTIAL);++ recptr = XLogInsert(RM_XLOG_ID, XLOG_OVERWRITE_CONTRECORD);++ XLogFlush(recptr);++ END_CRIT_SECTION();++ return recptr;+}+
/*
* Flush all data in shared memory to disk, and fsync
*
@@ -10295,6 +10368,10 @@ xlog_redo(XLogReaderState *record)
RecoveryRestartPoint(&checkPoint);
}
+ else if (info == XLOG_OVERWRITE_CONTRECORD)+ {+ /* XXX verify the LSN of the record we're "aborting" */+ }
else if (info == XLOG_END_OF_RECOVERY)
{
xl_end_of_recovery xlrec;
diff --git a/src/backend/access/transam/xloginsert.c b/src/backend/access/transam/xloginsert.cindex e596a0470a..23c46ba21f 100644--- a/src/backend/access/transam/xloginsert.c+++ b/src/backend/access/transam/xloginsert.c@@ -409,6 +409,9 @@ XLogRegisterBufData(uint8 block_id, char *data, int len)
* - XLOG_MARK_UNIMPORTANT, to signal that the record is not important for
* durability, which allows to avoid triggering WAL archiving and other
* background activity.
+ * - XLOG_INCLUDE_XID, ???+ * - XLOG_OVERWRITE_CONTRECORD, to signal that XLP_FIRST_IS_ABORTED_PARTIAL+ * is to be set.
*/
void
XLogSetRecordFlags(uint8 flags)
diff --git a/src/backend/access/transam/xlogreader.c b/src/backend/access/transam/xlogreader.cindex 5cf74e181a..dbfa6d3562 100644--- a/src/backend/access/transam/xlogreader.c+++ b/src/backend/access/transam/xlogreader.c@@ -278,6 +278,7 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
total_len;
uint32 targetRecOff;
uint32 pageHeaderSize;
+ bool assembled;
bool gotheader;
int readOff;
@@ -293,6 +294,7 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
state->errormsg_buf[0] = '\0';
ResetDecoder(state);
+ state->abortedContrecordPtr = InvalidXLogRecPtr;
RecPtr = state->EndRecPtr;
@@ -319,7 +321,9 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
randAccess = true;
}
+restart:
state->currRecPtr = RecPtr;
+ assembled = false;
targetPagePtr = RecPtr - (RecPtr % XLOG_BLCKSZ);
targetRecOff = RecPtr % XLOG_BLCKSZ;
@@ -415,6 +419,8 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
char *buffer;
uint32 gotlen;
+ assembled = true;+
/*
* Enlarge readRecordBuf as needed.
*/
@@ -442,14 +448,28 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
readOff = ReadPageInternal(state, targetPagePtr,
Min(total_len - gotlen + SizeOfXLogShortPHD,
XLOG_BLCKSZ));
-
if (readOff < 0)
goto err;
Assert(SizeOfXLogShortPHD <= readOff);
- /* Check that the continuation on next page looks valid */
pageHeader = (XLogPageHeader) state->readBuf;
++ /*+ * If we were expecting a continuation record and got an "aborted+ * partial" flag, that means the continuation record was lost.+ * Ignore the record we were reading, since we now know it's broken+ * and lost forever, and restart the read by assuming the address+ * to read is the location where we found this flag.+ */+ if (pageHeader->xlp_info & XLP_FIRST_IS_ABORTED_PARTIAL)+ {+ ResetDecoder(state);+ RecPtr = targetPagePtr;+ goto restart;+ }++ /* Check that the continuation on next page looks valid */
if (!(pageHeader->xlp_info & XLP_FIRST_IS_CONTRECORD))
{
report_invalid_record(state,
@@ -551,6 +571,21 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
return NULL;
err:
+ if (assembled)+ {+ /*+ * We get here when a record that spans multiple pages needs to be+ * assembled, but something went wrong -- perhaps a contrecord piece+ * was lost. We deal with this by setting abortedContrecordPtr to the+ * location of the piece we failed to read, or the start of the page+ * we read where validation failed. If caller is WAL replay, it will+ * know that recovery ended and that this is where to start writing+ * future WAL marking the next piece with XLP_FIRST_IS_ABORTED_PARTIAL,+ * which will in turn signal downstream WAL consumers that the broken+ * WAL record here is to be ignored.+ */+ state->abortedContrecordPtr = targetPagePtr;+ }
/*
* Invalidate the read state. We might read from a different source after
diff --git a/src/include/access/xlog.h b/src/include/access/xlog.hindex 0a8ede700d..caed9294a8 100644--- a/src/include/access/xlog.h+++ b/src/include/access/xlog.h@@ -213,6 +213,7 @@ extern bool XLOG_DEBUG;
#define XLOG_INCLUDE_ORIGIN 0x01 /* include the replication origin */
#define XLOG_MARK_UNIMPORTANT 0x02 /* record not important for durability */
#define XLOG_INCLUDE_XID 0x04 /* include XID of top-level xact */
+#define XLOG_SET_ABORTED_PARTIAL 0x08 /* set the aborted-partial page flag */
/* Checkpoint statistics */
diff --git a/src/include/access/xlog_internal.h b/src/include/access/xlog_internal.hindex 3b5eceff65..0fb44ba37d 100644--- a/src/include/access/xlog_internal.h+++ b/src/include/access/xlog_internal.h@@ -76,8 +76,20 @@ typedef XLogLongPageHeaderData *XLogLongPageHeader;
#define XLP_LONG_HEADER 0x0002
/* This flag indicates backup blocks starting in this page are optional */
#define XLP_BKP_REMOVABLE 0x0004
+/*+ * This flag marks a record that replaces a missing contrecord.+ * When on WAL replay we expect a continuation record at the start of+ * a page that is not there, recovery ends but the checkpoint record+ * that follows is marked with this flag, which indicates WAL readers+ * that the incomplete record is to be skipped, and that WAL reading+ * is to be resumed here. This is useful for downstream consumers of+ * WAL which have already received (the first half of) the original+ * broken WAL record, such as via archive_command or physical streaming+ * replication, which we cannot "rewind".+ */+#define XLP_FIRST_IS_ABORTED_PARTIAL 0x0008
/* All defined flag bits in xlp_info (used for validity checking of header) */
-#define XLP_ALL_FLAGS 0x0007+#define XLP_ALL_FLAGS 0x000F
#define XLogPageHeaderSize(hdr) \
(((hdr)->xlp_info & XLP_LONG_HEADER) ? SizeOfXLogLongPHD : SizeOfXLogShortPHD)
@@ -249,6 +261,12 @@ typedef struct xl_restore_point
char rp_name[MAXFNAMELEN];
} xl_restore_point;
+/* Overwrite of prior contrecord */+typedef struct xl_overwrite_contrecord+{+ XLogRecPtr overwritten_lsn;+} xl_overwrite_contrecord;+
/* End of recovery mark, when we don't do an END_OF_RECOVERY checkpoint */
typedef struct xl_end_of_recovery
{
diff --git a/src/include/access/xlogreader.h b/src/include/access/xlogreader.hindex 21d200d3df..96e5eab1c9 100644--- a/src/include/access/xlogreader.h+++ b/src/include/access/xlogreader.h@@ -175,6 +175,9 @@ struct XLogReaderState
XLogRecPtr ReadRecPtr; /* start of last record read */
XLogRecPtr EndRecPtr; /* end+1 of last record read */
+ /* end+1 of incomplete record at end of WAL */+ XLogRecPtr abortedContrecordPtr;+
/* ----------------------------------------
* Decoded representation of current record
diff --git a/src/include/catalog/pg_control.h b/src/include/catalog/pg_control.hindex e3f48158ce..179efa9f4a 100644--- a/src/include/catalog/pg_control.h+++ b/src/include/catalog/pg_control.h@@ -76,6 +76,7 @@ typedef struct CheckPoint
#define XLOG_END_OF_RECOVERY 0x90
#define XLOG_FPI_FOR_HINT 0xA0
#define XLOG_FPI 0xB0
+#define XLOG_OVERWRITE_CONTRECORD 0xC0
/*
--
2.30.2
Reply instructions:
You may reply publicly to this message via plain-text email
using any one of the following methods:
* Reply to all the recipients using the --to and --cc options:
reply via email
To: pgsql-hackers@postgresql.org
Cc: alvherre@alvh.no-ip.org, horikyota.ntt@gmail.com, bossartn@amazon.com, andres@anarazel.de, masao.fujii@oss.nttdata.com, pgsql-hackers@lists.postgresql.org, mengjuan.cmj@alibaba-inc.com, Jakub.Wartak@tomtom.com
Subject: Re: prevent immature WAL streaming
In-Reply-To: <202109150132.73bzmi4bcppw@alvherre.pgsql>
* Save the following mbox file, import it into your mail client,
and reply-to-all from there: mbox
This inbox is served by DDX for PostgreSQL; see mirroring instructions
for how to clone and mirror all data and code used for this inbox