Repository: nats-io/nats-server

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.

Discussion:
@hampusohlsson: Created a simple test that replicates the issue, if you want to try it.

<details>

<summary>Show code</summary>

```go
package main

import (
	"context"
	"fmt"
	"log"
	"os"
	"time"

	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

func main() {
	ctx := context.Background()
	// Create temporary directory for storing data
	storeDir, err := os.MkdirTemp(os.TempDir(), "js-*")
	if err != nil {
		log.Fatal(err)
	}
	// Create server with JetStream support
	s, err := server.NewServer(&server.Options{
		NoLog:     false,
		NoSigs:    true,
		JetStream: true,
		StoreDir:  storeDir,
	})
	if err != nil {
		log.Fatal(err)
	}
	// Start server
	go s.Start()
	if !s.ReadyForConnections(1 * time.Second) {
		log.Fatal(err)
	}

	log.Println("Server started")

	// Connection
	nc, err := nats.Connect(nats.DefaultURL)
	if err != nil {
		log.Fatalf("Error connecting to NATS at %s: %v", nats.DefaultURL, err)
	}
	defer nc.Close()
	log.Printf("Connected to NATS server at %s", nc.ConnectedUrl())
	// Jetstream
	js, err := jetstream.New(nc)
	if err != nil {
		log.Fatalf("Error creating JetStream context: %v", err)
	}

	// Create stream
	streamName := "mystream"
	if _, err := js.CreateOrUpdateStream(ctx, jetstream.StreamConfig{
		Name:      streamName,
		Subjects:  []string{"stream.>"},
		Retention: jetstream.LimitsPolicy,
		MaxAge:    1 * time.Hour,
		Storage:   jetstream.FileStorage,
		Replicas:  1,
	}); err != nil {
		log.Fatalf("Error creating/updating stream '%s': %v", streamName, err)
	}

	// Publish retained messages
	publishSubject := "stream.A"
	publishMessageCount := 10
	// Create a channel to record messages received
	messages := make(chan jetstream.Msg, publishMessageCount)

	for i := 1; i <= publishMessageCount; i++ {
		data := []byte(fmt.Sprintf("message #%d", i))
		_, err := js.Publish(ctx, publishSubject, data)
		if err != nil {
			log.Fatalf("Error publishing message %d: %v", i, err)
		}
	}
	log.Printf("Publishing %d messages to subject '%s'", publishMessageCount, publishSubject)

	// Create consumer
	consumer, err := js.CreateOrUpdateConsumer(ctx, streamName, jetstream.ConsumerConfig{
		DeliverPolicy: jetstream.DeliverAllPolicy,
		ReplayPolicy:  jetstream.ReplayInstantPolicy,
		AckPolicy:     jetstream.AckExplicitPolicy,
		AckWait:       30 * time.Second,
		FilterSubjects: []string{
			"stream.A",
			"stream.A.>",
		},
	})
	if err != nil {
		log.Fatal(err)
	}

	log.Println("Consuming messages...")
	if _, err := consumer.Consume(func(msg jetstream.Msg) {
		messages <- msg
		msg.Ack()
	}); err != nil {
		log.Fatal(err)
	}

	timer := time.NewTimer(5 * time.Second)
	var count int = 0
	for {
		if count == publishMessageCount {
			log.Println("All messages received.")
			break
		}
		select {
		case msg := <-messages:
			log.Printf("Received message: %s\n", string(msg.Data()))
			count++
		case <-timer.C:
			log.Printf("Timeout reached, %d messages received. Exiting.", count)
			break
		}
	}
}
```
</details>

## Output

#### `v2.10.26`
```
2025/04/24 08:09:10 Server started
2025/04/24 08:09:10 Connected to NATS server at nats://127.0.0.1:4222
2025/04/24 08:09:10 Publishing 10 messages to subject 'stream.A'
2025/04/24 08:09:10 Consuming messages...
2025/04/24 08:09:15 Timeout reached, 0 messages received. Exiting.
```

#### `v2.10.25`
```
2025/04/24 08:09:55 Server started
2025/04/24 08:09:55 Connected to NATS server at nats://127.0.0.1:4222
2025/04/24 08:09:55 Publishing 10 messages to subject 'stream.A'
2025/04/24 08:09:55 Consuming messages...
2025/04/24 08:09:55 Received message: message #1
2025/04/24 08:09:55 Received message: message #2
2025/04/24 08:09:55 Received message: message #3
2025/04/24 08:09:55 Received message: message #4
2025/04/24 08:09:55 Received message: message #5
2025/04/24 08:09:55 Received message: message #6
2025/04/24 08:09:55 Received message: message #7
2025/04/24 08:09:55 Received message: message #8
2025/04/24 08:09:55 Received message: message #9
2025/04/24 08:09:55 Received message: message #10
2025/04/24 08:09:55 All messages received.
```@hampusohlsson: A member in our team ran `git bisect` to find the first bad commit that introduced the regression.

It's pointing to https://github.com/nats-io/nats-server/commit/440ea35fc86b50c818d966cf5400db18e76f0f4e

> **Optimise firstMatchingMulti using subject tree intersection**
>
> This avoids a potentially expensive linear walk if it is obvious that the
block FSS contains less subjects than sequences that we would otherwise
have to scan.

Thought it could be helpful for root causing.@MauriceVanVeen: Thank you for the report and test! Also managed to bisect my way to that commit, we're close to a fix :) @neilalexander: Will be fixed in 2.10.28 and 2.11.2, thanks once again for the information!@hampusohlsson: > Will be fixed in 2.10.28 and 2.11.2, thanks once again for the information!

That's great, thanks for the swift response!

Solution patch:
diff --git a/server/filestore.go b/server/filestore.go
index 78a2c934..37874030 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -4602,9 +4602,9 @@ func (fs *fileStore) EraseMsg(seq uint64) (bool, error) {
 
 // Convenience function to remove per subject tracking at the filestore level.
 // Lock should be held.
-func (fs *fileStore) removePerSubject(subj string) {
+func (fs *fileStore) removePerSubject(subj string) uint64 {
 	if len(subj) == 0 || fs.psim == nil {
-		return
+		return 0
 	}
 	// We do not update sense of fblk here but will do so when we resolve during lookup.
 	bsubj := stringToBytes(subj)
@@ -4615,10 +4615,12 @@ func (fs *fileStore) removePerSubject(subj string) {
 		} else if info.total == 0 {
 			if _, ok = fs.psim.Delete(bsubj); ok {
 				fs.tsl -= len(subj)
-				return
+				return 0
 			}
 		}
+		return info.total
 	}
+	return 0
 }
 
 // Remove a message, optionally rewriting the mb file.
@@ -5753,7 +5755,7 @@ func (fs *fileStore) handleRemovalOrSdm(seq uint64, subj string, sdm bool, sdmTT
 		var _hdr [128]byte
 		hdr := fmt.Appendf(
 			_hdr[:0],
-			"NATS/1.0\r\n%s: %s\r\n%s: %s\r\n%s: %s\r\n\r\n\r\n",
+			"NATS/1.0\r\n%s: %s\r\n%s: %s\r\n%s: %s\r\n\r\n",
 			JSMarkerReason, JSMarkerReasonMaxAge,
 			JSMessageTTL, time.Duration(sdmTTL)*time.Second,
 			JSMsgRollup, JSMsgRollupSubject,
@@ -5836,6 +5838,14 @@ func (mb *msgBlock) writeTombstone(seq uint64, ts int64) error {
 	return mb.writeMsgRecord(emptyRecordLen, seq|tbit, _EMPTY_, nil, nil, ts, true)
 }
 
+// Helper function to place a delete tombstone without flush.
+// Lock should not be held.
+func (mb *msgBlock) writeTombstoneNoFlush(seq uint64, ts int64) error {
+	mb.mu.Lock()
+	defer mb.mu.Unlock()
+	return mb.writeMsgRecordLocked(emptyRecordLen, seq|tbit, _EMPTY_, nil, nil, ts, false, false)
+}
+
 // Will write the message record to the underlying message block.
 // filestore lock will be held.
 func (mb *msgBlock) writeMsgRecord(rl, seq uint64, subj string, mhdr, msg []byte, ts int64, flush bool) error {
@@ -6075,10 +6085,13 @@ func (fs *fileStore) checkLastBlock(rl uint64) (lmb *msgBlock, err error) {
 	lmb = fs.lmb
 	rbytes := lmb.blkSize()
 	if lmb == nil || (rbytes > 0 && rbytes+rl > fs.fcfg.BlockSize) {
-		if lmb != nil && fs.fcfg.Compression != NoCompression {
-			// We've now reached the end of this message block, if we want
-			// to compress blocks then now's the time to do it.
-			go lmb.recompressOnDiskIfNeeded()
+		if lmb != nil {
+			lmb.flushPendingMsgs()
+			if fs.fcfg.Compression != NoCompression {
+				// We've now reached the end of this message block, if we want
+				// to compress blocks then now's the time to do it.
+				go lmb.recompressOnDiskIfNeeded()
+			}
 		}
 		if lmb, err = fs.newMsgBlockForWrite(); err != nil {
 			return nil, err
@@ -6124,18 +6137,12 @@ func (fs *fileStore) writeTombstone(seq uint64, ts int64) error {
 // This version does not flush contents.
 // Lock should be held.
 func (fs *fileStore) writeTombstoneNoFlush(seq uint64, ts int64) error {
-	// Grab our current last message block.
-	olmb := fs.lmb
 	lmb, err := fs.checkLastBlock(emptyRecordLen)
 	if err != nil {
 		return err
 	}
-	// If we swapped out our lmb, flush any pending.
-	if olmb != lmb {
-		olmb.flushPendingMsgs()
-	}
 	// Write tombstone without flush or kick.
-	return lmb.writeTombstone(seq, ts)
+	return lmb.writeTombstoneNoFlush(seq, ts)
 }
 
 func (mb *msgBlock) recompressOnDiskIfNeeded() error {
@@ -7942,6 +7949,11 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 		}
 	}
 
+	// Make sure to not leave subject if empty and we reach this spot.
+	if subject == _EMPTY_ {
+		subject = fwcs
+	}
+
 	eq, wc := compareFn(subject), subjectHasWildcard(subject)
 	var firstSeqNeedsUpdate bool
 	var bytes uint64
@@ -7990,6 +8002,8 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 			shouldExpire = true
 		}
 
+		var nrg uint64 // Number of remaining messages globally after removal from psim.
+
 		for seq, te := f, len(tombs); seq <= l; seq++ {
 			if sm, _ := mb.cacheLookupNoCopy(seq, &smv); sm != nil && eq(sm.subj, subject) {
 				rl := fileStoreMsgSize(sm.subj, sm.hdr, sm.msg)
@@ -8013,8 +8027,8 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 					bytes += rl
 				}
 				// PSIM and FSS updates.
-				mb.removeSeqPerSubject(sm.subj, seq)
-				fs.removePerSubject(sm.subj)
+				nr := mb.removeSeqPerSubject(sm.subj, seq)
+				nrg = fs.removePerSubject(sm.subj)
 
 				// Track tombstones we need to write.
 				tombs = append(tombs, msgId{sm.seq, sm.ts})
@@ -8044,6 +8058,11 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 				if mb.isEmpty() || (maxp > 0 && purged >= maxp) {
 					break
 				}
+				// Also break if we know we have no more messages matching here.
+				// This is only applicable for non-wildcarded filters.
+				if !wc && nr == 0 {
+					break
+				}
 			}
 		}
 		// Expire if we were responsible for loading and we do not seem to be doing successive purgeEx calls.
@@ -8058,6 +8077,10 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 		if maxp > 0 && purged >= maxp {
 			break
 		}
+		// Also check if not wildcarded and we have no remaining matches.
+		if !wc && nrg == 0 {
+			break
+		}
 	}
 	if firstSeqNeedsUpdate {
 		fs.selectNextFirst()
@@ -8074,6 +8097,7 @@ func (fs *fileStore) PurgeEx(subject string, sequence, keep uint64) (purged uint
 				return purged, err
 			}
 		}
+		// Flush any pending. If we change blocks the checkLastBlock() will flush any pending for us.
 		if lmb := fs.lmb; lmb != nil {
 			lmb.flushPendingMsgs()
 		}
@@ -8727,21 +8751,21 @@ func (mb *msgBlock) dirtyCloseWithRemove(remove bool) error {
 
 // Remove a seq from the fss and select new first.
 // Lock should be held.
-func (mb *msgBlock) removeSeqPerSubject(subj string, seq uint64) {
+func (mb *msgBlock) removeSeqPerSubject(subj string, seq uint64) uint64 {
 	mb.ensurePerSubjectInfoLoaded()
 	if mb.fss == nil {
-		return
+		return 0
 	}
 	bsubj := stringToBytes(subj)
 	ss, ok := mb.fss.Find(bsubj)
 	if !ok || ss == nil {
-		return
+		return 0
 	}
 
 	mb.fs.sdm.removeSeqAndSubject(seq, subj)
 	if ss.Msgs == 1 {
 		mb.fss.Delete(bsubj)
-		return
+		return 0
 	}
 
 	ss.Msgs--
@@ -8751,18 +8775,20 @@ func (mb *msgBlock) removeSeqPerSubject(subj string, seq uint64) {
 		if !ss.lastNeedsUpdate && seq != ss.Last {
 			ss.First = ss.Last
 			ss.firstNeedsUpdate = false
-			return
+			return 1
 		}
 		if !ss.firstNeedsUpdate && seq != ss.First {
 			ss.Last = ss.First
 			ss.lastNeedsUpdate = false
-			return
+			return 1
 		}
 	}
 
 	// We can lazily calculate the first/last sequence when needed.
 	ss.firstNeedsUpdate = seq == ss.First || ss.firstNeedsUpdate
 	ss.lastNeedsUpdate = seq == ss.Last || ss.lastNeedsUpdate
+
+	return ss.Msgs
 }
 
 // Will recalculate the first and/or last sequence for this subject in this block.
diff --git a/server/memstore.go b/server/memstore.go
index 70848598..411ff714 100644
--- a/server/memstore.go
+++ b/server/memstore.go
@@ -1193,7 +1193,7 @@ func (ms *memStore) handleRemovalOrSdm(seq uint64, subj string, sdm bool, sdmTTL
 		var _hdr [128]byte
 		hdr := fmt.Appendf(
 			_hdr[:0],
-			"NATS/1.0\r\n%s: %s\r\n%s: %s\r\n%s: %s\r\n\r\n\r\n",
+			"NATS/1.0\r\n%s: %s\r\n%s: %s\r\n%s: %s\r\n\r\n",
 			JSMarkerReason, JSMarkerReasonMaxAge,
 			JSMessageTTL, time.Duration(sdmTTL)*time.Second,
 			JSMsgRollup, JSMsgRollupSubject,
diff --git a/server/sdm.go b/server/sdm.go
index 4de53582..74314795 100644
--- a/server/sdm.go
+++ b/server/sdm.go
@@ -1,3 +1,16 @@
+// Copyright 2025 The NATS Authors
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package server
 
 import "time"
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/filestore_test.go b/server/filestore_test.go
index a08654c3..74665610 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -9529,7 +9529,7 @@ func TestFileStoreRemoveMsgBlockLast(t *testing.T) {
 
 func TestFileStoreAllLastSeqs(t *testing.T) {
 	fs, err := newFileStore(
-		FileStoreConfig{StoreDir: t.TempDir()}, // Make block size small to test multiblock selections with maxSeq
+		FileStoreConfig{StoreDir: t.TempDir()},
 		StreamConfig{Name: "zzz", Subjects: []string{"*.*"}, MaxMsgsPer: 50, Storage: FileStorage})
 	require_NoError(t, err)
 	defer fs.Stop()
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/jetstream_test.go b/server/jetstream_test.go
index f527661b..cd06cc99 100644
--- a/server/jetstream_test.go
+++ b/server/jetstream_test.go
@@ -19787,3 +19787,40 @@ func TestJetStreamRejectLargePublishes(t *testing.T) {
 	require_Error(t, err)
 	require_Contains(t, err.Error(), ErrMsgTooLarge.Error())
 }
+
+func TestJetStreamDirectGetSubjectDeleteMarker(t *testing.T) {
+	for _, storageType := range []nats.StorageType{nats.FileStorage, nats.MemoryStorage} {
+		t.Run(storageType.String(), func(t *testing.T) {
+			s := RunBasicJetStreamServer(t)
+			defer s.Shutdown()
+
+			nc, js := jsClientConnect(t, s)
+			defer nc.Close()
+
+			_, err := js.AddStream(&nats.StreamConfig{
+				Name:                   "TEST",
+				Subjects:               []string{"test"},
+				Storage:                storageType,
+				SubjectDeleteMarkerTTL: time.Second,
+				AllowMsgTTL:            true,
+				AllowDirect:            true,
+			})
+			require_NoError(t, err)
+
+			m := nats.NewMsg("test")
+			m.Header.Set(JSMessageTTL, "1s")
+			_, err = js.PublishMsg(m)
+			require_NoError(t, err)
+
+			first, err := js.GetLastMsg("TEST", "test", nats.DirectGet())
+			require_NoError(t, err)
+			require_Equal(t, first.Header.Get(JSSequence), "1")
+
+			time.Sleep(1500 * time.Millisecond)
+
+			second, err := js.GetLastMsg("TEST", "test", nats.DirectGet())
+			require_NoError(t, err)
+			require_Equal(t, second.Header.Get(JSSequence), "2")
+		})
+	}
+}
diff --git a/server/norace_2_test.go b/server/norace_2_test.go
index 12b0c3ba..ccafb450 100644
--- a/server/norace_2_test.go
+++ b/server/norace_2_test.go
@@ -3352,3 +3352,46 @@ func TestNoRaceJetStreamClusterConsumerDeleteInterestPolicyUniqueFiltersPerf(t *
 	}
 	expectedStreamMsgs(0)
 }
+
+func TestNoRaceFileStorePurgeExAsyncTombstones(t *testing.T) {
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: t.TempDir()},
+		StreamConfig{Name: "zzz", Subjects: []string{"*.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	msg := []byte("zzz")
+
+	fs.StoreMsg("foo.A", nil, msg, 0)
+	fs.StoreMsg("foo.B", nil, msg, 0)
+	for i := 0; i < 500; i++ {
+		fs.StoreMsg("foo.C", nil, msg, 0)
+	}
+	fs.StoreMsg("foo.D", nil, msg, 0)
+
+	// Load all blocks to avoid that being a factor in timing.
+	fs.mu.RLock()
+	for _, mb := range fs.blks {
+		mb.loadMsgs()
+	}
+	fs.mu.RUnlock()
+
+	// Now purge 1 that is not the first message and take note of time.
+	// Since we are loaded this should mostly be the time to write / flush tombstones.
+	start := time.Now()
+	n, err := fs.PurgeEx("foo.B", 0, 0)
+	elapsed := time.Since(start)
+	require_NoError(t, err)
+	require_Equal(t, n, 1)
+
+	start = time.Now()
+	n, err = fs.PurgeEx("foo.C", 0, 0)
+	elapsed2 := time.Since(start)
+	require_NoError(t, err)
+	require_Equal(t, n, 500)
+
+	// If we are flushing for each tombstone the second elapsed time will be a larger multiple of the single message purge.
+	// In testing this is like >200x
+	// With async and flush for all tombstones will be ~30x
+	require_True(t, elapsed*50 > elapsed2)
+}
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)
 	})