[IMPROVED] Stage meta operations during Raft catchup (#7540)

When the Raft node underlying the meta layer enters catchup from another
server, it gets placed in the same "recovery mode" such that it can
stage changes into `ru *recoveryUpdates` such that added and deleted
consumers become a noop.

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

Author: neilalexander

server/consumer.go

@@ -963,7 +963,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 	}
 
 	mset.mu.RLock()
-	s, jsa, cfg, acc := mset.srv, mset.jsa, mset.cfg, mset.acc
+	s, js, jsa, cfg, acc := mset.srv, mset.js, mset.jsa, mset.cfg, mset.acc
 	mset.mu.RUnlock()
 
 	// If we do not have the consumer currently assigned to us in cluster mode we will proceed but warn.
@@ -1134,6 +1134,13 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 		created:   time.Now().UTC(),
 	}
 
+	// Add created timestamp used for the store, must match that of the consumer assignment if it exists.
+	if ca != nil {
+		js.mu.RLock()
+		o.created = ca.Created
+		js.mu.RUnlock()
+	}
+
 	// Bind internal client to the user account.
 	o.client.registerWithAccount(a)
 	// Bind to the system account.
@@ -1186,7 +1193,7 @@ func (mset *stream) addConsumerWithAssignment(config *ConsumerConfig, oname stri
 
 	// Setup our storage if not a direct consumer.
 	if !config.Direct {
-		store, err := mset.store.ConsumerStore(o.name, config)
+		store, err := mset.store.ConsumerStore(o.name, o.created, config)
 		if err != nil {
 			mset.mu.Unlock()
 			o.deleteWithoutAdvisory()

server/filestore.go

@@ -10903,7 +10903,7 @@ type consumerFileStore struct {
 	closed  bool
 }
 
-func (fs *fileStore) ConsumerStore(name string, cfg *ConsumerConfig) (ConsumerStore, error) {
+func (fs *fileStore) ConsumerStore(name string, created time.Time, cfg *ConsumerConfig) (ConsumerStore, error) {
 	if fs == nil {
 		return nil, fmt.Errorf("filestore is nil")
 	}
@@ -10926,7 +10926,7 @@ func (fs *fileStore) ConsumerStore(name string, cfg *ConsumerConfig) (ConsumerSt
 	if err := os.MkdirAll(odir, defaultDirPerms); err != nil {
 		return nil, fmt.Errorf("could not create consumer directory - %v", err)
 	}
-	csi := &FileConsumerInfo{Name: name, Created: time.Now().UTC(), ConsumerConfig: *cfg}
+	csi := &FileConsumerInfo{Name: name, Created: created, ConsumerConfig: *cfg}
 	o := &consumerFileStore{
 		fs:   fs,
 		cfg:  csi,
@@ -10983,7 +10983,6 @@ func (fs *fileStore) ConsumerStore(name string, cfg *ConsumerConfig) (ConsumerSt
 	meta := filepath.Join(odir, JetStreamMetaFile)
 	if _, err := os.Stat(meta); err != nil && os.IsNotExist(err) {
 		didCreate = true
-		csi.Created = time.Now().UTC()
 		if err := o.writeConsumerMeta(); err != nil {
 			os.RemoveAll(odir)
 			return nil, err

server/filestore_test.go

@@ -1460,7 +1460,7 @@ func TestFileStoreMeta(t *testing.T) {
 		AckPolicy:      AckAll,
 	}
 	oname := "obs22"
-	obs, err := fs.ConsumerStore(oname, &oconfig)
+	obs, err := fs.ConsumerStore(oname, time.Time{}, &oconfig)
 	if err != nil {
 		t.Fatalf("Unexpected error: %v", err)
 	}
@@ -1809,11 +1809,11 @@ func TestFileStoreSnapshot(t *testing.T) {
 		}
 
 		// Create a few consumers.
-		o1, err := fs.ConsumerStore("o22", &ConsumerConfig{})
+		o1, err := fs.ConsumerStore("o22", time.Time{}, &ConsumerConfig{})
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
-		o2, err := fs.ConsumerStore("o33", &ConsumerConfig{})
+		o2, err := fs.ConsumerStore("o33", time.Time{}, &ConsumerConfig{})
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -1977,7 +1977,7 @@ func TestFileStoreConsumer(t *testing.T) {
 		require_NoError(t, err)
 		defer fs.Stop()
 
-		o, err := fs.ConsumerStore("obs22", &ConsumerConfig{})
+		o, err := fs.ConsumerStore("obs22", time.Time{}, &ConsumerConfig{})
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -2504,7 +2504,7 @@ func TestFileStoreConsumerRedeliveredLost(t *testing.T) {
 		defer fs.Stop()
 
 		cfg := &ConsumerConfig{AckPolicy: AckExplicit}
-		o, err := fs.ConsumerStore("o22", cfg)
+		o, err := fs.ConsumerStore("o22", time.Time{}, cfg)
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -2513,7 +2513,7 @@ func TestFileStoreConsumerRedeliveredLost(t *testing.T) {
 			t.Helper()
 			o.Stop()
 			time.Sleep(20 * time.Millisecond) // Wait for all things to settle.
-			o, err = fs.ConsumerStore("o22", cfg)
+			o, err = fs.ConsumerStore("o22", time.Time{}, cfg)
 			if err != nil {
 				t.Fatalf("Unexpected error: %v", err)
 			}
@@ -2569,7 +2569,7 @@ func TestFileStoreConsumerFlusher(t *testing.T) {
 		require_NoError(t, err)
 		defer fs.Stop()
 
-		o, err := fs.ConsumerStore("o22", &ConsumerConfig{})
+		o, err := fs.ConsumerStore("o22", time.Time{}, &ConsumerConfig{})
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -2601,7 +2601,7 @@ func TestFileStoreConsumerDeliveredUpdates(t *testing.T) {
 		defer fs.Stop()
 
 		// Simple consumer, no ack policy configured.
-		o, err := fs.ConsumerStore("o22", &ConsumerConfig{})
+		o, err := fs.ConsumerStore("o22", time.Time{}, &ConsumerConfig{})
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -2655,7 +2655,7 @@ func TestFileStoreConsumerDeliveredAndAckUpdates(t *testing.T) {
 		defer fs.Stop()
 
 		// Simple consumer, no ack policy configured.
-		o, err := fs.ConsumerStore("o22", &ConsumerConfig{AckPolicy: AckExplicit})
+		o, err := fs.ConsumerStore("o22", time.Time{}, &ConsumerConfig{AckPolicy: AckExplicit})
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -2745,7 +2745,7 @@ func TestFileStoreConsumerDeliveredAndAckUpdates(t *testing.T) {
 		}
 		o.Stop()
 
-		o, err = fs.ConsumerStore("o22", &ConsumerConfig{AckPolicy: AckExplicit})
+		o, err = fs.ConsumerStore("o22", time.Time{}, &ConsumerConfig{AckPolicy: AckExplicit})
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -2847,7 +2847,7 @@ func TestFileStoreConsumerPerf(t *testing.T) {
 		require_NoError(t, err)
 		defer fs.Stop()
 
-		o, err := fs.ConsumerStore("o22", &ConsumerConfig{AckPolicy: AckExplicit})
+		o, err := fs.ConsumerStore("o22", time.Time{}, &ConsumerConfig{AckPolicy: AckExplicit})
 		if err != nil {
 			t.Fatalf("Unexpected error: %v", err)
 		}
@@ -4171,7 +4171,7 @@ func TestFileStoreEncrypted(t *testing.T) {
 			fs.StoreMsg(subj, nil, msg, 0)
 		}
 
-		o, err := fs.ConsumerStore("o22", &ConsumerConfig{})
+		o, err := fs.ConsumerStore("o22", time.Time{}, &ConsumerConfig{})
 		require_NoError(t, err)
 
 		state := &ConsumerState{}
@@ -4194,7 +4194,7 @@ func TestFileStoreEncrypted(t *testing.T) {
 		require_NoError(t, err)
 		require_True(t, string(sm.msg) == "aes ftw")
 
-		o, err = fs.ConsumerStore("o22", &ConsumerConfig{})
+		o, err = fs.ConsumerStore("o22", time.Time{}, &ConsumerConfig{})
 		require_NoError(t, err)
 		rstate, err := o.State()
 		require_NoError(t, err)
@@ -4990,7 +4990,7 @@ func TestFileStoreConsumerStoreEncodeAfterRestart(t *testing.T) {
 			require_NoError(t, err)
 			defer fs.Stop()
 
-			o, err := fs.ConsumerStore("o22", &ConsumerConfig{AckPolicy: AckExplicit})
+			o, err := fs.ConsumerStore("o22", time.Time{}, &ConsumerConfig{AckPolicy: AckExplicit})
 			require_NoError(t, err)
 			defer o.Stop()
 
@@ -5007,7 +5007,7 @@ func TestFileStoreConsumerStoreEncodeAfterRestart(t *testing.T) {
 			require_NoError(t, err)
 			defer fs.Stop()
 
-			o, err := fs.ConsumerStore("o22", &ConsumerConfig{AckPolicy: AckExplicit})
+			o, err := fs.ConsumerStore("o22", time.Time{}, &ConsumerConfig{AckPolicy: AckExplicit})
 			require_NoError(t, err)
 			defer o.Stop()
 
@@ -8366,7 +8366,7 @@ func Benchmark_FileStoreCreateConsumerStores(b *testing.B) {
 			b.ResetTimer()
 			for i := 0; i < b.N; i++ {
 				oname := fmt.Sprintf("obs22_%d", i)
-				ofs, err := fs.ConsumerStore(oname, &oconfig)
+				ofs, err := fs.ConsumerStore(oname, time.Time{}, &oconfig)
 				require_NoError(b, err)
 				require_NoError(b, ofs.Stop())
 			}

server/jetstream_api.go

@@ -2821,11 +2821,11 @@ func (s *Server) jsLeaderAccountPurgeRequest(sub *subscription, c *client, _ *Ac
 	for _, osa := range streams {
 		for _, oca := range osa.consumers {
 			oca.deleted = true
-			ca := &consumerAssignment{Group: oca.Group, Stream: oca.Stream, Name: oca.Name, Config: oca.Config, Subject: subject, Client: oca.Client}
+			ca := &consumerAssignment{Group: oca.Group, Stream: oca.Stream, Name: oca.Name, Config: oca.Config, Subject: subject, Client: oca.Client, Created: oca.Created}
 			meta.Propose(encodeDeleteConsumerAssignment(ca))
 			nc++
 		}
-		sa := &streamAssignment{Group: osa.Group, Config: osa.Config, Subject: subject, Client: osa.Client}
+		sa := &streamAssignment{Group: osa.Group, Config: osa.Config, Subject: subject, Client: osa.Client, Created: osa.Created}
 		meta.Propose(encodeDeleteStreamAssignment(sa))
 		ns++
 	}