Repository: nats-io/nats-server (go)

Issue:
Regression in FilterSubjects Handling with DeliverAll/ReplayInstant between v2.10.25 and v2.10.26
### Observed behavior

A regression appears to have been introduced between `nats-server` v2.10.25 and v2.10.26 that affects how `FilterSubjects` are handled for consumers configured with `DeliverPolicy: DeliverAllPolicy` and `ReplayPolicy: ReplayInstantPolicy`.

When a consumer uses `FilterSubjects` containing both an exact subject and a related wildcard (e.g., `["stream.A", "stream.A.>"]`), it often fails to receive the expected historical messages from the stream upon startup **if and only if there are no messages currently in the stream matching the wildcard filter (`stream.A.>`) at the time the consumer starts.**

If at least one message matching the wildcard filter (e.g., `stream.A.child`) exists in the stream *before* the consumer starts, the consumer behaves correctly and receives the full history for *both* filters. If only messages matching the exact subject (`stream.A`) exist, the bug manifests, and historical messages are often not delivered.

Newly published messages matching the filters are received correctly after startup, regardless of the initial state.

This behavior is inconsistent when the bug manifests; occasionally (though rarely), the consumer *does* receive the full history even without pre-existing wildcard matches, making it difficult to reliably reproduce without controlling the stream state precisely.

#### Configuration

* **Stream:** Configured with a `LimitsPolicy` (or other policy) to retain messages. The stream captures subjects matching `stream.>`. Subjects like `stream.A` and `stream.A.child` are published to it.
* **Consumer:**
    * `DeliverPolicy: DeliverAllPolicy`
    * `ReplayPolicy: ReplayInstantPolicy`
    * `AckPolicy: AckExplicitPolicy` (or potentially others)
    * `FilterSubjects: ["stream.A", "stream.A.>"]` (Problematic case)

#### Observed Behavior

* **Condition:** No messages matching `stream.A.>` exist in the stream. Only messages matching `stream.A` exist.
    * **With `FilterSubjects: ["stream.A", "stream.A.>"]`:**
        * Most of the time, the consumer starts but receives **zero** historical messages. It only receives new messages published *after* it started.
        * Occasionally, the consumer starts and correctly receives the full history matching `stream.A`.
* **Condition:** At least one message matching `stream.A.>` (e.g., `stream.A.child`) exists in the stream (along with messages matching `stream.A`).
    * **With `FilterSubjects: ["stream.A", "stream.A.>"]`:**
        * The consumer reliably starts and receives the full history matching both `stream.A` and `stream.A.>` instantly, as expected.
* **With `FilterSubjects: ["stream.A"]` (regardless of stream content):**
    * The consumer reliably starts and receives the full history matching `stream.A` instantly, as expected.

#### Versions
This issue was reproduced locally by testing various `nats-server` versions:

* **v2.10.25:** Works as expected (full history delivered reliably regardless of initial stream content).
* **v2.10.26:** Exhibits the bug under the conditions described above.
* **v2.10.27 - v2.11.x:** Also exhibit the bug under the conditions described above.

This strongly suggests a change introduced between v2.10.25 and v2.10.26 is responsible.


### Expected behavior

When a consumer starts with `DeliverAllPolicy`, `ReplayInstantPolicy`, and `FilterSubjects: ["stream.A", "stream.A.>"]`, it should reliably receive *all* historical messages present in the stream whose subjects match *either* `stream.A` *or* `stream.A.>`, regardless of whether messages matching `stream.A.>` were present at the exact moment the consumer started.


### Server and client version

```
nats-server  v2.10.26+
nats.go      v1.39.1
```

### Host environment

Issue first detected from inconsistencies in k8s (via nats helm chart) after v2.11 upgrade from v2.10.25

Replicated locally on Apple MacBook Pro M4

### Steps to reproduce

1.  Set up `nats-server` (v2.10.26 or later).
2.  Create a stream e.g. `mystream` configured to retain messages on disk and capture subjects matching `stream.>`.
3.  **Scenario A (Bug Manifests):**
    * Publish several messages, **only** with subject `stream.A`. Ensure **no** messages exist with subject `stream.A.child` (or any other subject matching `stream.A.>`).
    * Create and run a consumer application connecting to `mystream` with:
        * `DeliverPolicy: DeliverAllPolicy`
        * `ReplayPolicy: ReplayInstantPolicy`
        * `FilterSubjects: ["stream.A", "stream.A.>"]`
    * Observe that the consumer often fails to log the historical `stream.A` messages immediately upon startup. Repeat consumer startup multiple times to observe inconsistency.
4.  **Scenario B (Works Correctly):**
    * Publish several messages with subject `stream.A`.
    * Publish **at least one** message with subject `stream.A.child`.
    * Create and run the *same* consumer application as in Step 3b.
    * Observe that the consumer now reliably logs *all* historical messages (`stream.A` and `stream.A.child`) immediately upon startup.
5.  **Verification:** Modify the consumer to use `FilterSubjects: ["stream.A"]` and repeat Scenario A (Step 3a). Observe that history for `stream.A` is now delivered reliably even without `stream.A.child` messages present.
6.  **Regression Check:** Optionally, downgrade `nats-server` to v2.10.25 and repeat Scenario A (Step 3a) with the original `FilterSubjects: ["stream.A", "stream.A.>"]` to confirm it works correctly even without pre-existing `stream.A.child` messages.

Pull request:
Fix subject tree intersection to match literals when followed by additional wildcards
The subject tree intersection could miss non-wildcard matches if they were followed up by further extended wildcards in other sublist filters. This is now fixed.

Fixes #6824.

Signed-off-by: Neil Twigg <neil@nats.io>
Co-authored-by: Maurice van Veen <github@mauricevanveen.com>

Patch:
diff --git a/server/raft.go b/server/raft.go
index 95351ac5..16a1b3cd 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -154,7 +154,7 @@ type raft struct {
 	qn    int             // Number of nodes needed to establish quorum
 	peers map[string]*lps // Other peers in the Raft group
 
-	removed map[string]time.Time           // Peers that were removed from the group
+	removed map[string]struct{}            // Peers that were removed from the group
 	acks    map[uint64]map[string]struct{} // Append entry responses/acks, map of entry index -> peer ID
 	pae     map[uint64]*appendEntry        // Pending append entries
 
@@ -253,7 +253,6 @@ const (
 	lostQuorumIntervalDefault      = hbIntervalDefault * 10 // 10 seconds
 	lostQuorumCheckIntervalDefault = hbIntervalDefault * 10 // 10 seconds
 	observerModeIntervalDefault    = 48 * time.Hour
-	peerRemoveTimeoutDefault       = 5 * time.Minute
 )
 
 var (
@@ -265,7 +264,6 @@ var (
 	lostQuorumInterval   = lostQuorumIntervalDefault
 	lostQuorumCheck      = lostQuorumCheckIntervalDefault
 	observerModeInterval = observerModeIntervalDefault
-	peerRemoveTimeout    = peerRemoveTimeoutDefault
 )
 
 type RaftConfig struct {
@@ -894,9 +892,9 @@ func (n *raft) ProposeAddPeer(peer string) error {
 func (n *raft) doRemovePeerAsLeader(peer string) {
 	n.Lock()
 	if n.removed == nil {
-		n.removed = map[string]time.Time{}
+		n.removed = map[string]struct{}{}
 	}
-	n.removed[peer] = time.Now()
+	n.removed[peer] = struct{}{}
 	if _, ok := n.peers[peer]; ok {
 		delete(n.peers, peer)
 		// We should decrease our cluster size since we are tracking this peer and the peer is most likely already gone.
@@ -2970,9 +2968,9 @@ func (n *raft) applyCommit(index uint64) error {
 
 			// Make sure we have our removed map.
 			if n.removed == nil {
-				n.removed = make(map[string]time.Time)
+				n.removed = make(map[string]struct{})
 			}
-			n.removed[peer] = time.Now()
+			n.removed[peer] = struct{}{}
 
 			if _, ok := n.peers[peer]; ok {
 				delete(n.peers, peer)
@@ -3087,13 +3085,8 @@ func (n *raft) adjustClusterSizeAndQuorum() {
 func (n *raft) trackPeer(peer string) error {
 	n.Lock()
 	var needPeerAdd, isRemoved bool
-	var rts time.Time
 	if n.removed != nil {
-		rts, isRemoved = n.removed[peer]
-		// Removed peers can rejoin after timeout.
-		if isRemoved && time.Since(rts) >= peerRemoveTimeout {
-			isRemoved = false
-		}
+		_, isRemoved = n.removed[peer]
 	}
 	if n.State() == Leader {
 		if lp, ok := n.peers[peer]; !ok || !lp.kp {
diff --git a/server/sublist.go b/server/sublist.go
index 004150aa..985cc351 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -1747,13 +1747,20 @@ func IntersectStree[T any](st *stree.SubjectTree[T], sl *Sublist, cb func(subj [
 }
 
 func intersectStree[T any](st *stree.SubjectTree[T], r *level, subj []byte, cb func(subj []byte, entry *T)) {
+	// This level could potentially match literals, despite being followed up by
+	// additional wildcards. For literals we can use Find since it is considerably
+	// faster. Then we can carry on checking for further matches in the usual way.
+	wc := subjectHasWildcard(bytesToString(subj))
+	if !wc {
+		if e, ok := st.Find(subj); ok {
+			cb(subj, e)
+		}
+	}
 	if r.numNodes() == 0 {
-		// For wildcards we can't avoid Match, but if it's a literal subject at
-		// this point, using Find is considerably cheaper.
-		if subjectHasWildcard(bytesToString(subj)) {
+		// No further recursions to be made at this point but there's still a wildcard
+		// to match, so let the subject tree work it out.
+		if wc {
 			st.Match(subj, cb)
-		} else if e, ok := st.Find(subj); ok {
-			cb(subj, e)
 		}
 		return
 	}


Test patch:
diff --git a/server/jetstream_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index 5ef1d7c9..aa2bdaef 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -3776,89 +3776,6 @@ func TestJetStreamClusterPeerRemovalAndStreamReassignmentWithoutSpace(t *testing
 	streamCurrent(2)
 }
 
-func TestJetStreamClusterPeerRemovalAndServerBroughtBack(t *testing.T) {
-	// Speed up for this test
-	peerRemoveTimeout = 2 * time.Second
-	defer func() {
-		peerRemoveTimeout = peerRemoveTimeoutDefault
-	}()
-
-	c := createJetStreamClusterExplicit(t, "R5S", 5)
-	defer c.shutdown()
-
-	// Client based API
-	ml := c.leader()
-	nc, err := nats.Connect(ml.ClientURL(), nats.UserInfo("admin", "s3cr3t!"))
-	if err != nil {
-		t.Fatalf("Failed to create system client: %v", err)
-	}
-	defer nc.Close()
-
-	getPeersCount := func() int {
-		js := ml.getJetStream()
-		if js == nil {
-			return 0
-		}
-		js.mu.RLock()
-		defer js.mu.RUnlock()
-
-		cc := js.cluster
-		if !cc.isLeader() || cc.meta == nil {
-			return 0
-		}
-		return len(cc.meta.Peers())
-	}
-
-	checkFor(t, 2*time.Second, 250*time.Millisecond, func() error {
-		if l := getPeersCount(); l != 5 {
-			return fmt.Errorf("expected 5 peers, got %d", l)
-		}
-		return nil
-	})
-
-	// Shutdown server first.
-	rs := c.randomNonLeader()
-	rs.Shutdown()
-
-	// Peers should still remain the same, even if one server is shut down.
-	checkFor(t, 2*time.Second, 250*time.Millisecond, func() error {
-		if l := getPeersCount(); l != 5 {
-			return fmt.Errorf("expected 5 peers, got %d", l)
-		}
-		return nil
-	})
-
-	// Peer-remove after shutdown.
-	req := &JSApiMetaServerRemoveRequest{Server: rs.Name()}
-	jsreq, err := json.Marshal(req)
-	require_NoError(t, err)
-	rmsg, err := nc.Request(JSApiRemoveServer, jsreq, time.Second)
-	require_NoError(t, err)
-
-	var resp JSApiMetaServerRemoveResponse
-	require_NoError(t, json.Unmarshal(rmsg.Data, &resp))
-	if resp.Error != nil {
-		t.Fatalf("Unexpected error: %+v", resp.Error)
-	}
-
-	// Peer should be removed.
-	checkFor(t, 2*time.Second, 250*time.Millisecond, func() error {
-		if l := getPeersCount(); l != 4 {
-			return fmt.Errorf("expected 4 peers, got %d", l)
-		}
-		return nil
-	})
-
-	// Bringing back the server should re-add to peers after peer-remove timeout.
-	c.restartServer(rs)
-	checkFor(t, 5*time.Second, 250*time.Millisecond, func() error {
-		if l := getPeersCount(); l != 5 {
-			return fmt.Errorf("expected 5 peers, got %d", l)
-		}
-		return nil
-	})
-}
-
 func TestJetStreamClusterPeerExclusionTag(t *testing.T) {
 	c := createJetStreamClusterWithTemplateAndModHook(t, jsClusterTempl, "C", 3,
 		func(serverName, clusterName, storeDir, conf string) string {
diff --git a/server/jetstream_consumer_test.go b/server/jetstream_consumer_test.go
index 1a1f4fa2..48465091 100644
--- a/server/jetstream_consumer_test.go
+++ b/server/jetstream_consumer_test.go
@@ -17,6 +17,7 @@ package server
 
 import (
 	"bytes"
+	"context"
 	crand "crypto/rand"
 	"encoding/base64"
 	"encoding/json"
@@ -34,6 +35,8 @@ import (
 	"testing"
 	"time"
 
+	"github.com/nats-io/nats.go/jetstream"
+
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nuid"
 )
@@ -9473,3 +9476,58 @@ func TestJetStreamConsumerPullMaxBytes(t *testing.T) {
 	checkHeader(m, badReq)
 	checkSubsPending(t, sub, 0)
 }
+
+// https://github.com/nats-io/nats-server/issues/6824
+func TestJetStreamConsumerDeliverAllOverlappingFilterSubjects(t *testing.T) {
+	s := RunBasicJetStreamServer(t)
+	defer s.Shutdown()
+
+	nc, js := jsClientConnectNewAPI(t, s)
+	defer nc.Close()
+
+	ctx := context.Background()
+	_, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
+		Name:     "TEST",
+		Subjects: []string{"stream.>"},
+	})
+	require_NoError(t, err)
+
+	publishMessageCount := 10
+	for i := 0; i < publishMessageCount; i++ {
+		_, err = js.Publish(ctx, "stream.A", nil)
+		require_NoError(t, err)
+	}
+
+	// Create consumer
+	consumer, err := js.CreateOrUpdateConsumer(ctx, "TEST", jetstream.ConsumerConfig{
+		DeliverPolicy: jetstream.DeliverAllPolicy,
+		FilterSubjects: []string{
+			"stream.A",
+			"stream.A.>",
+		},
+	})
+	require_NoError(t, err)
+
+	messages := make(chan jetstream.Msg)
+	cc, err := consumer.Consume(func(msg jetstream.Msg) {
+		messages <- msg
+		msg.Ack()
+	})
+	require_NoError(t, err)
+	defer cc.Drain()
+
+	var count = 0
+	for {
+		if count == publishMessageCount {
+			// All messages received.
+			return
+		}
+		select {
+		case <-messages:
+			count++
+		case <-time.After(2 * time.Second):
+			t.Errorf("Timeout reached, %d messages received. Exiting.", count)
+			return
+		}
+	}
+}
diff --git a/server/sublist_test.go b/server/sublist_test.go
index 5e8145c5..e685894b 100644
--- a/server/sublist_test.go
+++ b/server/sublist_test.go
@@ -1990,8 +1990,11 @@ func TestSublistInterestBasedIntersection(t *testing.T) {
 	st.Insert([]byte("one.two.six"), struct{}{})
 	st.Insert([]byte("one.two.seven"), struct{}{})
 	st.Insert([]byte("eight.nine"), struct{}{})
+	st.Insert([]byte("stream.A"), struct{}{})
+	st.Insert([]byte("stream.A.child"), struct{}{})
 
 	require_NoDuplicates := func(t *testing.T, got map[string]int) {
+		t.Helper()
 		for _, c := range got {
 			require_Equal(t, c, 1)
 		}
@@ -2044,7 +2047,7 @@ func TestSublistInterestBasedIntersection(t *testing.T) {
 		IntersectStree(st, sl, func(subj []byte, entry *struct{}) {
 			got[string(subj)]++
 		})
-		require_Len(t, len(got), 5)
+		require_Len(t, len(got), 7)
 		require_NoDuplicates(t, got)
 	})
 
@@ -2071,6 +2074,18 @@ func TestSublistInterestBasedIntersection(t *testing.T) {
 		require_NoDuplicates(t, got)
 	})
 
+	t.Run("FWCExtended", func(t *testing.T) {
+		got := map[string]int{}
+		sl := NewSublistNoCache()
+		sl.Insert(newSub("stream.A.>"))
+		sl.Insert(newSub("stream.A"))
+		IntersectStree(st, sl, func(subj []byte, entry *struct{}) {
+			got[string(subj)]++
+		})
+		require_Len(t, len(got), 2)
+		require_NoDuplicates(t, got)
+	})
+
 	t.Run("FWCAll", func(t *testing.T) {
 		got := map[string]int{}
 		sl := NewSublistNoCache()
@@ -2078,7 +2093,7 @@ func TestSublistInterestBasedIntersection(t *testing.T) {
 		IntersectStree(st, sl, func(subj []byte, entry *struct{}) {
 			got[string(subj)]++
 		})
-		require_Len(t, len(got), 5)
+		require_Len(t, len(got), 7)
 		require_NoDuplicates(t, got)
 	})