Repair sometimes-incorrect computation of StartUpID after a crash, per

example from Rao Kumar.  This is a very corner corner-case, requiring
a minimum of three closely-spaced database crashes and an unlucky
positioning of the second recovery's checkpoint record before you'd notice
any problem.  But the consequences are dire enough that it's a must-fix.

diff --git a/src/backend/access/transam/xlog.c b/src/backend/access/transam/xlog.c
index f68fc7053669851eb381646c6b87ea533804dd9a..7fa274922b0a8564120b56c1f3777f7dc8d38a64 100644 (file)
--- a/src/backend/access/transam/xlog.c
+++ b/src/backend/access/transam/xlog.c
@@ -7,7 +7,7 @@
  * Portions Copyright (c) 1996-2002, PostgreSQL Global Development Group
  * Portions Copyright (c) 1994, Regents of the University of California
  *
- * $Header: /cvsroot/pgsql/src/backend/access/transam/xlog.c,v 1.115 2003/05/10 18:01:31 tgl Exp $
+ * $Header: /cvsroot/pgsql/src/backend/access/transam/xlog.c,v 1.116 2003/05/22 14:39:28 tgl Exp $
  *
  *-------------------------------------------------------------------------
  */
@@ -2568,7 +2568,18 @@ StartupXLOG(void)
    ShmemVariableCache->nextOid = checkPoint.nextOid;
    ShmemVariableCache->oidCount = 0;
 
-   ThisStartUpID = checkPoint.ThisStartUpID;
+   /*
+    * If it was a shutdown checkpoint, then any following WAL entries were
+    * created under the next StartUpID; if it was a regular checkpoint then
+    * any following WAL entries were created under the same StartUpID.
+    * We must replay WAL entries using the same StartUpID they were created
+    * under, so temporarily adopt that SUI (see also xlog_redo()).
+    */
+   if (wasShutdown)
+       ThisStartUpID = checkPoint.ThisStartUpID + 1;
+   else
+       ThisStartUpID = checkPoint.ThisStartUpID;
+
    RedoRecPtr = XLogCtl->Insert.RedoRecPtr =
        XLogCtl->SavedRedoRecPtr = checkPoint.redo;
 
@@ -2672,55 +2683,21 @@ StartupXLOG(void)
    ControlFile->logSeg = openLogSeg + 1;
    Insert = &XLogCtl->Insert;
    Insert->PrevRecord = LastRec;
+   XLogCtl->xlblocks[0].xlogid = openLogId;
+   XLogCtl->xlblocks[0].xrecoff =
+       ((EndOfLog.xrecoff - 1) / BLCKSZ + 1) * BLCKSZ;
 
    /*
-    * If the next record will go to the new page then initialize for that
-    * one.
+    * Tricky point here: readBuf contains the *last* block that the
+    * LastRec record spans, not the one it starts in.  The last block
+    * is indeed the one we want to use.
     */
-   if ((BLCKSZ - EndOfLog.xrecoff % BLCKSZ) < SizeOfXLogRecord)
-       EndOfLog.xrecoff += (BLCKSZ - EndOfLog.xrecoff % BLCKSZ);
-   if (EndOfLog.xrecoff % BLCKSZ == 0)
-   {
-       XLogRecPtr  NewPageEndPtr;
-
-       NewPageEndPtr = EndOfLog;
-       if (NewPageEndPtr.xrecoff >= XLogFileSize)
-       {
-           /* crossing a logid boundary */
-           NewPageEndPtr.xlogid += 1;
-           NewPageEndPtr.xrecoff = BLCKSZ;
-       }
-       else
-           NewPageEndPtr.xrecoff += BLCKSZ;
-       XLogCtl->xlblocks[0] = NewPageEndPtr;
-       Insert->currpage->xlp_magic = XLOG_PAGE_MAGIC;
-       if (InRecovery)
-           Insert->currpage->xlp_sui = ThisStartUpID;
-       else
-           Insert->currpage->xlp_sui = ThisStartUpID + 1;
-       Insert->currpage->xlp_pageaddr.xlogid = NewPageEndPtr.xlogid;
-       Insert->currpage->xlp_pageaddr.xrecoff = NewPageEndPtr.xrecoff - BLCKSZ;
-       /* rest of buffer was zeroed in XLOGShmemInit */
-       Insert->currpos = (char *) Insert->currpage + SizeOfXLogPHD;
-   }
-   else
-   {
-       XLogCtl->xlblocks[0].xlogid = openLogId;
-       XLogCtl->xlblocks[0].xrecoff =
-           ((EndOfLog.xrecoff - 1) / BLCKSZ + 1) * BLCKSZ;
-
-       /*
-        * Tricky point here: readBuf contains the *last* block that the
-        * LastRec record spans, not the one it starts in.  The last block
-        * is indeed the one we want to use.
-        */
-       Assert(readOff == (XLogCtl->xlblocks[0].xrecoff - BLCKSZ) % XLogSegSize);
-       memcpy((char *) Insert->currpage, readBuf, BLCKSZ);
-       Insert->currpos = (char *) Insert->currpage +
-           (EndOfLog.xrecoff + BLCKSZ - XLogCtl->xlblocks[0].xrecoff);
-       /* Make sure rest of page is zero */
-       memset(Insert->currpos, 0, INSERT_FREESPACE(Insert));
-   }
+   Assert(readOff == (XLogCtl->xlblocks[0].xrecoff - BLCKSZ) % XLogSegSize);
+   memcpy((char *) Insert->currpage, readBuf, BLCKSZ);
+   Insert->currpos = (char *) Insert->currpage +
+       (EndOfLog.xrecoff + BLCKSZ - XLogCtl->xlblocks[0].xrecoff);
+   /* Make sure rest of page is zero */
+   MemSet(Insert->currpos, 0, INSERT_FREESPACE(Insert));
 
    LogwrtResult.Write = LogwrtResult.Flush = EndOfLog;
 
@@ -2774,9 +2751,22 @@ StartupXLOG(void)
        MyXactMadeXLogEntry = false;
        MyXactMadeTempRelUpdate = false;
 
+       /*
+        * At this point, ThisStartUpID is the largest SUI that we could
+        * find evidence for in the WAL entries.  But check it against
+        * pg_control's latest checkpoint, to make sure that we can't
+        * accidentally re-use an already-used SUI.
+        */
+       if (ThisStartUpID < ControlFile->checkPointCopy.ThisStartUpID)
+           ThisStartUpID = ControlFile->checkPointCopy.ThisStartUpID;
+
        /*
         * Perform a new checkpoint to update our recovery activity to disk.
         *
+        * Note that we write a shutdown checkpoint.  This is correct since
+        * the records following it will use SUI one more than what is shown
+        * in the checkpoint's ThisStartUpID.
+        *
         * In case we had to use the secondary checkpoint, make sure that
         * it will still be shown as the secondary checkpoint after this
         * CreateCheckPoint operation; we don't want the broken primary
@@ -2790,21 +2780,39 @@ StartupXLOG(void)
         */
        XLogCloseRelationCache();
    }
+   else
+   {
+       /*
+        * If we are not doing recovery, then we saw a checkpoint with nothing
+        * after it, and we can safely use StartUpID equal to one more than
+        * the checkpoint's SUI.  But just for paranoia's sake, check against
+        * pg_control too.
+        */
+       ThisStartUpID = checkPoint.ThisStartUpID;
+       if (ThisStartUpID < ControlFile->checkPointCopy.ThisStartUpID)
+           ThisStartUpID = ControlFile->checkPointCopy.ThisStartUpID;
+   }
 
    /*
     * Preallocate additional log files, if wanted.
     */
    PreallocXlogFiles(EndOfLog);
 
+   /*
+    * Advance StartUpID to one more than the highest value used previously.
+    */
+   ThisStartUpID++;
+   XLogCtl->ThisStartUpID = ThisStartUpID;
+
+   /*
+    * Okay, we're officially UP.
+    */
    InRecovery = false;
 
    ControlFile->state = DB_IN_PRODUCTION;
    ControlFile->time = time(NULL);
    UpdateControlFile();
 
-   ThisStartUpID++;
-   XLogCtl->ThisStartUpID = ThisStartUpID;
-
    /* Start up the commit log, too */
    StartupCLOG();
 
@@ -3000,7 +3008,7 @@ CreateCheckPoint(bool shutdown, bool force)
        UpdateControlFile();
    }
 
-   memset(&checkPoint, 0, sizeof(checkPoint));
+   MemSet(&checkPoint, 0, sizeof(checkPoint));
    checkPoint.ThisStartUpID = ThisStartUpID;
    checkPoint.time = time(NULL);
 
@@ -3259,6 +3267,8 @@ xlog_redo(XLogRecPtr lsn, XLogRecord *record)
        ShmemVariableCache->nextXid = checkPoint.nextXid;
        ShmemVariableCache->nextOid = checkPoint.nextOid;
        ShmemVariableCache->oidCount = 0;
+       /* Any later WAL records should be run with shutdown SUI plus 1 */
+       ThisStartUpID = checkPoint.ThisStartUpID + 1;
    }
    else if (info == XLOG_CHECKPOINT_ONLINE)
    {
@@ -3274,6 +3284,8 @@ xlog_redo(XLogRecPtr lsn, XLogRecord *record)
            ShmemVariableCache->nextOid = checkPoint.nextOid;
            ShmemVariableCache->oidCount = 0;
        }
+       /* Any later WAL records should be run with the then-active SUI */
+       ThisStartUpID = checkPoint.ThisStartUpID;
    }
 }