Repository: nats-io/nats-server
Pull request #3238 (2022-07-03 18:23:49)

Direct stream access enhancements
As direct access becomes more important through materialized views like KeyValue, and having alternatives to a full blown JetStream consumers is important in future hybrid architectures, this PR introduces two enhancements.

Both of these are opt in and will be easily identified by clients.

1. DirectAccess now uses a queue group and the system allows non-leader replicas to participate.
2. Mirrors can now optionally be a part of the origin's group.

So for the second, this could allow a KV store with multiple mirrors to have a single subject to retrieve a message that operates in a single queue group. Since these are cluster aware, meaning we prefer the ones closer first and only move to a nearby cluster when no interest exists, adding a stream mirror when a large amount of traffic is direct retrieval, like geo diverse KVs, this would instantly improve latency and scale.

Signed-off-by: Derek Collison <derek@nats.io>

/cc @nats-io/core



Patch:
diff --git a/server/jetstream_api.go b/server/jetstream_api.go
index 41ff9efe..4ba29dda 100644
--- a/server/jetstream_api.go
+++ b/server/jetstream_api.go
@@ -1303,6 +1303,13 @@ func (s *Server) jsStreamCreateRequest(sub *subscription, c *client, _ *Account,
 		return
 	}
 
+	// If we are told to do mirror direct but are not mirroring, error.
+	if cfg.MirrorDirect && cfg.Mirror == nil {
+		resp.Error = NewJSStreamInvalidConfigError(fmt.Errorf("stream has no mirror but does have mirror direct"))
+		s.sendAPIErrResponse(ci, acc, subject, reply, string(msg), s.jsonResponse(&resp))
+		return
+	}
+
 	// Hand off to cluster for processing.
 	if s.JetStreamIsClustered() {
 		s.jsClusteredStreamRequest(ci, acc, subject, reply, rmsg, &cfg)
@@ -1798,11 +1805,6 @@ func (s *Server) jsStreamInfoRequest(sub *subscription, c *client, a *Account, s
 	if clusterWideConsCount > 0 {
 		resp.StreamInfo.State.Consumers = clusterWideConsCount
 	}
-	if mset.isMirror() {
-		resp.StreamInfo.Mirror = mset.mirrorInfo()
-	} else if mset.hasSources() {
-		resp.StreamInfo.Sources = mset.sourcesInfo()
-	}
 
 	// Check if they have asked for subject details.
 	if subjects != _EMPTY_ {
diff --git a/server/jetstream_cluster.go b/server/jetstream_cluster.go
index ca48af1d..3819258e 100644
--- a/server/jetstream_cluster.go
+++ b/server/jetstream_cluster.go
@@ -1707,6 +1707,7 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 	// For migration tracking.
 	var migrating bool
 	var peerGroup peerMigrateType
+
 	var mmt *time.Ticker
 	var mmtc <-chan time.Time
 
@@ -1720,11 +1721,31 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 	stopMigrationMonitoring := func() {
 		if mmt != nil {
 			mmt.Stop()
-			mmtc = nil
+			mmt, mmtc = nil, nil
 		}
 	}
 	defer stopMigrationMonitoring()
 
+	// This is to optionally track when we are ready as a non-leader for direct access participation.
+	// Either direct or if we are a direct mirror, or both.
+	var dat *time.Ticker
+	var datc <-chan time.Time
+
+	startDirectAccessMonitoring := func() {
+		if dat == nil {
+			dat = time.NewTicker(1 * time.Second)
+			datc = dat.C
+		}
+	}
+
+	stopDirectMonitoring := func() {
+		if dat != nil {
+			dat.Stop()
+			dat, datc = nil, nil
+		}
+	}
+	defer stopDirectMonitoring()
+
 	// This is triggered during a scale up from 1 to clustered mode. We need the new followers to catchup,
 	// similar to how we trigger the catchup mechanism post a backup/restore. It's ok to do here and preferred
 	// over waiting to be elected, this just queues it up for the new members to see first and trigger the above
@@ -1787,6 +1808,9 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 				} else if n.NeedSnapshot() {
 					doSnapshot()
 				}
+				// Always cancel if this was running.
+				stopDirectMonitoring()
+
 			} else if n.GroupLeader() != noLeader {
 				js.setStreamAssignmentRecovering(sa)
 			}
@@ -1806,6 +1830,48 @@ func (js *jetStream) monitorStream(mset *stream, sa *streamAssignment, sendSnaps
 					stopMigrationMonitoring()
 				}
 			}
+
+			// Here we are checking if we are not the leader but we have been asked to allow
+			// direct access. We now allow non-leaders to participate in the queue group.
+			if !isLeader && mset != nil {
+				mset.mu.Lock()
+				// Check direct gets first.
+				if mset.cfg.AllowDirect {
+					if mset.directSub == nil && mset.isCurrent() {
+						mset.subscribeToDirect()
+					} else {
+						startDirectAccessMonitoring()
+					}
+				}
+				// Now check for mirror directs as well.
+				if mset.cfg.MirrorDirect {
+					if mset.mirror != nil && mset.mirror.dsub == nil && mset.isCurrent() {
+						mset.subscribeToMirrorDirect()
+					} else {
+						startDirectAccessMonitoring()
+					}
+				}
+				mset.mu.Unlock()
+			}
+
+		case <-datc:
+			mset.mu.Lock()
+			ad, md, current := mset.cfg.AllowDirect, mset.cfg.MirrorDirect, mset.isCurrent()
+			if !current {
+				mset.mu.Unlock()
+				continue
+			}
+			// We are current, cancel monitoring and create the direct subs as needed.
+			if ad {
+				mset.subscribeToDirect()
+			}
+			if md {
+				mset.subscribeToMirrorDirect()
+			}
+			mset.mu.Unlock()
+			// Stop monitoring.
+			stopDirectMonitoring()
+
 		case <-t.C:
 			doSnapshot()
 		case <-uch:
@@ -5984,6 +6050,15 @@ func (mset *stream) isCatchingUp() bool {
 	return mset.catchup
 }
 
+// Determine if a non-leader is current.
+// Lock should be held.
+func (mset *stream) isCurrent() bool {
+	if mset.node == nil {
+		return true
+	}
+	return mset.node.Current() && !mset.catchup
+}
+
 // Maximum requests for the whole server that can be in flight.
 const maxConcurrentSyncRequests = 8
 
diff --git a/server/stream.go b/server/stream.go
index bde3566a..94110685 100644
--- a/server/stream.go
+++ b/server/stream.go
@@ -72,8 +72,11 @@ type StreamConfig struct {
 	// AllowRollup allows messages to be placed into the system and purge
 	// all older messages using a special msg header.
 	AllowRollup bool `json:"allow_rollup_hdrs"`
-	// Allow higher peformance, direct access to get individual messages.
+
+	// Allow higher performance, direct access to get individual messages. E.g. KeyValue
 	AllowDirect bool `json:"allow_direct,omitempty"`
+	// Allow higher performance and unified direct access for mirrors as well.
+	MirrorDirect bool `json:"mirror_direct,omitempty"`
 }
 
 // RePublish is for republishing messages once committed to a stream.
@@ -231,6 +234,9 @@ type stream struct {
 	lqsent   time.Time
 	catchups map[string]uint64
 	uch      chan struct{}
+
+	// Direct get subscription.
+	directSub *subscription
 }
 
 type sourceInfo struct {
@@ -238,6 +244,7 @@ type sourceInfo struct {
 	iname string
 	cname string
 	sub   *subscription
+	dsub  *subscription
 	msgs  *ipQueue // of *inMsg
 	sseq  uint64
 	dseq  uint64
@@ -251,6 +258,12 @@ type sourceInfo struct {
 	wg    sync.WaitGroup
 }
 
+// For mirrors and direct get
+const (
+	dgetGroup          = "_zz_"
+	dgetCaughtUpThresh = 10
+)
+
 // Headers for published messages.
 const (
 	JSMsgId               = "Nats-Msg-Id"
@@ -1512,12 +1525,6 @@ func (mset *stream) isMirror() bool {
 	return mset.cfg.Mirror != nil
 }
 
-func (mset *stream) hasSources() bool {
-	mset.mu.RLock()
-	defer mset.mu.RUnlock()
-	return len(mset.sources) > 0
-}
-
 func (mset *stream) sourcesInfo() (sis []*StreamSourceInfo) {
 	mset.mu.RLock()
 	defer mset.mu.RUnlock()
@@ -1840,6 +1847,15 @@ func (mset *stream) processInboundMirrorMsg(m *inMsg) bool {
 		mset.mirror.lag = pending - 1
 	}
 
+	// Check if we allow mirror direct here. If so check they we have mostly caught up.
+	// The reason we do not require 0 is if the source is active we may always be slightly behind.
+	if mset.cfg.MirrorDirect && mset.mirror.dsub == nil && pending < dgetCaughtUpThresh {
+		if err := mset.subscribeToMirrorDirect(); err != nil {
+			// Disable since we had problems above.
+			mset.cfg.MirrorDirect = false
+		}
+	}
+
 	js, stype := mset.js, mset.cfg.Storage
 	mset.mu.Unlock()
 
@@ -2219,6 +2235,11 @@ func (mset *stream) cancelSourceInfo(si *sourceInfo) {
 		mset.unsubscribe(si.sub)
 		si.sub = nil
 	}
+	// In case we had a mirror direct subscription.
+	if si.dsub != nil {
+		mset.unsubscribe(si.dsub)
+		si.dsub = nil
+	}
 	mset.removeInternalConsumer(si)
 	if si.qch != nil {
 		close(si.qch)
@@ -2843,9 +2864,9 @@ func (mset *stream) subscribeToStream() error {
 			return err
 		}
 	}
+	// Check for direct get access.
 	if mset.cfg.AllowDirect {
-		dsubj := fmt.Sprintf(JSDirectMsgGetT, mset.cfg.Name)
-		if _, err := mset.subscribeInternal(dsubj, mset.processDirectGetRequest); err != nil {
+		if err := mset.subscribeToDirect(); err != nil {
 			return err
 		}
 	}
@@ -2854,6 +2875,36 @@ func (mset *stream) subscribeToStream() error {
 	return nil
 }
 
+// Lock should be held.
+func (mset *stream) subscribeToDirect() error {
+	if mset.directSub != nil {
+		return nil
+	}
+	dsubj := fmt.Sprintf(JSDirectMsgGetT, mset.cfg.Name)
+	// We will make this listen on a queue group by default, which can allow mirrors to participate on opt-in basis.
+	if sub, err := mset.queueSubscribeInternal(dsubj, dgetGroup, mset.processDirectGetRequest); err != nil {
+		return err
+	} else {
+		mset.directSub = sub
+	}
+	return nil
+}
+
+// Lock should be held.
+func (mset *stream) subscribeToMirrorDirect() error {
+	if mset.mirror == nil || mset.mirror.dsub != nil {
+		return nil
+	}
+	dsubj := fmt.Sprintf(JSDirectMsgGetT, mset.mirror.name)
+	// We will make this listen on a queue group by default, which can allow mirrors to participate on opt-in basis.
+	if sub, err := mset.queueSubscribeInternal(dsubj, dgetGroup, mset.processDirectGetRequest); err != nil {
+		return err
+	} else {
+		mset.mirror.dsub = sub
+	}
+	return nil
+}
+
 // Stop our source consumers.
 // Lock should be held.
 func (mset *stream) stopSourceConsumers() {
@@ -2881,12 +2932,15 @@ func (mset *stream) unsubscribeToStream() error {
 		mset.mirror = nil
 	}
 
-	if len(mset.cfg.Sources) > 0 {
+	if len(mset.sources) > 0 {
 		mset.stopSourceConsumers()
 	}
 
 	// In case we had a direct get subscription.
-	mset.unsubscribeInternal(fmt.Sprintf(JSDirectMsgGetT, mset.cfg.Name))
+	if mset.directSub != nil {
+		mset.unsubscribe(mset.directSub)
+		mset.directSub = nil
+	}
 
 	mset.active = false
 	return nil
@@ -2915,6 +2969,22 @@ func (mset *stream) subscribeInternalUnlocked(subject string, cb msgHandler) (*s
 	return mset.subscribeInternal(subject, cb)
 }
 
+// Lock should be held.
+func (mset *stream) queueSubscribeInternal(subject, group string, cb msgHandler) (*subscription, error) {
+	c := mset.client
+	if c == nil {
+		return nil, fmt.Errorf("invalid stream")
+	}
+	if cb == nil {
+		return nil, fmt.Errorf("undefined message handler")
+	}
+
+	mset.sid++
+
+	// Now create the subscription
+	return c.processSub([]byte(subject), []byte(group), []byte(strconv.Itoa(mset.sid)), cb, false)
+}
+
 // This will unsubscribe us from the exact subject given.
 // We do not currently track the subs so do not have the sid.
 // This should be called only on an update.


Test patch:
diff --git a/server/jetstream_helpers_test.go b/server/jetstream_helpers_test.go
index fd87e08b..3124bfa3 100644
--- a/server/jetstream_helpers_test.go
+++ b/server/jetstream_helpers_test.go
@@ -17,15 +17,18 @@
 package server
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"math/rand"
+	"net"
 	"strings"
 	"testing"
 	"time"
 
 	"github.com/nats-io/nats.go"
+	"golang.org/x/time/rate"
 )
 
 // Support functions
@@ -1254,11 +1257,13 @@ func (c *cluster) waitOnClusterReadyWithNumPeers(numPeersExpected int) {
 	}
 
 	if leader == nil {
+		c.shutdown()
 		c.t.Fatalf("Failed to elect a meta-leader")
 	}
 
 	peersSeen := len(leader.JetStreamClusterPeers())
 	c.shutdown()
+
 	if leader == nil {
 		c.t.Fatalf("Expected a cluster leader and fully formed cluster, no leader")
 	} else {
@@ -1365,6 +1370,24 @@ func addStream(t *testing.T, nc *nats.Conn, cfg *StreamConfig) *StreamInfo {
 	return resp.StreamInfo
 }
 
+func updateStream(t *testing.T, nc *nats.Conn, cfg *StreamConfig) *StreamInfo {
+	t.Helper()
+	req, err := json.Marshal(cfg)
+	require_NoError(t, err)
+	rmsg, err := nc.Request(fmt.Sprintf(JSApiStreamUpdateT, cfg.Name), req, time.Second)
+	require_NoError(t, err)
+	var resp JSApiStreamCreateResponse
+	err = json.Unmarshal(rmsg.Data, &resp)
+	require_NoError(t, err)
+	if resp.Type != JSApiStreamUpdateResponseType {
+		t.Fatalf("Invalid response type %s expected %s", resp.Type, JSApiStreamUpdateResponseType)
+	}
+	if resp.Error != nil {
+		t.Fatalf("Unexpected error: %+v", resp.Error)
+	}
+	return resp.StreamInfo
+}
+
 // setInActiveDeleteThreshold sets the delete threshold for how long to wait
 // before deleting an inactive consumer.
 func (o *consumer) setInActiveDeleteThreshold(dthresh time.Duration) error {
@@ -1380,3 +1403,79 @@ func (o *consumer) setInActiveDeleteThreshold(dthresh time.Duration) error {
 	}
 	return nil
 }
+
+// Net Proxy - For introducing RTT and BW constraints.
+type netProxy struct {
+	listener net.Listener
+	conns    []net.Conn
+	url      string
+}
+
+func newNetProxy(rtt time.Duration, upRate, downRate int, serverURL string) *netProxy {
+	hp := net.JoinHostPort("127.0.0.1", "0")
+	l, e := net.Listen("tcp", hp)
+	if e != nil {
+		panic(fmt.Sprintf("Error listening on port: %s, %q", hp, e))
+	}
+	port := l.Addr().(*net.TCPAddr).Port
+	proxy := &netProxy{listener: l}
+	go func() {
+		client, err := l.Accept()
+		if err != nil {
+			return
+		}
+		server, err := net.DialTimeout("tcp", serverURL[7:], time.Second)
+		if err != nil {
+			panic("Can't connect to NATS server")
+		}
+		proxy.conns = append(proxy.conns, client, server)
+		go proxy.loop(rtt, upRate, client, server)
+		go proxy.loop(rtt, downRate, server, client)
+	}()
+	proxy.url = fmt.Sprintf("nats://127.0.0.1:%d", port)
+	return proxy
+}
+
+func (np *netProxy) clientURL() string {
+	return np.url
+}
+
+func (np *netProxy) loop(rtt time.Duration, tbw int, r, w net.Conn) {
+	delay := rtt / 2
+	const rbl = 8192
+	var buf [rbl]byte
+	ctx := context.Background()
+
+	rl := rate.NewLimiter(rate.Limit(tbw), rbl)
+
+	for fr := true; ; {
+		sr := time.Now()
+		n, err := r.Read(buf[:])
+		if err != nil {
+			return
+		}
+		// RTT delays
+		if fr || time.Since(sr) > 2*time.Millisecond {
+			fr = false
+			if delay > 0 {
+				time.Sleep(delay)
+			}
+		}
+		if err := rl.WaitN(ctx, n); err != nil {
+			return
+		}
+		if _, err = w.Write(buf[:n]); err != nil {
+			return
+		}
+	}
+}
+
+func (np *netProxy) stop() {
+	if np.listener != nil {
+		np.listener.Close()
+		np.listener = nil
+		for _, c := range np.conns {
+			c.Close()
+		}
+	}
+}
diff --git a/server/jetstream_super_cluster_test.go b/server/jetstream_super_cluster_test.go
index 036fece7..64ed1580 100644
--- a/server/jetstream_super_cluster_test.go
+++ b/server/jetstream_super_cluster_test.go
@@ -2483,3 +2483,112 @@ func TestJetStreamSuperClusterStateOnRestartPreventsConsumerRecovery(t *testing.
 		t.Fatalf("Consumer was not properly restarted")
 	}
 }
+
+// We allow mirrors to opt-in to direct get in a distributed queue group.
+func TestJetStreamSuperClusterStreamDirectGetMirrorQueueGroup(t *testing.T) {
+	sc := createJetStreamTaggedSuperCluster(t)
+	defer sc.shutdown()
+
+	nc, js := jsClientConnect(t, sc.randomServer())
+	defer nc.Close()
+
+	// C1
+	// Do by hand for now.
+	cfg := &StreamConfig{
+		Name:        "SOURCE",
+		Subjects:    []string{"kv.>"},
+		MaxMsgsPer:  1,
+		Placement:   &Placement{Tags: []string{"cloud:aws", "country:us"}},
+		AllowDirect: true,
+		Replicas:    3,
+		Storage:     MemoryStorage,
+	}
+	addStream(t, nc, cfg)
+
+	num := 100
+	for i := 0; i < num; i++ {
+		js.PublishAsync(fmt.Sprintf("kv.%d", i), []byte("VAL"))
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
+
+	// C2
+	cfg = &StreamConfig{
+		Name:         "M1",
+		Mirror:       &StreamSource{Name: "SOURCE"},
+		Placement:    &Placement{Tags: []string{"cloud:gcp", "country:uk"}},
+		MirrorDirect: true,
+		Storage:      MemoryStorage,
+	}
+	addStream(t, nc, cfg)
+
+	// C3 (clustered)
+	cfg = &StreamConfig{
+		Name:         "M2",
+		Mirror:       &StreamSource{Name: "SOURCE"},
+		Replicas:     3,
+		Placement:    &Placement{Tags: []string{"country:jp"}},
+		MirrorDirect: true,
+		Storage:      MemoryStorage,
+	}
+	addStream(t, nc, cfg)
+
+	checkFor(t, 5*time.Second, 100*time.Millisecond, func() error {
+		si, err := js.StreamInfo("M2")
+		require_NoError(t, err)
+		if si.State.Msgs != uint64(num) {
+			return fmt.Errorf("Expected %d msgs, got state: %d", num, si.State.Msgs)
+		}
+		return nil
+	})
+
+	// Since last one was an R3, check and wait for the direct subscription.
+	checkFor(t, 2*time.Second, 100*time.Millisecond, func() error {
+		sl := sc.clusterForName("C3").streamLeader("$G", "M2")
+		if mset, err := sl.GlobalAccount().lookupStream("M2"); err == nil {
+			mset.mu.RLock()
+			ok := mset.mirror.dsub != nil
+			mset.mu.RUnlock()
+			if ok {
+				return nil
+			}
+		}
+		return fmt.Errorf("No dsub yet")
+	})
+
+	// Always do a direct get to the source, but check that we are getting answers from the mirrors when connected to their cluster.
+	getSubj := fmt.Sprintf(JSDirectMsgGetT, "SOURCE")
+	req := []byte(`{"last_by_subj":"kv.22"}`)
+	getMsg := func(c *nats.Conn) *nats.Msg {
+		m, err := c.Request(getSubj, req, time.Second)
+		require_NoError(t, err)
+		require_True(t, string(m.Data) == "VAL")
+		require_True(t, m.Header.Get(JSSequence) == "23")
+		require_True(t, m.Header.Get(JSSubject) == "kv.22")
+		return m
+	}
+
+	// C1 -> SOURCE
+	nc, _ = jsClientConnect(t, sc.clusterForName("C1").randomServer())
+	defer nc.Close()
+
+	m := getMsg(nc)
+	require_True(t, m.Header.Get(JSStream) == "SOURCE")
+
+	// C2 -> M1
+	nc, _ = jsClientConnect(t, sc.clusterForName("C2").randomServer())
+	defer nc.Close()
+
+	m = getMsg(nc)
+	require_True(t, m.Header.Get(JSStream) == "M1")
+
+	// C3 -> M2
+	nc, _ = jsClientConnect(t, sc.clusterForName("C3").randomServer())
+	defer nc.Close()
+
+	m = getMsg(nc)
+	require_True(t, m.Header.Get(JSStream) == "M2")
+}
diff --git a/server/monitor_test.go b/server/monitor_test.go
index e4b6bc30..9a0db9ed 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -3984,17 +3984,10 @@ func TestMonitorAuthorizedUsers(t *testing.T) {
 // Helper function to check that a JS cluster is formed
 func checkForJSClusterUp(t *testing.T, servers ...*Server) {
 	t.Helper()
-	checkFor(t, 10*time.Second, 100*time.Millisecond, func() error {
-		for _, s := range servers {
-			if !s.JetStreamEnabled() {
-				return fmt.Errorf("jetstream not enabled")
-			}
-			if !s.JetStreamIsCurrent() {
-				return fmt.Errorf("jetstream not current")
-			}
-		}
-		return nil
-	})
+	// We will use the other JetStream helpers here.
+	c := &cluster{t: t, servers: servers}
+	c.checkClusterFormed()
+	c.waitOnClusterReady()
 }
 
 func TestMonitorJszNonJszServer(t *testing.T) {
diff --git a/server/mqtt_test.go b/server/mqtt_test.go
index 91decee0..f7fce85a 100644
--- a/server/mqtt_test.go
+++ b/server/mqtt_test.go
@@ -3167,8 +3167,10 @@ func TestMQTTLeafnodeWithoutJSToClusterWithJSNoSharedSysAcc(t *testing.T) {
 			for _, s := range cluster {
 				if s.JetStreamIsLeader() {
 					// Need to wait for usage updates now to propagate to meta leader.
-					time.Sleep(250 * time.Millisecond)
-					return nil
+					if len(s.JetStreamClusterPeers()) == len(cluster) {
+						time.Sleep(100 * time.Millisecond)
+						return nil
+					}
 				}
 			}
 			return fmt.Errorf("no leader yet")
diff --git a/server/norace_test.go b/server/norace_test.go
index 3a642d84..ad173f9b 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -20,7 +20,6 @@ import (
 	"bufio"
 	"bytes"
 	"compress/gzip"
-	"context"
 	"encoding/binary"
 	"encoding/json"
 	"fmt"
@@ -48,7 +47,6 @@ import (
 	"github.com/nats-io/jwt/v2"
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nkeys"
-	"golang.org/x/time/rate"
 )
 
 // IMPORTANT: Tests in this file are not executed when running with the -race flag.
@@ -5171,6 +5169,7 @@ func TestNoRaceJetStreamClusterInterestPullConsumerStreamLimitBug(t *testing.T)
 					return
 				}
 			case <-qch:
+				pt.Stop()
 				return
 			}
 		}
@@ -5230,78 +5229,107 @@ func TestNoRaceJetStreamClusterInterestPullConsumerStreamLimitBug(t *testing.T)
 	}
 }
 
-// Net Proxy - For introducing RTT and BW constraints.
-type netProxy struct {
-	listener net.Listener
-	conns    []net.Conn
-	url      string
-}
+// Test that all peers have the direct access subs that participate in a queue group,
+// but only when they are current and ready. So we will start with R1, add in messages
+// then scale up while also still adding messages.
+func TestNoRaceJetStreamClusterDirectAccessAllPeersSubs(t *testing.T) {
+	c := createJetStreamClusterExplicit(t, "JSC", 3)
+	defer c.shutdown()
 
-func newNetProxy(rtt time.Duration, upRate, downRate int, serverURL string) *netProxy {
-	hp := net.JoinHostPort("127.0.0.1", "0")
-	l, e := net.Listen("tcp", hp)
-	if e != nil {
-		panic(fmt.Sprintf("Error listening on port: %s, %q", hp, e))
+	nc, js := jsClientConnect(t, c.randomServer())
+	defer nc.Close()
+
+	// Start as R1
+	cfg := &StreamConfig{
+		Name:        "TEST",
+		Subjects:    []string{"kv.>"},
+		MaxMsgsPer:  10,
+		AllowDirect: true,
+		Replicas:    1,
+		Storage:     FileStorage,
 	}
-	port := l.Addr().(*net.TCPAddr).Port
-	proxy := &netProxy{listener: l}
-	go func() {
-		client, err := l.Accept()
-		if err != nil {
-			return
-		}
-		server, err := net.DialTimeout("tcp", serverURL[7:], time.Second)
-		if err != nil {
-			panic("Can't connect to NATS server")
-		}
-		proxy.conns = append(proxy.conns, client, server)
-		go proxy.loop(rtt, upRate, client, server)
-		go proxy.loop(rtt, downRate, server, client)
-	}()
-	proxy.url = fmt.Sprintf("nats://127.0.0.1:%d", port)
-	return proxy
-}
+	addStream(t, nc, cfg)
 
-func (np *netProxy) clientURL() string {
-	return np.url
-}
+	// Seed with enough messages to start then we will scale up while still adding more messages.
+	num, msg := 1000, bytes.Repeat([]byte("XYZ"), 64)
+	for i := 0; i < num; i++ {
+		js.PublishAsync(fmt.Sprintf("kv.%d", i), msg)
+	}
+	select {
+	case <-js.PublishAsyncComplete():
+	case <-time.After(5 * time.Second):
+		t.Fatalf("Did not receive completion signal")
+	}
 
-func (np *netProxy) loop(rtt time.Duration, tbw int, r, w net.Conn) {
-	delay := rtt / 2
-	const rbl = 8192
-	var buf [rbl]byte
-	ctx := context.Background()
+	getSubj := fmt.Sprintf(JSDirectMsgGetT, "TEST")
+	getMsg := func(key string) *nats.Msg {
+		req := []byte(fmt.Sprintf(`{"last_by_subj":%q}`, key))
+		m, err := nc.Request(getSubj, req, time.Second)
+		require_NoError(t, err)
+		require_True(t, m.Header.Get(JSSubject) == key)
+		return m
+	}
 
-	rl := rate.NewLimiter(rate.Limit(tbw), rbl)
+	// Just make sure we can succeed here.
+	getMsg("kv.22")
 
-	for fr := true; ; {
-		sr := time.Now()
-		n, err := r.Read(buf[:])
-		if err != nil {
-			return
-		}
-		// RTT delays
-		if fr || time.Since(sr) > 2*time.Millisecond {
-			fr = false
-			if delay > 0 {
-				time.Sleep(delay)
+	// Now crank up a go routine to continue sending more messages.
+	qch := make(chan bool)
+	var wg sync.WaitGroup
+
+	for i := 0; i < 5; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			nc, _ := jsClientConnect(t, c.randomServer())
+			js, _ := nc.JetStream(nats.MaxWait(500 * time.Millisecond))
+			defer nc.Close()
+			for {
+				pt := time.NewTimer(time.Duration(time.Millisecond))
+				select {
+				case <-pt.C:
+					js.Publish(fmt.Sprintf("kv.%d", rand.Intn(1000)), msg)
+				case <-qch:
+					pt.Stop()
+					return
+				}
 			}
-		}
-		if err := rl.WaitN(ctx, n); err != nil {
-			return
-		}
-		if _, err = w.Write(buf[:n]); err != nil {
-			return
-		}
+		}()
 	}
-}
 
-func (np *netProxy) stop() {
-	if np.listener != nil {
-		np.listener.Close()
-		np.listener = nil
-		for _, c := range np.conns {
-			c.Close()
-		}
+	time.Sleep(100 * time.Millisecond)
+
+	// Now let's scale up to an R3.
+	cfg.Replicas = 3
+	updateStream(t, nc, cfg)
+	c.waitOnStreamLeader("$G", "TEST")
+
+	// For each non-leader check that the direct sub fires up.
+	// We just test all, the leader will already have a directSub.
+	for _, s := range c.servers {
+		mset, err := s.GlobalAccount().lookupStream("TEST")
+		require_NoError(t, err)
+		checkFor(t, 10*time.Second, 500*time.Millisecond, func() error {
+			mset.mu.RLock()
+			ok := mset.directSub != nil
+			mset.mu.RUnlock()
+			if ok {
+				return nil
+			}
+			return fmt.Errorf("No directSub yet")
+		})
+	}
+
+	close(qch)
+	wg.Wait()
+
+	// Just make sure we can succeed here.
+	getMsg("kv.22")
+
+	si, err := js.StreamInfo("TEST")
+	require_NoError(t, err)
+
+	if si.State.Msgs == uint64(num) {
+		t.Fatalf("Expected to see messages increase, got %d", si.State.Msgs)
 	}
 }