Start Hot Standby faster when initial snapshot is incomplete.

If the initial snapshot had overflowed then we can start whenever
the latest snapshot is empty, not overflowed or as we did already,
start when the xmin on primary was higher than xmax of our starting
snapshot, which proves we have full snapshot data.

Bug report by Chris Redekop

diff --git a/src/backend/storage/ipc/procarray.c b/src/backend/storage/ipc/procarray.c
index 89ec7f046196af9cbad247da08c3ea5c0474e676..292e330367e74328e336d41ce71227fead461e7a 100644 (file)
--- a/src/backend/storage/ipc/procarray.c
+++ b/src/backend/storage/ipc/procarray.c
@@ -475,9 +475,10 @@ ProcArrayApplyRecoveryInfo(RunningTransactions running)
        return;
 
    /*
-    * If our initial RunningTransactionsData had an overflowed snapshot then we knew
-    * we were missing some subxids from our snapshot. We can use this data as
-    * an initial snapshot, but we cannot yet mark it valid. We know that the
+    * If our initial RunningTransactionsData had an overflowed snapshot then
+    * we knew we were missing some subxids from our snapshot. If we continue
+    * to see overflowed snapshots then we might never be able to start up,
+    * so we make another test to see if our snapshot is now valid. We know
     * missing subxids are equal to or earlier than nextXid. After we
     * initialise we continue to apply changes during recovery, so once the
     * oldestRunningXid is later than the nextXid from the initial snapshot we
@@ -486,21 +487,31 @@ ProcArrayApplyRecoveryInfo(RunningTransactions running)
     */
    if (standbyState == STANDBY_SNAPSHOT_PENDING)
    {
-       if (TransactionIdPrecedes(standbySnapshotPendingXmin,
-                                 running->oldestRunningXid))
+       /*
+        * If the snapshot isn't overflowed or if its empty we can
+        * reset our pending state and use this snapshot instead.
+        */
+       if (!running->subxid_overflow || running->xcnt == 0)
        {
-           standbyState = STANDBY_SNAPSHOT_READY;
-           elog(trace_recovery(DEBUG2),
-                "running xact data now proven complete");
-           elog(trace_recovery(DEBUG2),
-                "recovery snapshots are now enabled");
+           standbyState = STANDBY_INITIALIZED;
        }
        else
-           elog(trace_recovery(DEBUG2),
-                "recovery snapshot waiting for %u oldest active xid on standby is %u",
-                standbySnapshotPendingXmin,
-                running->oldestRunningXid);
-       return;
+       {
+           if (TransactionIdPrecedes(standbySnapshotPendingXmin,
+                                     running->oldestRunningXid))
+           {
+               standbyState = STANDBY_SNAPSHOT_READY;
+               elog(trace_recovery(DEBUG1),
+                    "recovery snapshots are now enabled");
+           }
+           else
+               elog(trace_recovery(DEBUG1),
+                    "recovery snapshot waiting for non-overflowed snapshot or "
+                    "until oldest active xid on standby is at least %u (now %u)",
+                    standbySnapshotPendingXmin,
+                    running->oldestRunningXid);
+           return;
+       }
    }
 
    Assert(standbyState == STANDBY_INITIALIZED);
@@ -606,7 +617,6 @@ ProcArrayApplyRecoveryInfo(RunningTransactions running)
        standbyState = STANDBY_SNAPSHOT_READY;
 
        standbySnapshotPendingXmin = InvalidTransactionId;
-       procArray->lastOverflowedXid = InvalidTransactionId;
    }
 
    /*
@@ -630,13 +640,15 @@ ProcArrayApplyRecoveryInfo(RunningTransactions running)
 
    LWLockRelease(ProcArrayLock);
 
-   elog(trace_recovery(DEBUG2), "running transaction data initialized");
    KnownAssignedXidsDisplay(trace_recovery(DEBUG3));
    if (standbyState == STANDBY_SNAPSHOT_READY)
-       elog(trace_recovery(DEBUG2), "recovery snapshots are now enabled");
+       elog(trace_recovery(DEBUG1), "recovery snapshots are now enabled");
    else
-       ereport(LOG,
-               (errmsg("consistent state delayed because recovery snapshot incomplete")));
+       elog(trace_recovery(DEBUG1),
+            "recovery snapshot waiting for non-overflowed snapshot or "
+            "until oldest active xid on standby is at least %u (now %u)",
+            standbySnapshotPendingXmin,
+            running->oldestRunningXid);
 }
 
 /*