Repository: nats-io/nats-server
Pull request #6831 (2025-04-24 15:03:38)

Cherry-picks for 2.10.28-RC.3
Includes the following:

* https://github.com/nats-io/nats-server/pull/6815
* https://github.com/nats-io/nats-server/pull/6825
* https://github.com/nats-io/nats-server/pull/6827

Signed-off-by: Neil Twigg <neil@nats.io>



Patch:
diff --git a/server/filestore.go b/server/filestore.go
index 57a98466..da5acb1a 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -4455,9 +4455,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)
@@ -4468,9 +4468,12 @@ func (fs *fileStore) removePerSubject(subj string) {
 		} else if info.total == 0 {
 			if _, ok = fs.psim.Delete(bsubj); ok {
 				fs.tsl -= len(subj)
+				return 0
 			}
 		}
+		return info.total
 	}
+	return 0
 }
 
 // Remove a message, optionally rewriting the mb file.
@@ -5504,6 +5507,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 {
@@ -5743,10 +5754,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
@@ -5792,18 +5806,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 {
@@ -7584,6 +7592,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
@@ -7632,6 +7645,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)
@@ -7655,8 +7670,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})
@@ -7686,6 +7701,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.
@@ -7700,6 +7720,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()
@@ -7716,6 +7740,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()
 		}
@@ -8364,20 +8389,20 @@ 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
 	}
 
 	if ss.Msgs == 1 {
 		mb.fss.Delete(bsubj)
-		return
+		return 0
 	}
 
 	ss.Msgs--
@@ -8387,18 +8412,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/raft.go b/server/raft.go
index 596adaf8..2b3aafb0 100644
--- a/server/raft.go
+++ b/server/raft.go
@@ -151,7 +151,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]struct{}            // Peers that were removed from the group
+	removed map[string]time.Time           // 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
 
@@ -246,6 +246,7 @@ const (
 	lostQuorumIntervalDefault      = hbIntervalDefault * 10 // 10 seconds
 	lostQuorumCheckIntervalDefault = hbIntervalDefault * 10 // 10 seconds
 	observerModeIntervalDefault    = 48 * time.Hour
+	peerRemoveTimeoutDefault       = 5 * time.Minute
 )
 
 var (
@@ -257,6 +258,7 @@ var (
 	lostQuorumInterval   = lostQuorumIntervalDefault
 	lostQuorumCheck      = lostQuorumCheckIntervalDefault
 	observerModeInterval = observerModeIntervalDefault
+	peerRemoveTimeout    = peerRemoveTimeoutDefault
 )
 
 type RaftConfig struct {
@@ -787,9 +789,9 @@ func (n *raft) ProposeAddPeer(peer string) error {
 func (n *raft) doRemovePeerAsLeader(peer string) {
 	n.Lock()
 	if n.removed == nil {
-		n.removed = map[string]struct{}{}
+		n.removed = map[string]time.Time{}
 	}
-	n.removed[peer] = struct{}{}
+	n.removed[peer] = time.Now()
 	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.
@@ -2853,9 +2855,9 @@ func (n *raft) applyCommit(index uint64) error {
 
 			// Make sure we have our removed map.
 			if n.removed == nil {
-				n.removed = make(map[string]struct{})
+				n.removed = make(map[string]time.Time)
 			}
-			n.removed[peer] = struct{}{}
+			n.removed[peer] = time.Now()
 
 			if _, ok := n.peers[peer]; ok {
 				delete(n.peers, peer)
@@ -2962,8 +2964,13 @@ 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 {
-		_, isRemoved = n.removed[peer]
+		rts, isRemoved = n.removed[peer]
+		// Removed peers can rejoin after timeout.
+		if isRemoved && time.Since(rts) >= peerRemoveTimeout {
+			isRemoved = false
+		}
 	}
 	if n.State() == Leader {
 		if lp, ok := n.peers[peer]; !ok || !lp.kp {
diff --git a/server/sublist.go b/server/sublist.go
index 9f79dfe1..6b931b12 100644
--- a/server/sublist.go
+++ b/server/sublist.go
@@ -1743,13 +1743,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 a36c1592..adc82bdf 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -9102,7 +9102,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_cluster_1_test.go b/server/jetstream_cluster_1_test.go
index c94ee711..d20fe422 100644
--- a/server/jetstream_cluster_1_test.go
+++ b/server/jetstream_cluster_1_test.go
@@ -3804,6 +3804,89 @@ 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 4652c19d..e008a8a1 100644
--- a/server/jetstream_consumer_test.go
+++ b/server/jetstream_consumer_test.go
@@ -17,6 +17,7 @@
 package server
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -30,6 +31,7 @@ import (
 	"time"
 
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
 	"github.com/nats-io/nuid"
 )
 
@@ -1760,3 +1762,58 @@ func TestJetStreamConsumerDeliveryCount(t *testing.T) {
 	require_Equal(t, o.deliveryCount(2), 1)
 
 }
+
+// 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_helpers_test.go b/server/jetstream_helpers_test.go
index 9f0400ad..9fe5a8b4 100644
--- a/server/jetstream_helpers_test.go
+++ b/server/jetstream_helpers_test.go
@@ -34,6 +34,7 @@ import (
 	"time"
 
 	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
 	"golang.org/x/time/rate"
 )
 
@@ -1229,6 +1230,19 @@ func jsClientConnectEx(t testing.TB, s *Server, jsOpts []nats.JSOpt, opts ...nat
 	return nc, js
 }
 
+func jsClientConnectNewAPI(t testing.TB, s *Server, opts ...nats.Option) (*nats.Conn, jetstream.JetStream) {
+	t.Helper()
+	nc, err := nats.Connect(s.ClientURL(), opts...)
+	if err != nil {
+		t.Fatalf("Failed to create client: %v", err)
+	}
+	js, err := jetstream.New(nc, jetstream.WithDefaultTimeout(10*time.Second))
+	if err != nil {
+		t.Fatalf("Unexpected error getting JetStream context: %v", err)
+	}
+	return nc, js
+}
+
 func jsClientConnectURL(t testing.TB, url string, opts ...nats.Option) (*nats.Conn, nats.JetStreamContext) {
 	t.Helper()
 
diff --git a/server/norace_test.go b/server/norace_test.go
index 9552044d..5b85d3c6 100644
--- a/server/norace_test.go
+++ b/server/norace_test.go
@@ -11560,3 +11560,46 @@ func TestNoRaceProducerStallLimits(t *testing.T) {
 	// just past 12ms, hence the Max setting below to avoid a flapper.
 	require_True(t, elapsed < stallTotalAllowed+stallClientMaxDuration)
 }
+
+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)
+	fs.StoreMsg("foo.B", nil, msg)
+	for i := 0; i < 500; i++ {
+		fs.StoreMsg("foo.C", nil, msg)
+	}
+	fs.StoreMsg("foo.D", nil, msg)
+
+	// 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 9b782960..e999659f 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)
 	})