add variable to mark reconciled consumers

Author: tilakraj94

server/consumer.go

@@ -5746,8 +5746,6 @@ func (o *consumer) hasSkipListPending() bool {
 	return o.lss != nil && len(o.lss.seqs) > 0
 }
 
-}
-
 // reconcileStateWithStream reconciles consumer state when the stream has reverted
 // due to data loss (e.g., VM crash). This handles the case where consumer state
 // is ahead of the stream's last sequence.

server/filestore.go

@@ -11035,22 +11035,23 @@ func (fs *fileStore) SyncDeleted(dbs DeleteBlocks) {
 ////////////////////////////////////////////////////////////////////////////////
 
 type consumerFileStore struct {
-	mu      sync.Mutex
-	fs      *fileStore
-	cfg     *FileConsumerInfo
-	prf     keyGen
-	aek     cipher.AEAD
-	name    string
-	odir    string
-	ifn     string
-	hh      *highwayhash.Digest64
-	state   ConsumerState
-	fch     chan struct{}
-	qch     chan struct{}
-	flusher bool
-	writing bool
-	dirty   bool
-	closed  bool
+	mu          sync.Mutex
+	fs          *fileStore
+	cfg         *FileConsumerInfo
+	prf         keyGen
+	aek         cipher.AEAD
+	name        string
+	odir        string
+	ifn         string
+	hh          *highwayhash.Digest64
+	state       ConsumerState
+	fch         chan struct{}
+	qch         chan struct{}
+	flusher     bool
+	writing     bool
+	dirty       bool
+	closed      bool
+	stateLoaded bool // Set when state is explicitly loaded/updated (reconciled).
 }
 
 func (fs *fileStore) ConsumerStore(name string, created time.Time, cfg *ConsumerConfig) (ConsumerStore, error) {
@@ -11632,6 +11633,7 @@ func (o *consumerFileStore) ForceUpdate(state *ConsumerState) error {
 	o.state.AckFloor = state.AckFloor
 	o.state.Pending = pending
 	o.state.Redelivered = redelivered
+	o.stateLoaded = true
 
 	o.kickFlusher()
 
@@ -11835,7 +11837,7 @@ func (o *consumerFileStore) stateWithCopyLocked(doCopy bool) (*ConsumerState, er
 	state := &ConsumerState{}
 
 	// See if we have a running state or if we need to read in from disk.
-	if o.state.Delivered.Consumer != 0 || o.state.Delivered.Stream != 0 {
+	if o.stateLoaded || o.state.Delivered.Consumer != 0 || o.state.Delivered.Stream != 0 {
 		state.Delivered = o.state.Delivered
 		state.AckFloor = o.state.AckFloor
 		if len(o.state.Pending) > 0 {