reset consumer state to stream state

Author: tilakraj94

server/consumer.go

@@ -963,7 +963,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	}
 
 	mset.mu.RLock()
-	s, js, jsa, cfg, acc := mset.srv, mset.js, mset.jsa, mset.cfg, mset.acc
+	s, js, jsa, cfg, acc, lseq := mset.srv, mset.js, mset.jsa, mset.cfg, mset.acc, mset.lseq
 	mset.mu.RUnlock()
 
 	// If we do not have the consumer currently assigned to us in cluster mode we will proceed but warn.
@@ -1228,6 +1228,31 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 		o.mu.Lock()
 		o.readStoredState()
 		o.mu.Unlock()
+
+		// only for streams with r1; if we are recovering and the consumer sequence is greater than the last sequence,
+		// we need to select a new starting sequence
+		// run only for single replica streams
+		if mset.cfg.Replicas == 1 && o.sseq > 2 && lseq < o.sseq && isRecovering {
+			o.srv.Warnf("Consumer %q recovering with sseq %d greater than stream last seq %d, selecting new start", o.name, o.sseq, lseq)
+
+			// set starting sequences to in-memory state
+			state := &ConsumerState{}
+			o.mu.Lock()
+			o.selectStartingSeqNo()
+
+			state.Delivered.Stream = o.sseq
+			state.Delivered.Consumer = o.dseq
+			state.AckFloor.Stream = o.asflr
+			state.AckFloor.Consumer = o.adflr
+			o.mu.Unlock()
+			err := o.store.ForceUpdate(state)
+			if err != nil {
+				s.Warnf("Consumer %q error updating state: %v", o.name, err)
+				return nil, NewJSConsumerStoreFailedError(err)
+			}
+
+			o.srv.Warnf("Consumer %q adjusted starting sequence to %d", o.name, o.sseq)
+		}
 	} else {
 		// Select starting sequence number
 		o.selectStartingSeqNo()

server/filestore.go

@@ -11593,6 +11593,31 @@ func (o *consumerFileStore) Update(state *ConsumerState) error {
 	return nil
 }
 
+// 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")
+	}
+	if state.AckFloor.Stream > state.Delivered.Stream {
+		return fmt.Errorf("bad ack floor for stream")
+	}
+
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	o.state.Delivered = state.Delivered
+	o.state.AckFloor = state.AckFloor
+	o.state.Pending = make(map[uint64]*Pending)
+	o.state.Redelivered = make(map[uint64]uint64)
+
+	o.kickFlusher()
+
+	return nil
+}
+
 // Will encrypt the state with our asset key. Will be a no-op if encryption not enabled.
 // Lock should be held.
 func (o *consumerFileStore) encryptState(buf []byte) ([]byte, error) {

server/filestore_test.go

@@ -8518,6 +8518,94 @@ func TestFileStoreRecoverFullStateDetectCorruptState(t *testing.T) {
 	require_Error(t, err, errCorruptState)
 }
 
+func TestFileStoreResetConsumerToStreamState(t *testing.T) {
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: t.TempDir()},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	msg := []byte("abc")
+	for i := 1; i <= 30; i++ {
+		_, _, err = fs.StoreMsg(fmt.Sprintf("foo.%d", i), nil, msg, 0)
+		require_NoError(t, err)
+	}
+
+	err = fs.writeFullState()
+	require_NoError(t, err)
+
+	obs, err := fs.ConsumerStore("c1", time.Now(), &ConsumerConfig{
+		Durable:       "c1",
+		FilterSubject: "foo.*",
+		AckPolicy:     AckNone,
+		DeliverPolicy: DeliverAll,
+	})
+
+	require_NoError(t, err)
+	defer obs.Stop()
+
+	state := &ConsumerState{}
+	state.Delivered = SequencePair{
+		Consumer: 5,
+		Stream:   5,
+	}
+
+	state.AckFloor = SequencePair{
+		Consumer: 5,
+		Stream:   5,
+	}
+
+	// set to 5
+	err = obs.Update(state)
+	require_NoError(t, err)
+
+	currState, err := obs.State()
+	require_NoError(t, err)
+
+	require_Equal(t, fs.state.LastSeq, uint64(30))
+	require_Equal(t, fs.state.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.state.FirstSeq = 0
+	fs.state.LastSeq = 0
+
+	// set back to lower values
+	newState := &ConsumerState{}
+	newState.Delivered = SequencePair{
+		Consumer: 1,
+		Stream:   4,
+	}
+
+	newState.AckFloor = SequencePair{
+		Consumer: 1,
+		Stream:   3,
+	}
+
+	// update should fail but force update should pass
+	err = obs.Update(newState)
+	require_Error(t, err, fmt.Errorf("old update ignored"))
+
+	err = obs.ForceUpdate(newState)
+	require_NoError(t, err)
+
+	fullState, err := obs.State()
+	require_NoError(t, err)
+
+	borrowedState, err := obs.BorrowState()
+	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))
+}
+
 func TestFileStoreNumPendingMulti(t *testing.T) {
 	fs, err := newFileStore(
 		FileStoreConfig{StoreDir: t.TempDir()},

server/memstore.go

@@ -2311,6 +2311,27 @@ func (o *consumerMemStore) Update(state *ConsumerState) error {
 	return nil
 }
 
+func (o *consumerMemStore) ForceUpdate(state *ConsumerState) error {
+	// Sanity checks.
+	if state.AckFloor.Consumer > state.Delivered.Consumer {
+		return fmt.Errorf("bad ack floor for consumer")
+	}
+	if state.AckFloor.Stream > state.Delivered.Stream {
+		return fmt.Errorf("bad ack floor for stream")
+	}
+
+	// Replace our state.
+	o.mu.Lock()
+	defer o.mu.Unlock()
+
+	o.state.Delivered = state.Delivered
+	o.state.AckFloor = state.AckFloor
+	o.state.Pending = make(map[uint64]*Pending)
+	o.state.Redelivered = make(map[uint64]uint64)
+
+	return nil
+}
+
 // SetStarting sets our starting stream sequence.
 func (o *consumerMemStore) SetStarting(sseq uint64) error {
 	o.mu.Lock()

server/store.go

@@ -363,6 +363,7 @@ type ConsumerStore interface {
 	UpdateAcks(dseq, sseq uint64) error
 	UpdateConfig(cfg *ConsumerConfig) error
 	Update(*ConsumerState) error
+	ForceUpdate(*ConsumerState) error
 	State() (*ConsumerState, error)
 	BorrowState() (*ConsumerState, error)
 	EncodedState() ([]byte, error)