Regression in FilterSubjects Handling with DeliverAll/ReplayInstant between v2.10.25 and v2.10.26

[hampusohlsson]

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.

[hampusohlsson]

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

Show code

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
		}
	}
}

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 440ea35

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!