Skip to content

Commit 1d97d3d

Browse files
alvherrealvherre
committed
Fix WAL replay in presence of an incomplete record
Physical replication always ships WAL segment files to replicas once they are complete. This is a problem if one WAL record is split across a segment boundary and the primary server crashes before writing down the segment with the next portion of the WAL record: WAL writing after crash recovery would happily resume at the point where the broken record started, overwriting that record ... but any standby or backup may have already received a copy of that segment, and they are not rewinding. This causes standbys to stop following the primary after the latter crashes: LOG: invalid contrecord length 7262 at A8/D9FFFBC8 because the standby is still trying to read the continuation record (contrecord) for the original long WAL record, but it is not there and it will never be. A workaround is to stop the replica, delete the WAL file, and restart it -- at which point a fresh copy is brought over from the primary. But that's pretty labor intensive, and I bet many users would just give up and re-clone the standby instead. A fix for this problem was already attempted in commit 515e3d8, but it only addressed the case for the scenario of WAL archiving, so streaming replication would still be a problem (as well as other things such as taking a filesystem-level backup while the server is down after having crashed), and it had performance scalability problems too; so it had to be reverted. This commit fixes the problem using an approach suggested by Andres Freund, whereby the initial portion(s) of the split-up WAL record are kept, and a special type of WAL record is written where the contrecord was lost, so that WAL replay in the replica knows to skip the broken parts. With this approach, we can continue to stream/archive segment files as soon as they are complete, and replay of the broken records will proceed across the crash point without a hitch. Because a new type of WAL record is added, users should be careful to upgrade standbys first, primaries later. Otherwise they risk the standby being unable to start if the primary happens to write such a record. A new TAP test that exercises this is added, but the portability of it is yet to be seen. This has been wrong since the introduction of physical replication, so backpatch all the way back. In stable branches, keep the new XLogReaderState members at the end of the struct, to avoid an ABI break. Author: Álvaro Herrera <alvherre@alvh.no-ip.org> Reviewed-by: Kyotaro Horiguchi <horikyota.ntt@gmail.com> Reviewed-by: Nathan Bossart <bossartn@amazon.com> Discussion: https://postgr.es/m/202108232252.dh7uxf6oxwcy@alvherre.pgsql
1 parent 8cf4f71 commit 1d97d3d

File tree

8 files changed

+323
-7
lines changed

8 files changed

+323
-7
lines changed

‎src/backend/access/rmgrdesc/xlogdesc.c

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -140,6 +140,16 @@ xlog_desc(StringInfo buf, XLogReaderState *record)
140140
xlrec.ThisTimeLineID, xlrec.PrevTimeLineID,
141141
timestamptz_to_str(xlrec.end_time));
142142
}
143+
else if (info == XLOG_OVERWRITE_CONTRECORD)
144+
{
145+
xl_overwrite_contrecord xlrec;
146+
147+
memcpy(&xlrec, rec, sizeof(xl_overwrite_contrecord));
148+
appendStringInfo(buf, "lsn %X/%X; time %s",
149+
(uint32) (xlrec.overwritten_lsn >> 32),
150+
(uint32) xlrec.overwritten_lsn,
151+
timestamptz_to_str(xlrec.overwrite_time));
152+
}
143153
}
144154

145155
const char *
@@ -179,6 +189,9 @@ xlog_identify(uint8 info)
179189
case XLOG_END_OF_RECOVERY:
180190
id = "END_OF_RECOVERY";
181191
break;
192+
case XLOG_OVERWRITE_CONTRECORD:
193+
id = "OVERWRITE_CONTRECORD";
194+
break;
182195
case XLOG_FPI:
183196
id = "FPI";
184197
break;

‎src/backend/access/transam/xlog.c

Lines changed: 152 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -212,6 +212,15 @@ static XLogRecPtr LastRec;
212212
static XLogRecPtr flushedUpto = 0;
213213
static TimeLineID receiveTLI = 0;
214214

215+
/*
216+
* abortedRecPtr is the start pointer of a broken record at end of WAL when
217+
* recovery completes; missingContrecPtr is the location of the first
218+
* contrecord that went missing. See CreateOverwriteContrecordRecord for
219+
* details.
220+
*/
221+
static XLogRecPtr abortedRecPtr;
222+
static XLogRecPtr missingContrecPtr;
223+
215224
/*
216225
* During recovery, lastFullPageWrites keeps track of full_page_writes that
217226
* the replayed WAL records indicate. It's initialized with full_page_writes
@@ -894,8 +903,11 @@ static void CheckRequiredParameterValues(void);
894903
static void XLogReportParameters(void);
895904
static void checkTimeLineSwitch(XLogRecPtr lsn, TimeLineID newTLI,
896905
TimeLineID prevTLI);
906+
static void VerifyOverwriteContrecord(xl_overwrite_contrecord *xlrec,
907+
XLogReaderState *state);
897908
static void LocalSetXLogInsertAllowed(void);
898909
static void CreateEndOfRecoveryRecord(void);
910+
static XLogRecPtr CreateOverwriteContrecordRecord(XLogRecPtr aborted_lsn);
899911
static void CheckPointGuts(XLogRecPtr checkPointRedo, int flags);
900912
static void KeepLogSeg(XLogRecPtr recptr, XLogSegNo *logSegNo);
901913
static XLogRecPtr XLogGetReplicationSlotMinimumLSN(void);
@@ -2246,6 +2258,18 @@ AdvanceXLInsertBuffer(XLogRecPtr upto, bool opportunistic)
22462258
if (!Insert->forcePageWrites)
22472259
NewPage->xlp_info |= XLP_BKP_REMOVABLE;
22482260

2261+
/*
2262+
* If a record was found to be broken at the end of recovery, and
2263+
* we're going to write on the page where its first contrecord was
2264+
* lost, set the XLP_FIRST_IS_OVERWRITE_CONTRECORD flag on the page
2265+
* header. See CreateOverwriteContrecordRecord().
2266+
*/
2267+
if (missingContrecPtr == NewPageBeginPtr)
2268+
{
2269+
NewPage->xlp_info |= XLP_FIRST_IS_OVERWRITE_CONTRECORD;
2270+
missingContrecPtr = InvalidXLogRecPtr;
2271+
}
2272+
22492273
/*
22502274
* If first page of an XLOG segment file, make it a long header.
22512275
*/
@@ -4346,6 +4370,19 @@ ReadRecord(XLogReaderState *xlogreader, int emode,
43464370
EndRecPtr = xlogreader->EndRecPtr;
43474371
if (record == NULL)
43484372
{
4373+
/*
4374+
* When not in standby mode we find that WAL ends in an incomplete
4375+
* record, keep track of that record. After recovery is done,
4376+
* we'll write a record to indicate downstream WAL readers that
4377+
* that portion is to be ignored.
4378+
*/
4379+
if (!StandbyMode &&
4380+
!XLogRecPtrIsInvalid(xlogreader->abortedRecPtr))
4381+
{
4382+
abortedRecPtr = xlogreader->abortedRecPtr;
4383+
missingContrecPtr = xlogreader->missingContrecPtr;
4384+
}
4385+
43494386
if (readFile >= 0)
43504387
{
43514388
close(readFile);
@@ -6877,6 +6914,12 @@ StartupXLOG(void)
68776914
InRecovery = true;
68786915
}
68796916

6917+
/*
6918+
* Start recovery assuming that the final record isn't lost.
6919+
*/
6920+
abortedRecPtr = InvalidXLogRecPtr;
6921+
missingContrecPtr = InvalidXLogRecPtr;
6922+
68806923
/* REDO */
68816924
if (InRecovery)
68826925
{
@@ -7469,8 +7512,9 @@ StartupXLOG(void)
74697512

74707513
/*
74717514
* Kill WAL receiver, if it's still running, before we continue to write
7472-
* the startup checkpoint record. It will trump over the checkpoint and
7473-
* subsequent records if it's still alive when we start writing WAL.
7515+
* the startup checkpoint and aborted-contrecord records. It will trump
7516+
* over these records and subsequent ones if it's still alive when we
7517+
* start writing WAL.
74747518
*/
74757519
ShutdownWalRcv();
74767520

@@ -7503,8 +7547,12 @@ StartupXLOG(void)
75037547
StandbyMode = false;
75047548

75057549
/*
7506-
* Re-fetch the last valid or last applied record, so we can identify the
7507-
* exact endpoint of what we consider the valid portion of WAL.
7550+
* Determine where to start writing WAL next.
7551+
*
7552+
* When recovery ended in an incomplete record, write a WAL record about
7553+
* that and continue after it. In all other cases, re-fetch the last
7554+
* valid or last applied record, so we can identify the exact endpoint of
7555+
* what we consider the valid portion of WAL.
75087556
*/
75097557
XLogBeginRead(xlogreader, LastRec);
75107558
record = ReadRecord(xlogreader, PANIC, false);
@@ -7654,6 +7702,18 @@ StartupXLOG(void)
76547702
XLogCtl->ThisTimeLineID = ThisTimeLineID;
76557703
XLogCtl->PrevTimeLineID = PrevTimeLineID;
76567704

7705+
/*
7706+
* Actually, if WAL ended in an incomplete record, skip the parts that
7707+
* made it through and start writing after the portion that persisted.
7708+
* (It's critical to first write an OVERWRITE_CONTRECORD message, which
7709+
* we'll do as soon as we're open for writing new WAL.)
7710+
*/
7711+
if (!XLogRecPtrIsInvalid(missingContrecPtr))
7712+
{
7713+
Assert(!XLogRecPtrIsInvalid(abortedRecPtr));
7714+
EndOfLog = missingContrecPtr;
7715+
}
7716+
76577717
/*
76587718
* Prepare to write WAL starting at EndOfLog location, and init xlog
76597719
* buffer cache using the block containing the last record from the
@@ -7706,13 +7766,23 @@ StartupXLOG(void)
77067766
XLogCtl->LogwrtRqst.Write = EndOfLog;
77077767
XLogCtl->LogwrtRqst.Flush = EndOfLog;
77087768

7769+
LocalSetXLogInsertAllowed();
7770+
7771+
/* If necessary, write overwrite-contrecord before doing anything else */
7772+
if (!XLogRecPtrIsInvalid(abortedRecPtr))
7773+
{
7774+
Assert(!XLogRecPtrIsInvalid(missingContrecPtr));
7775+
CreateOverwriteContrecordRecord(abortedRecPtr);
7776+
abortedRecPtr = InvalidXLogRecPtr;
7777+
missingContrecPtr = InvalidXLogRecPtr;
7778+
}
7779+
77097780
/*
77107781
* Update full_page_writes in shared memory and write an XLOG_FPW_CHANGE
77117782
* record before resource manager writes cleanup WAL records or checkpoint
77127783
* record is written.
77137784
*/
77147785
Insert->fullPageWrites = lastFullPageWrites;
7715-
LocalSetXLogInsertAllowed();
77167786
UpdateFullPageWrites();
77177787
LocalXLogInsertAllowed = -1;
77187788

@@ -9158,6 +9228,53 @@ CreateEndOfRecoveryRecord(void)
91589228
LocalXLogInsertAllowed = -1; /* return to "check" state */
91599229
}
91609230

9231+
/*
9232+
* Write an OVERWRITE_CONTRECORD message.
9233+
*
9234+
* When on WAL replay we expect a continuation record at the start of a page
9235+
* that is not there, recovery ends and WAL writing resumes at that point.
9236+
* But it's wrong to resume writing new WAL back at the start of the record
9237+
* that was broken, because downstream consumers of that WAL (physical
9238+
* replicas) are not prepared to "rewind". So the first action after
9239+
* finishing replay of all valid WAL must be to write a record of this type
9240+
* at the point where the contrecord was missing; to support xlogreader
9241+
* detecting the special case, XLP_FIRST_IS_OVERWRITE_CONTRECORD is also added
9242+
* to the page header where the record occurs. xlogreader has an ad-hoc
9243+
* mechanism to report metadata about the broken record, which is what we
9244+
* use here.
9245+
*
9246+
* At replay time, XLP_FIRST_IS_OVERWRITE_CONTRECORD instructs xlogreader to
9247+
* skip the record it was reading, and pass back the LSN of the skipped
9248+
* record, so that its caller can verify (on "replay" of that record) that the
9249+
* XLOG_OVERWRITE_CONTRECORD matches what was effectively overwritten.
9250+
*/
9251+
static XLogRecPtr
9252+
CreateOverwriteContrecordRecord(XLogRecPtr aborted_lsn)
9253+
{
9254+
xl_overwrite_contrecord xlrec;
9255+
XLogRecPtr recptr;
9256+
9257+
/* sanity check */
9258+
if (!RecoveryInProgress())
9259+
elog(ERROR, "can only be used at end of recovery");
9260+
9261+
xlrec.overwritten_lsn = aborted_lsn;
9262+
xlrec.overwrite_time = GetCurrentTimestamp();
9263+
9264+
START_CRIT_SECTION();
9265+
9266+
XLogBeginInsert();
9267+
XLogRegisterData((char *) &xlrec, sizeof(xl_overwrite_contrecord));
9268+
9269+
recptr = XLogInsert(RM_XLOG_ID, XLOG_OVERWRITE_CONTRECORD);
9270+
9271+
XLogFlush(recptr);
9272+
9273+
END_CRIT_SECTION();
9274+
9275+
return recptr;
9276+
}
9277+
91619278
/*
91629279
* Flush all data in shared memory to disk, and fsync
91639280
*
@@ -10080,6 +10197,13 @@ xlog_redo(XLogReaderState *record)
1008010197

1008110198
RecoveryRestartPoint(&checkPoint);
1008210199
}
10200+
else if (info == XLOG_OVERWRITE_CONTRECORD)
10201+
{
10202+
xl_overwrite_contrecord xlrec;
10203+
10204+
memcpy(&xlrec, XLogRecGetData(record), sizeof(xl_overwrite_contrecord));
10205+
VerifyOverwriteContrecord(&xlrec, record);
10206+
}
1008310207
else if (info == XLOG_END_OF_RECOVERY)
1008410208
{
1008510209
xl_end_of_recovery xlrec;
@@ -10240,6 +10364,29 @@ xlog_redo(XLogReaderState *record)
1024010364
}
1024110365
}
1024210366

10367+
/*
10368+
* Verify the payload of a XLOG_OVERWRITE_CONTRECORD record.
10369+
*/
10370+
static void
10371+
VerifyOverwriteContrecord(xl_overwrite_contrecord *xlrec, XLogReaderState *state)
10372+
{
10373+
if (xlrec->overwritten_lsn != state->overwrittenRecPtr)
10374+
elog(FATAL, "mismatching overwritten LSN %X/%X -> %X/%X",
10375+
(uint32) (xlrec->overwritten_lsn >> 32),
10376+
(uint32) xlrec->overwritten_lsn,
10377+
(uint32) (state->overwrittenRecPtr >> 32),
10378+
(uint32) state->overwrittenRecPtr);
10379+
10380+
ereport(LOG,
10381+
(errmsg("sucessfully skipped missing contrecord at %X/%X, overwritten at %s",
10382+
(uint32) (xlrec->overwritten_lsn >> 32),
10383+
(uint32) xlrec->overwritten_lsn,
10384+
timestamptz_to_str(xlrec->overwrite_time))));
10385+
10386+
/* Verifying the record should only happen once */
10387+
state->overwrittenRecPtr = InvalidXLogRecPtr;
10388+
}
10389+
1024310390
#ifdef WAL_DEBUG
1024410391

1024510392
static void

‎src/backend/access/transam/xlogreader.c

Lines changed: 39 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -275,6 +275,7 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
275275
total_len;
276276
uint32 targetRecOff;
277277
uint32 pageHeaderSize;
278+
bool assembled;
278279
bool gotheader;
279280
int readOff;
280281

@@ -290,6 +291,8 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
290291
state->errormsg_buf[0] = '\0';
291292

292293
ResetDecoder(state);
294+
state->abortedRecPtr = InvalidXLogRecPtr;
295+
state->missingContrecPtr = InvalidXLogRecPtr;
293296

294297
RecPtr = state->EndRecPtr;
295298

@@ -316,7 +319,9 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
316319
randAccess = true;
317320
}
318321

322+
restart:
319323
state->currRecPtr = RecPtr;
324+
assembled = false;
320325

321326
targetPagePtr = RecPtr - (RecPtr % XLOG_BLCKSZ);
322327
targetRecOff = RecPtr % XLOG_BLCKSZ;
@@ -412,6 +417,8 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
412417
char *buffer;
413418
uint32 gotlen;
414419

420+
assembled = true;
421+
415422
/*
416423
* Enlarge readRecordBuf as needed.
417424
*/
@@ -446,8 +453,25 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
446453

447454
Assert(SizeOfXLogShortPHD <= readOff);
448455

449-
/* Check that the continuation on next page looks valid */
450456
pageHeader = (XLogPageHeader) state->readBuf;
457+
458+
/*
459+
* If we were expecting a continuation record and got an
460+
* "overwrite contrecord" flag, that means the continuation record
461+
* was overwritten with a different record. Restart the read by
462+
* assuming the address to read is the location where we found
463+
* this flag; but keep track of the LSN of the record we were
464+
* reading, for later verification.
465+
*/
466+
if (pageHeader->xlp_info & XLP_FIRST_IS_OVERWRITE_CONTRECORD)
467+
{
468+
state->overwrittenRecPtr = state->currRecPtr;
469+
ResetDecoder(state);
470+
RecPtr = targetPagePtr;
471+
goto restart;
472+
}
473+
474+
/* Check that the continuation on next page looks valid */
451475
if (!(pageHeader->xlp_info & XLP_FIRST_IS_CONTRECORD))
452476
{
453477
report_invalid_record(state,
@@ -548,6 +572,20 @@ XLogReadRecord(XLogReaderState *state, char **errormsg)
548572
return NULL;
549573

550574
err:
575+
if (assembled)
576+
{
577+
/*
578+
* We get here when a record that spans multiple pages needs to be
579+
* assembled, but something went wrong -- perhaps a contrecord piece
580+
* was lost. If caller is WAL replay, it will know where the aborted
581+
* record was and where to direct followup WAL to be written, marking
582+
* the next piece with XLP_FIRST_IS_OVERWRITE_CONTRECORD, which will
583+
* in turn signal downstream WAL consumers that the broken WAL record
584+
* is to be ignored.
585+
*/
586+
state->abortedRecPtr = RecPtr;
587+
state->missingContrecPtr = targetPagePtr;
588+
}
551589

552590
/*
553591
* Invalidate the read state. We might read from a different source after

‎src/include/access/xlog_internal.h

Lines changed: 10 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -79,8 +79,10 @@ typedef XLogLongPageHeaderData *XLogLongPageHeader;
7979
#define XLP_LONG_HEADER 0x0002
8080
/* This flag indicates backup blocks starting in this page are optional */
8181
#define XLP_BKP_REMOVABLE 0x0004
82+
/* Replaces a missing contrecord; see CreateOverwriteContrecordRecord */
83+
#define XLP_FIRST_IS_OVERWRITE_CONTRECORD 0x0008
8284
/* All defined flag bits in xlp_info (used for validity checking of header) */
83-
#define XLP_ALL_FLAGS 0x0007
85+
#define XLP_ALL_FLAGS 0x000F
8486

8587
#define XLogPageHeaderSize(hdr) \
8688
(((hdr)->xlp_info & XLP_LONG_HEADER) ? SizeOfXLogLongPHD : SizeOfXLogShortPHD)
@@ -252,6 +254,13 @@ typedef struct xl_restore_point
252254
char rp_name[MAXFNAMELEN];
253255
} xl_restore_point;
254256

257+
/* Overwrite of prior contrecord */
258+
typedef struct xl_overwrite_contrecord
259+
{
260+
XLogRecPtr overwritten_lsn;
261+
TimestampTz overwrite_time;
262+
} xl_overwrite_contrecord;
263+
255264
/* End of recovery mark, when we don't do an END_OF_RECOVERY checkpoint */
256265
typedef struct xl_end_of_recovery
257266
{

0 commit comments

Comments
 (0)