[FIXED] Stream desync after out-of-order SkipMsg (#7400)

Author: neilalexander

server/filestore.go

@@ -4541,19 +4541,27 @@ func (mb *msgBlock) skipMsg(seq uint64, now int64) {
 }
 
 // SkipMsg will use the next sequence number but not store anything.
-func (fs *fileStore) SkipMsg() uint64 {
+func (fs *fileStore) SkipMsg(seq uint64) (uint64, error) {
+	// Grab time.
+	now := ats.AccessTime()
+
 	fs.mu.Lock()
 	defer fs.mu.Unlock()
 
+	// Check sequence matches our last sequence.
+	if seq != fs.state.LastSeq+1 {
+		if seq > 0 {
+			return 0, ErrSequenceMismatch
+		}
+		seq = fs.state.LastSeq + 1
+	}
+
 	// Grab our current last message block.
 	mb, err := fs.checkLastBlock(emptyRecordLen)
 	if err != nil {
-		return 0
+		return 0, err
 	}
 
-	// Grab time and last seq.
-	now, seq := ats.AccessTime(), fs.state.LastSeq+1
-
 	// Write skip msg.
 	mb.skipMsg(seq, now)
 
@@ -4568,7 +4576,7 @@ func (fs *fileStore) SkipMsg() uint64 {
 	// Mark as dirty for stream state.
 	fs.dirty++
 
-	return seq
+	return seq, nil
 }
 
 // Skip multiple msgs. We will determine if we can fit into current lmb or we need to create a new block.

server/filestore_test.go

@@ -345,7 +345,7 @@ func TestFileStoreSkipMsg(t *testing.T) {
 
 		numSkips := 10
 		for i := 0; i < numSkips; i++ {
-			fs.SkipMsg()
+			fs.SkipMsg(0)
 		}
 		state := fs.State()
 		if state.Msgs != 0 {
@@ -356,10 +356,10 @@ func TestFileStoreSkipMsg(t *testing.T) {
 		}
 
 		fs.StoreMsg("zzz", nil, []byte("Hello World!"), 0)
-		fs.SkipMsg()
-		fs.SkipMsg()
+		fs.SkipMsg(0)
+		fs.SkipMsg(0)
 		fs.StoreMsg("zzz", nil, []byte("Hello World!"), 0)
-		fs.SkipMsg()
+		fs.SkipMsg(0)
 
 		state = fs.State()
 		if state.Msgs != 2 {
@@ -393,7 +393,7 @@ func TestFileStoreSkipMsg(t *testing.T) {
 			t.Fatalf("Message did not match")
 		}
 
-		fs.SkipMsg()
+		fs.SkipMsg(0)
 		nseq, _, err := fs.StoreMsg("AAA", nil, []byte("Skip?"), 0)
 		if err != nil {
 			t.Fatalf("Unexpected error looking up seq 11: %v", err)
@@ -5080,7 +5080,7 @@ func TestFileStoreSkipMsgAndNumBlocks(t *testing.T) {
 
 	fs.StoreMsg(subj, nil, msg, 0)
 	for i := 0; i < numMsgs; i++ {
-		fs.SkipMsg()
+		fs.SkipMsg(0)
 	}
 	fs.StoreMsg(subj, nil, msg, 0)
 	require_Equal(t, fs.numMsgBlocks(), 3)
@@ -6824,7 +6824,7 @@ func TestFileStoreEraseMsgWithDbitSlots(t *testing.T) {
 
 	fs.StoreMsg("foo", nil, []byte("abd"), 0)
 	for i := 0; i < 10; i++ {
-		fs.SkipMsg()
+		fs.SkipMsg(0)
 	}
 	fs.StoreMsg("foo", nil, []byte("abd"), 0)
 	// Now grab that first block and compact away the skips which will
@@ -6853,7 +6853,7 @@ func TestFileStoreEraseMsgWithAllTrailingDbitSlots(t *testing.T) {
 	fs.StoreMsg("foo", nil, []byte("abcdefg"), 0)
 
 	for i := 0; i < 10; i++ {
-		fs.SkipMsg()
+		fs.SkipMsg(0)
 	}
 	// Now grab that first block and compact away the skips which will
 	// introduce dbits into our idx.
@@ -7200,7 +7200,7 @@ func TestFileStoreReloadAndLoseLastSequence(t *testing.T) {
 	defer fs.Stop()
 
 	for i := 0; i < 22; i++ {
-		fs.SkipMsg()
+		fs.SkipMsg(0)
 	}
 
 	// Restart 5 times.
@@ -9053,7 +9053,7 @@ func TestFileStoreLeftoverSkipMsgInDmap(t *testing.T) {
 	}
 
 	// Only skip a message.
-	fs.SkipMsg()
+	fs.SkipMsg(0)
 
 	// Confirm state.
 	state := fs.State()

server/jetstream_cluster.go

@@ -3630,7 +3630,7 @@ func (js *jetStream) applyStreamMsgOp(mset *stream, op entryOp, mbuf []byte, isR
 	// Messages to be skipped have no subject or timestamp or msg or hdr.
 	if subject == _EMPTY_ && ts == 0 && len(msg) == 0 && len(hdr) == 0 {
 		// Skip and update our lseq.
-		last := mset.store.SkipMsg()
+		last, _ := mset.store.SkipMsg(0)
 		if needLock {
 			mset.mu.Lock()
 		}
@@ -9478,7 +9478,7 @@ func (mset *stream) processCatchupMsg(msg []byte) (uint64, error) {
 	// Messages to be skipped have no subject or timestamp.
 	// TODO(dlc) - formalize with skipMsgOp
 	if subj == _EMPTY_ && ts == 0 {
-		if lseq := mset.store.SkipMsg(); lseq != seq {
+		if _, err = mset.store.SkipMsg(seq); err != nil {
 			return 0, errCatchupWrongSeqForSkip
 		}
 	} else if err := mset.store.StoreRawMsg(subj, hdr, msg, seq, ts, ttl); err != nil {

server/jetstream_cluster_1_test.go

@@ -10330,6 +10330,71 @@ func TestJetStreamClusterDeleteMsgEOF(t *testing.T) {
 	}
 }
 
+func TestJetStreamClusterCatchupSkipMsgDesync(t *testing.T) {
+	for _, storage := range []nats.StorageType{nats.FileStorage, nats.MemoryStorage} {
+		t.Run(storage.String(), func(t *testing.T) {
+			c := createJetStreamClusterExplicit(t, "R3S", 3)
+			defer c.shutdown()
+
+			nc, js := jsClientConnect(t, c.randomServer())
+			defer nc.Close()
+
+			_, err := js.AddStream(&nats.StreamConfig{
+				Name:     "TEST",
+				Subjects: []string{"foo"},
+				Storage:  storage,
+				Replicas: 3,
+			})
+			require_NoError(t, err)
+			checkFor(t, 2*time.Second, 200*time.Millisecond, func() error {
+				return checkState(t, c, globalAccountName, "TEST")
+			})
+
+			rs := c.randomNonStreamLeader(globalAccountName, "TEST")
+			mset, err := rs.globalAccount().lookupStream("TEST")
+			require_NoError(t, err)
+			sa := mset.streamAssignment()
+
+			// Make sure this server can't become the leader.
+			n := mset.raftNode().(*raft)
+			n.SetObserver(true)
+
+			sysNc, err := nats.Connect(rs.ClientURL(), nats.UserInfo("admin", "s3cr3t!"))
+			require_NoError(t, err)
+			defer sysNc.Close()
+
+			sjs := rs.getJetStream()
+			sjs.mu.RLock()
+			syncSubj := sa.Sync
+			sjs.mu.RUnlock()
+
+			// Respond to the catchup with an out-of-order SkipMsg.
+			var eof bool
+			sub, err := sysNc.Subscribe(syncSubj, func(msg *nats.Msg) {
+				if !eof {
+					msg.Respond(encodeStreamMsg(_EMPTY_, _EMPTY_, nil, nil, 10, 0, false))
+					eof = true
+				}
+				msg.Respond(nil)
+			})
+			require_NoError(t, err)
+			defer sub.Drain()
+			require_NoError(t, sysNc.Flush()) // Must flush, otherwise our subscription could be too late.
+
+			err = mset.processSnapshot(&StreamReplicatedState{FirstSeq: 1, LastSeq: 1}, 1)
+			require_Error(t, err, errCatchupTooManyRetries)
+			c.waitOnStreamLeader(globalAccountName, "TEST")
+
+			pubAck, err := js.Publish("foo", nil)
+			require_NoError(t, err)
+			require_Equal(t, pubAck.Sequence, 1)
+			checkFor(t, 2*time.Second, 200*time.Millisecond, func() error {
+				return checkState(t, c, globalAccountName, "TEST")
+			})
+		})
+	}
+}
+
 //
 // DO NOT ADD NEW TESTS IN THIS FILE (unless to balance test times)
 // Add at the end of jetstream_cluster_<n>_test.go, with <n> being the highest value.

server/memstore.go

@@ -359,12 +359,21 @@ func (ms *memStore) StoreMsg(subj string, hdr, msg []byte, ttl int64) (uint64, i
 }
 
 // SkipMsg will use the next sequence number but not store anything.
-func (ms *memStore) SkipMsg() uint64 {
+func (ms *memStore) SkipMsg(seq uint64) (uint64, error) {
 	// Grab time.
 	now := time.Unix(0, ats.AccessTime()).UTC()
 
 	ms.mu.Lock()
-	seq := ms.state.LastSeq + 1
+	defer ms.mu.Unlock()
+
+	// Check sequence matches our last sequence.
+	if seq != ms.state.LastSeq+1 {
+		if seq > 0 {
+			return 0, ErrSequenceMismatch
+		}
+		seq = ms.state.LastSeq + 1
+	}
+
 	ms.state.LastSeq = seq
 	ms.state.LastTime = now
 	if ms.state.Msgs == 0 {
@@ -373,8 +382,7 @@ func (ms *memStore) SkipMsg() uint64 {
 	} else {
 		ms.dmap.Insert(seq)
 	}
-	ms.mu.Unlock()
-	return seq
+	return seq, nil
 }
 
 // Skip multiple msgs.

server/memstore_test.go

@@ -1379,7 +1379,7 @@ func Benchmark_MemStoreNumPendingWithLargeInteriorDeletesScan(b *testing.B) {
 	msg := []byte("abc")
 	ms.StoreMsg("foo.bar.baz", nil, msg, 0)
 	for i := 1; i <= 1_000_000; i++ {
-		ms.SkipMsg()
+		ms.SkipMsg(0)
 	}
 	ms.StoreMsg("foo.bar.baz", nil, msg, 0)
 
@@ -1406,7 +1406,7 @@ func Benchmark_MemStoreNumPendingWithLargeInteriorDeletesExclude(b *testing.B) {
 	msg := []byte("abc")
 	ms.StoreMsg("foo.bar.baz", nil, msg, 0)
 	for i := 1; i <= 1_000_000; i++ {
-		ms.SkipMsg()
+		ms.SkipMsg(0)
 	}
 	ms.StoreMsg("foo.bar.baz", nil, msg, 0)
 

server/store.go

@@ -91,7 +91,7 @@ type ProcessJetStreamMsgHandler func(*inMsg)
 type StreamStore interface {
 	StoreMsg(subject string, hdr, msg []byte, ttl int64) (uint64, int64, error)
 	StoreRawMsg(subject string, hdr, msg []byte, seq uint64, ts int64, ttl int64) error
-	SkipMsg() uint64
+	SkipMsg(seq uint64) (uint64, error)
 	SkipMsgs(seq uint64, num uint64) error
 	FlushAllPending()
 	LoadMsg(seq uint64, sm *StoreMsg) (*StoreMsg, error)

server/store_test.go

@@ -639,7 +639,7 @@ func TestStoreStreamInteriorDeleteAccounting(t *testing.T) {
 		{
 			title: "SkipMsg",
 			action: func(s StreamStore, lseq uint64) {
-				s.SkipMsg()
+				s.SkipMsg(0)
 			},
 		},
 		{

server/stream.go

@@ -5935,7 +5935,7 @@ func (mset *stream) processJetStreamMsg(subject, reply string, hdr, msg []byte,
 
 	// Skip msg here.
 	if noInterest {
-		mset.lseq = store.SkipMsg()
+		mset.lseq, _ = store.SkipMsg(0)
 		mset.lmsgId = msgId
 		// If we have a msgId make sure to save.
 		if msgId != _EMPTY_ {