(2.12) Default AsyncFlush for replicated streams (#7163)

`AllowAsyncFlush` used to be an opt-in stream-level setting, which
allows a replicated stream to asynchronously flush writes to the
underlying JetStream stream. This is a safe operation as it's backed by
a replicated log that _is_ flushed, and we make sure to flush any
remaining async writes before snapshotting and compacting the log. After
upgrading to 2.12, you'll see improved performance by default, with no
action required by users.

This setting is automatically enabled for streams with `Replicas > 1`
and with the `SyncAlways` server setting disabled (default).

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

Author: neilalexander

server/filestore.go

@@ -679,7 +679,7 @@ func (fs *fileStore) UpdateConfig(cfg *StreamConfig) error {
 
 	if lmb := fs.lmb; lmb != nil {
 		// Enable/disable async flush depending on if it's supported and already initialized.
-		supportsAsyncFlush := cfg.AllowAsyncFlush && cfg.Replicas > 1
+		supportsAsyncFlush := !fs.fcfg.SyncAlways && cfg.Replicas > 1
 		if supportsAsyncFlush && !fs.fcfg.AsyncFlush {
 			fs.fcfg.AsyncFlush = true
 			lmb.spinUpFlushLoop()

server/jetstream_benchmark_test.go

@@ -858,14 +858,13 @@ func BenchmarkJetStreamPublish(b *testing.B) {
 		replicas    int
 		messageSize int
 		numSubjects int
-		asyncFlush  bool
 	}{
-		{1, 1, 10, 1, false},   // Single node, 10B messages
-		{1, 1, 1024, 1, false}, // Single node, 1KB messages
-		{3, 3, 10, 1, false},   // 3-nodes cluster, R=3, 10B messages
-		{3, 3, 1024, 1, false}, // 3-nodes cluster, R=3, 1KB messages
-		{3, 3, 10, 1, true},    // 3-nodes cluster, R=3, 10B messages (async flush)
-		{3, 3, 1024, 1, true},  // 3-nodes cluster, R=3, 1KB messages (async flush)
+		{1, 1, 10, 1},   // Single node, 10B messages
+		{1, 1, 1024, 1}, // Single node, 1KB messages
+		{3, 3, 10, 1},   // 3-nodes cluster, R=3, 10B messages
+		{3, 3, 1024, 1}, // 3-nodes cluster, R=3, 1KB messages
+		{3, 3, 10, 1},   // 3-nodes cluster, R=3, 10B messages (async flush)
+		{3, 3, 1024, 1}, // 3-nodes cluster, R=3, 1KB messages (async flush)
 	}
 
 	// All the cases above are run with each of the publisher cases below
@@ -887,10 +886,6 @@ func BenchmarkJetStreamPublish(b *testing.B) {
 			bc.messageSize,
 			bc.numSubjects,
 		)
-		if bc.asyncFlush {
-			name += ",AsyncFlush"
-		}
-
 		b.Run(
 			name,
 			func(b *testing.B) {
@@ -939,11 +934,10 @@ func BenchmarkJetStreamPublish(b *testing.B) {
 								b.Logf("Creating stream with R=%d and %d input subjects", bc.replicas, bc.numSubjects)
 							}
 							_, err = jsStreamCreate(b, nc, &StreamConfig{
-								Name:            streamName,
-								Subjects:        subjects,
-								Replicas:        bc.replicas,
-								Storage:         FileStorage,
-								AllowAsyncFlush: bc.asyncFlush,
+								Name:     streamName,
+								Subjects: subjects,
+								Replicas: bc.replicas,
+								Storage:  FileStorage,
 							})
 							if err != nil {
 								b.Fatalf("Error creating stream: %v", err)
@@ -1942,11 +1936,10 @@ func BenchmarkJetStreamPublishConcurrent(b *testing.B) {
 	replicasCases := []struct {
 		clusterSize int
 		replicas    int
-		asyncFlush  bool
 	}{
-		{1, 1, false},
-		{3, 3, false},
-		{3, 3, true},
+		{1, 1},
+		{3, 3},
+		{3, 3},
 	}
 
 	workload := func(b *testing.B, numPubs int, messageSize int64, clientUrl string) {
@@ -2056,9 +2049,6 @@ func BenchmarkJetStreamPublishConcurrent(b *testing.B) {
 	// benchmark case matrix
 	for _, replicasCase := range replicasCases {
 		title := fmt.Sprintf("N=%d,R=%d", replicasCase.clusterSize, replicasCase.replicas)
-		if replicasCase.asyncFlush {
-			title += ",AsyncFlush"
-		}
 		b.Run(
 			title,
 			func(b *testing.B) {
@@ -2079,11 +2069,10 @@ func BenchmarkJetStreamPublishConcurrent(b *testing.B) {
 
 										// create stream
 										_, err := jsStreamCreate(b, nc, &StreamConfig{
-											Name:            streamName,
-											Subjects:        []string{subject},
-											Replicas:        replicasCase.replicas,
-											Storage:         FileStorage,
-											AllowAsyncFlush: replicasCase.asyncFlush,
+											Name:     streamName,
+											Subjects: []string{subject},
+											Replicas: replicasCase.replicas,
+											Storage:  FileStorage,
 										})
 										if err != nil {
 											b.Fatal(err)

server/jetstream_cluster_1_test.go

@@ -9078,12 +9078,18 @@ func TestJetStreamClusterSetPreferredToOnlineNode(t *testing.T) {
 }
 
 func TestJetStreamClusterAsyncFlushBasics(t *testing.T) {
-	test := func(t *testing.T, replicas int) {
-		supportsAsync := replicas > 1
+	test := func(t *testing.T, syncAlways bool) {
+		supportsAsync := !syncAlways
 
 		c := createJetStreamClusterExplicit(t, "R3S", 3)
 		defer c.shutdown()
 
+		for _, s := range c.servers {
+			s.optsMu.Lock()
+			s.opts.SyncAlways = syncAlways
+			s.optsMu.Unlock()
+		}
+
 		nc, js := jsClientConnect(t, c.randomServer())
 		defer nc.Close()
 
@@ -9120,11 +9126,10 @@ func TestJetStreamClusterAsyncFlushBasics(t *testing.T) {
 		}
 
 		cfg := &StreamConfig{
-			Name:            "TEST",
-			Subjects:        []string{"foo"},
-			Storage:         FileStorage,
-			Replicas:        replicas,
-			AllowAsyncFlush: false,
+			Name:     "TEST",
+			Subjects: []string{"foo"},
+			Storage:  FileStorage,
+			Replicas: 1,
 		}
 		// Test disabled async flush on create.
 		_, err := jsStreamCreate(t, nc, cfg)
@@ -9135,15 +9140,18 @@ func TestJetStreamClusterAsyncFlushBasics(t *testing.T) {
 		checkStoreIsAsync(false)
 
 		// Enabling async flush.
-		cfg.AllowAsyncFlush = true
+		cfg.Replicas = 3
 		_, err = jsStreamUpdate(t, nc, cfg)
 		require_NoError(t, err)
+		checkFor(t, 2*time.Second, 200*time.Millisecond, func() error {
+			return checkState(t, c, globalAccountName, "TEST")
+		})
 		_, err = js.Publish("foo", nil)
 		require_NoError(t, err)
 		checkStoreIsAsync(supportsAsync)
 
 		// Disabling async flush.
-		cfg.AllowAsyncFlush = false
+		cfg.Replicas = 1
 		_, err = jsStreamUpdate(t, nc, cfg)
 		require_NoError(t, err)
 		_, err = js.Publish("foo", nil)
@@ -9152,7 +9160,7 @@ func TestJetStreamClusterAsyncFlushBasics(t *testing.T) {
 
 		// Test async flush on create.
 		require_NoError(t, js.DeleteStream("TEST"))
-		cfg.AllowAsyncFlush = true
+		cfg.Replicas = 3
 		_, err = jsStreamCreate(t, nc, cfg)
 		require_NoError(t, err)
 		s = c.streamLeader(globalAccountName, "TEST")
@@ -9161,8 +9169,8 @@ func TestJetStreamClusterAsyncFlushBasics(t *testing.T) {
 		checkStoreIsAsync(supportsAsync)
 	}
 
-	t.Run("R1", func(t *testing.T) { test(t, 1) })
-	t.Run("R3", func(t *testing.T) { test(t, 3) })
+	t.Run("Default", func(t *testing.T) { test(t, false) })
+	t.Run("SyncAlways", func(t *testing.T) { test(t, true) })
 }
 
 func TestJetStreamClusterAsyncFlushFileStoreFlushOnSnapshot(t *testing.T) {
@@ -9173,11 +9181,10 @@ func TestJetStreamClusterAsyncFlushFileStoreFlushOnSnapshot(t *testing.T) {
 	defer nc.Close()
 
 	_, err := jsStreamCreate(t, nc, &StreamConfig{
-		Name:            "TEST",
-		Subjects:        []string{"foo"},
-		Storage:         FileStorage,
-		Replicas:        3,
-		AllowAsyncFlush: true,
+		Name:     "TEST",
+		Subjects: []string{"foo"},
+		Storage:  FileStorage,
+		Replicas: 3,
 	})
 	require_NoError(t, err)
 

server/jetstream_versioning.go

@@ -55,11 +55,6 @@ func setStaticStreamMetadata(cfg *StreamConfig) {
 		requires(2)
 	}
 
-	// Async flush was added in v2.12 and require API level 2.
-	if cfg.AllowAsyncFlush {
-		requires(2)
-	}
-
 	cfg.Metadata[JSRequiredLevelMetadataKey] = strconv.Itoa(requiredApiLevel)
 }
 

server/jetstream_versioning_test.go

@@ -82,11 +82,6 @@ func TestJetStreamSetStaticStreamMetadata(t *testing.T) {
 			cfg:              &StreamConfig{AllowAtomicPublish: true},
 			expectedMetadata: metadataAtLevel("2"),
 		},
-		{
-			desc:             "AllowAsyncFlush",
-			cfg:              &StreamConfig{AllowAsyncFlush: true},
-			expectedMetadata: metadataAtLevel("2"),
-		},
 	} {
 		t.Run(test.desc, func(t *testing.T) {
 			setStaticStreamMetadata(test.cfg)

server/stream.go

@@ -116,10 +116,6 @@ type StreamConfig struct {
 	// AllowAtomicPublish allows atomic batch publishing into the stream.
 	AllowAtomicPublish bool `json:"allow_atomic"`
 
-	// AllowAsyncFlush allows replicated streams to asynchronously flush
-	// to the stream, improving throughput.
-	AllowAsyncFlush bool `json:"allow_async_flush"`
-
 	// Metadata is additional metadata for the Stream.
 	Metadata map[string]string `json:"metadata,omitempty"`
 }
@@ -798,12 +794,12 @@ func (a *Account) addStreamWithAssignment(config *StreamConfig, fsConfig *FileSt
 		}
 	}
 	fsCfg.StoreDir = storeDir
-	// Async flushing is only allowed if the stream has a sync log backing it.
-	fsCfg.AsyncFlush = config.AllowAsyncFlush && config.Replicas > 1
 	// Grab configured sync interval.
 	fsCfg.SyncInterval = s.getOpts().SyncInterval
 	fsCfg.SyncAlways = s.getOpts().SyncAlways
 	fsCfg.Compression = config.Compression
+	// Async flushing is only allowed if the stream has a sync log backing it.
+	fsCfg.AsyncFlush = !fsCfg.SyncAlways && config.Replicas > 1
 
 	if err := mset.setupStore(fsCfg); err != nil {
 		mset.stop(true, false)