add test case

Signed-off-by: Tilak Raj <tilak.raj94@gmail.com>

Author: tilakraj94

server/filestore.go

@@ -11657,7 +11657,6 @@ func (o *consumerFileStore) Update(state *ConsumerState) error {
 // ForceUpdate updates the consumer state without the backwards check.
 // This is used during recovery when we need to reset the consumer to an earlier sequence.
 func (o *consumerFileStore) ForceUpdate(state *ConsumerState) error {
-	o.fs.warn("Consumer %q force updating state: %+v", o.name, state)
 	// Sanity checks.
 	if state.AckFloor.Consumer > state.Delivered.Consumer {
 		return fmt.Errorf("bad ack floor for consumer")
@@ -11687,7 +11686,6 @@ func (o *consumerFileStore) ForceUpdate(state *ConsumerState) error {
 
 	// Replace our state.
 	o.mu.Lock()
-
 	o.state.Delivered = state.Delivered
 	o.state.AckFloor = state.AckFloor
 	o.state.Pending = pending

server/filestore_test.go

@@ -8545,15 +8545,8 @@ func TestFileStoreResetConsumerToStreamState(t *testing.T) {
 	defer obs.Stop()
 
 	state := &ConsumerState{}
-	state.Delivered = SequencePair{
-		Consumer: 5,
-		Stream:   5,
-	}
-
-	state.AckFloor = SequencePair{
-		Consumer: 5,
-		Stream:   5,
-	}
+	state.Delivered = SequencePair{Consumer: 5, Stream: 5}
+	state.AckFloor = SequencePair{Consumer: 5, Stream: 5}
 
 	// set to 5
 	err = obs.Update(state)
@@ -8562,27 +8555,23 @@ func TestFileStoreResetConsumerToStreamState(t *testing.T) {
 	currState, err := obs.State()
 	require_NoError(t, err)
 
-	require_Equal(t, fs.state.LastSeq, uint64(30))
-	require_Equal(t, fs.state.FirstSeq, uint64(1))
+	fsState := fs.State()
+	require_Equal(t, fsState.LastSeq, uint64(30))
+	require_Equal(t, fsState.FirstSeq, uint64(1))
 	require_Equal(t, currState.AckFloor, state.AckFloor)
 	require_Equal(t, currState.Delivered, state.Delivered)
 	require_Equal(t, len(currState.Redelivered), len(state.Redelivered))
 	require_Equal(t, len(currState.Pending), len(state.Pending))
 
-	// reset our state to 0.
+	fs.mu.Lock()
 	fs.state.FirstSeq = 0
 	fs.state.LastSeq = 0
+	fs.mu.Unlock()
 
 	// set back to lower values
-	newState := &ConsumerState{}
-	newState.Delivered = SequencePair{
-		Consumer: 1,
-		Stream:   4,
-	}
-
-	newState.AckFloor = SequencePair{
-		Consumer: 1,
-		Stream:   3,
+	newState := &ConsumerState{
+		Delivered: SequencePair{Consumer: 1, Stream: 4},
+		AckFloor:  SequencePair{Consumer: 1, Stream: 3},
 	}
 
 	// update should fail but force update should pass
@@ -8592,18 +8581,13 @@ func TestFileStoreResetConsumerToStreamState(t *testing.T) {
 	err = obs.ForceUpdate(newState)
 	require_NoError(t, err)
 
-	fullState, err := obs.State()
-	require_NoError(t, err)
-
-	borrowedState, err := obs.BorrowState()
+	currState, err = obs.State()
 	require_NoError(t, err)
 
-	require_Equal(t, fullState.Delivered, borrowedState.Delivered)
-	require_Equal(t, fullState.AckFloor, borrowedState.AckFloor)
-	require_Equal(t, newState.Delivered, borrowedState.Delivered)
-	require_Equal(t, newState.AckFloor, borrowedState.AckFloor)
-	require_Equal(t, len(newState.Redelivered), len(borrowedState.Redelivered))
-	require_Equal(t, len(newState.Pending), len(borrowedState.Pending))
+	require_Equal(t, currState.AckFloor, newState.AckFloor)
+	require_Equal(t, currState.Delivered, newState.Delivered)
+	require_Equal(t, len(currState.Redelivered), len(newState.Redelivered))
+	require_Equal(t, len(currState.Pending), len(newState.Pending))
 }
 
 func TestFileStoreNumPendingMulti(t *testing.T) {

server/jetstream_consumer_test.go

@@ -25,7 +25,8 @@ import (
 	"fmt"
 	"math/rand"
 	"net/url"
-	"os"
+	os "os"
+	"path/filepath"
 	"reflect"
 	"runtime"
 	"slices"
@@ -10587,3 +10588,117 @@ func TestJetStreamConsumerNoDeleteAfterConcurrentShutdownAndLeaderChange(t *test
 	o = mset.lookupConsumer("CONSUMER")
 	require_NotNil(t, o)
 }
+
+func TestJetStreamConsumerReconcileConsumerAfterStreamDataLoss(t *testing.T) {
+	test := func(t *testing.T, totalMsgs int) {
+		storeDir := t.TempDir()
+		conf := createConfFile(t, []byte(fmt.Sprintf(`
+		listen: 127.0.0.1:-1
+		jetstream: {store_dir: %q}
+	`, storeDir)))
+
+		s, _ := RunServerWithConfig(conf)
+		defer s.Shutdown()
+
+		nc, js := jsClientConnect(t, s)
+		defer nc.Close()
+
+		_, err := js.AddStream(&nats.StreamConfig{
+			Name:     "TEST",
+			Subjects: []string{"foo"},
+		})
+		require_NoError(t, err)
+
+		// Publish a few messages.
+		for range totalMsgs {
+			_, err = js.Publish("foo", nil)
+			require_NoError(t, err)
+		}
+
+		_, err = js.AddConsumer("TEST", &nats.ConsumerConfig{
+			Durable:   "DURABLE",
+			AckPolicy: nats.AckExplicitPolicy,
+		})
+		require_NoError(t, err)
+
+		sub, err := js.PullSubscribe(_EMPTY_, "CONSUMER", nats.BindStream("TEST"))
+		require_NoError(t, err)
+		defer sub.Drain()
+
+		// Consume all available messages.
+		msgs, err := sub.Fetch(totalMsgs, nats.MaxWait(200*time.Millisecond))
+		require_NoError(t, err)
+		require_Len(t, len(msgs), totalMsgs)
+		for _, msg := range msgs {
+			require_NoError(t, msg.AckSync())
+		}
+
+		// Confirm the consumer info reports all messages as delivered and acked.
+		lseq := uint64(totalMsgs)
+		ci, err := js.ConsumerInfo("TEST", "CONSUMER")
+		require_NoError(t, err)
+		require_Equal(t, ci.NumPending, 0)
+		require_Equal(t, ci.NumAckPending, 0)
+		require_Equal(t, ci.Delivered.Stream, lseq)
+		require_Equal(t, ci.AckFloor.Stream, lseq)
+		require_Equal(t, ci.Delivered.Consumer, lseq)
+
+		// Shut down the server and manually remove or truncate the message blocks, simulating data loss.
+		mset, err := s.globalAccount().lookupStream("TEST")
+		require_NoError(t, err)
+		fs := mset.store.(*fileStore)
+		blk := filepath.Join(fs.fcfg.StoreDir, msgDir, "1.blk")
+		index := filepath.Join(fs.fcfg.StoreDir, msgDir, streamStreamStateFile)
+		nc.Close()
+		s.Shutdown()
+		if totalMsgs > 1 {
+			stat, err := os.Stat(blk)
+			require_NoError(t, err)
+			require_NoError(t, os.Truncate(blk, stat.Size()/2+1))
+		} else {
+			require_NoError(t, os.Remove(blk))
+		}
+		require_NoError(t, os.Remove(index))
+
+		// Restart the server and reconnect.
+		s, _ = RunServerWithConfig(conf)
+		defer s.Shutdown()
+		nc, js = jsClientConnect(t, s)
+		defer nc.Close()
+
+		// Publish another message. Due to the simulated data loss, the stream sequence should continue
+		// counting after truncating the corrupted data.
+		pubAck, err := js.Publish("foo", nil)
+		require_NoError(t, err)
+		require_Equal(t, pubAck.Sequence, lseq)
+
+		sub, err = js.PullSubscribe(_EMPTY_, "CONSUMER", nats.BindStream("TEST"))
+		require_NoError(t, err)
+		defer sub.Drain()
+
+		// The consumer should be able to consume above message.
+		// Previously the consumer state would not be reconciled and would not be able to consume the message.
+		msgs, err = sub.Fetch(1, nats.MaxWait(200*time.Millisecond))
+		require_NoError(t, err)
+		require_Len(t, len(msgs), 1)
+		msg := msgs[0]
+		meta, err := msg.Metadata()
+		require_NoError(t, err)
+		require_Equal(t, meta.Sequence.Stream, lseq)
+		require_NoError(t, msg.AckSync())
+
+		// Confirm the consumer info reports all messages as delivered and acked.
+		// But the delivered sequence shouldn't be reset and still move monotonically.
+		ci, err = js.ConsumerInfo("TEST", "CONSUMER")
+		require_NoError(t, err)
+		require_Equal(t, ci.NumPending, 0)
+		require_Equal(t, ci.NumAckPending, 0)
+		require_Equal(t, ci.Delivered.Stream, lseq)
+		require_Equal(t, ci.AckFloor.Stream, lseq)
+		require_Equal(t, ci.Delivered.Consumer, lseq+1)
+	}
+
+	for _, totalMsgs := range []int{1, 2} {
+		t.Run(fmt.Sprint(totalMsgs), func(t *testing.T) { test(t, totalMsgs) })
+	}
+}