[FIXED] Desync after partial catchup from old leader

Signed-off-by: Maurice van Veen <[email protected]>

Author: MauriceVanVeen

server/jetstream_cluster.go

@@ -8618,7 +8618,28 @@ RETRY:
 				// Check for eof signaling.
 				if len(msg) == 0 {
 					msgsQ.recycle(&mrecs)
-					return nil
+
+					// Sanity check that we've received all data expected by the snapshot.
+					mset.mu.RLock()
+					lseq := mset.lseq
+					mset.mu.RUnlock()
+					if lseq >= snap.LastSeq {
+						return nil
+					}
+
+					// Make sure we do not spin and make things worse.
+					const minRetryWait = 2 * time.Second
+					elapsed := time.Since(reqSendTime)
+					if elapsed < minRetryWait {
+						select {
+						case <-s.quitCh:
+							return ErrServerNotRunning
+						case <-qch:
+							return errCatchupStreamStopped
+						case <-time.After(minRetryWait - elapsed):
+						}
+					}
+					goto RETRY
 				}
 				if _, err := mset.processCatchupMsg(msg); err == nil {
 					if mrec.reply != _EMPTY_ {
@@ -9122,6 +9143,15 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 		// In the latter case the request expects us to have more. Just continue and value availability here.
 		// This should only be possible if the logs have already desynced, and we shouldn't have become leader
 		// in the first place. Not much we can do here in this (hypothetical) scenario.
+
+		// Do another quick sanity check that we actually have enough data to satisfy the request.
+		// If not, let's step down and hope a new leader can correct this.
+		if state.LastSeq < last {
+			s.Warnf("Catchup for stream '%s > %s' skipped, requested sequence %d was larger than current state: %+v",
+				mset.account(), mset.name(), seq, state)
+			node.StepDown()
+			return
+		}
 	}
 
 	mset.setCatchupPeer(sreq.Peer, last-seq)
@@ -9241,7 +9271,7 @@ func (mset *stream) runCatchup(sendSubject string, sreq *streamSyncRequest) {
 						// The snapshot has a larger last sequence then we have. This could be due to a truncation
 						// when trying to recover after corruption, still not 100% sure. Could be off by 1 too somehow,
 						// but tested a ton of those with no success.
-						s.Warnf("Catchup for stream '%s > %s' completed, but requested sequence %d was larger then current state: %+v",
+						s.Warnf("Catchup for stream '%s > %s' completed, but requested sequence %d was larger than current state: %+v",
 							mset.account(), mset.name(), seq, state)
 						// Try our best to redo our invalidated snapshot as well.
 						if n := mset.raftNode(); n != nil {

server/jetstream_cluster_4_test.go

@@ -3528,8 +3528,118 @@ func TestJetStreamClusterConsumerDesyncAfterErrorDuringStreamCatchup(t *testing.
 	c.waitOnConsumerLeader(globalAccountName, "TEST", "CONSUMER")
 
 	// Outdated server must NOT become the leader.
-	newConsummerLeaderServer := c.consumerLeader(globalAccountName, "TEST", "CONSUMER")
-	require_Equal(t, newConsummerLeaderServer.Name(), clusterResetServerName)
+	newConsumerLeaderServer := c.consumerLeader(globalAccountName, "TEST", "CONSUMER")
+	require_Equal(t, newConsumerLeaderServer.Name(), clusterResetServerName)
+}
+
+func TestJetStreamClusterDesyncAfterEofFromOldStreamLeader(t *testing.T) {
+	test := func(t *testing.T, eof bool) {
+		c := createJetStreamClusterExplicit(t, "R5S", 5)
+		defer c.shutdown()
+
+		cs := c.randomServer()
+		nc, js := jsClientConnect(t, cs)
+		defer nc.Close()
+
+		_, err := js.AddStream(&nats.StreamConfig{
+			Name:     "TEST",
+			Subjects: []string{"foo"},
+			Replicas: 5,
+		})
+		require_NoError(t, err)
+
+		sl := c.streamLeader(globalAccountName, "TEST")
+		var rs *Server
+		var catchup *Server
+		for _, s := range c.servers {
+			if s != sl && s != cs {
+				if rs == nil {
+					rs = s
+				} else {
+					catchup = s
+					break
+				}
+			}
+		}
+
+		// Shutdown server that needs to catch up, so it gets a snapshot from the leader after restart.
+		catchup.Shutdown()
+
+		// One message is received and applied by all replicas.
+		_, err = js.Publish("foo", nil)
+		require_NoError(t, err)
+		checkFor(t, 2*time.Second, 200*time.Millisecond, func() error {
+			return checkState(t, c, globalAccountName, "TEST")
+		})
+
+		// Disable Raft and start cluster subs for server, simulating an old leader with an outdated log.
+		acc, err := rs.lookupAccount(globalAccountName)
+		require_NoError(t, err)
+		mset, err := acc.lookupStream("TEST")
+		require_NoError(t, err)
+		mset.startClusterSubs()
+		rn := mset.raftNode()
+		rn.Stop()
+		rn.WaitForStop()
+
+		// Temporarily disable cluster subs for this test.
+		// Normally due to multiple cluster subs responses will interleave, but this is simpler for this test.
+		acc, err = sl.lookupAccount(globalAccountName)
+		require_NoError(t, err)
+		mset, err = acc.lookupStream("TEST")
+		require_NoError(t, err)
+		mset.stopClusterSubs()
+
+		// Publish another message that the old leader will not get.
+		_, err = js.Publish("foo", nil)
+		require_NoError(t, err)
+		require_NoError(t, sl.JetStreamSnapshotStream(globalAccountName, "TEST"))
+
+		sa := sl.getJetStream().streamAssignment(globalAccountName, "TEST")
+		require_NotNil(t, sa)
+
+		// Send EOF immediately to requesting server, otherwise the server needs to time out and retry.
+		if eof {
+			snc, err := nats.Connect(rs.ClientURL(), nats.UserInfo("admin", "s3cr3t!"))
+			require_NoError(t, err)
+			defer snc.Close()
+
+			sub, err := snc.Subscribe(sa.Sync, func(msg *nats.Msg) {
+				// EOF
+				rs.sendInternalMsgLocked(msg.Reply, _EMPTY_, nil, nil)
+			})
+			require_NoError(t, err)
+			defer sub.Drain()
+		}
+
+		// Restart server so it starts catching up.
+		catchup = c.restartServer(catchup)
+
+		// Wait for server to start catching up.
+		// This shouldn't be a problem. The server should retry catchup, recognizing it wasn't caught up fully.
+		checkFor(t, 2*time.Second, 200*time.Millisecond, func() error {
+			if a, err := catchup.lookupAccount(globalAccountName); err != nil {
+				return err
+			} else if m, err := a.lookupStream("TEST"); err != nil {
+				return err
+			} else if !m.isCatchingUp() {
+				return errors.New("stream not catching up")
+			}
+			return nil
+		})
+
+		// Stop old leader, and re-enable cluster subs on proper leader.
+		rs.Shutdown()
+		mset.startClusterSubs()
+
+		// Server should automatically restart catchup and get the missing data.
+		checkFor(t, 10*time.Second, 200*time.Millisecond, func() error {
+			return checkState(t, c, globalAccountName, "TEST")
+		})
+	}
+
+	t.Run("eof", func(t *testing.T) { test(t, true) })
+	t.Run("retry", func(t *testing.T) { test(t, false) })
 }
 
 func TestJetStreamClusterReservedResourcesAccountingAfterClusterReset(t *testing.T) {