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

Issue:
Consumer dropping messages in 2.10.17 / 2.10.18
### Observed behavior

I have an example where there are two streams (full example below in **Steps to reproduce**):

* A **workqueue stream** (TEST_WORKQUEUE), populated with numbers from 0 to 129999.
* A stream of **active work items** (TEST_ACTIVE_WORK_ITEMS), where each item has an ID (number from 0 to 129999) and an expiry time.

Note:
* Both streams have MaxMessagesPerSubject set to 1, with 1 subject per item ID, to avoid item duplication.

These streams are used as follows:

* TEST_WORKQUEUE is initially populated with 130000 numbers.
* Then, 100000 items are fetched from TEST_WORKQUEUE and published to TEST_ACTIVE_WORK_ITEMS. Each item in the active work items stream has an ID (the number gotten from the workqueue) and an expiry, which is set to + 30 seconds.
* When publishing items to TEST_ACTIVE_WORK_ITEMS, we do it twice in a row; the second publish is meant to replace the previous one, bumping up that item ID's expiry time. Note that TEST_ACTIVE_WORK_ITEMS has a discard policy of DiscardOld to allow this behavior.
* Once in the active work items stream, a consumer handles item expiration. NakWithDelay is used to schedule expiration. When an item expires, it's published back into TEST_WORKQUEUE and acked by the consumer.

In the end, we enter a loop that prints some statistics for monitoring the stream and consumer state.

The expected behavior is that, after the consumer's ack pending count goes down to 0, the size of TEST_WORKQUEUE should go back to 130000. So, we should see the following in the logs when the consumer no longer has any pending ack:

```
TEST_ACTIVE_WORK_ITEMS consumer ack pending 0
TEST_WORKQUEUE stream size 130000
```

In NATS 2.10.12, this seems to work without any issues. However, in NATS 2.10.17 and 2.10.18, the consumer seems to drop items, and the final number of items going back to TEST_WORKQUEUE is almost always less than 130000. For example:

```
TEST_ACTIVE_WORK_ITEMS consumer ack pending 0     
TEST_WORKQUEUE stream size 97356
```

Also tested with latest commit in main; same issue.

### Expected behavior

The expected behavior is that, after the consumer's ack pending count goes down to 0, the size of TEST_WORKQUEUE should go back to 130000. So, we should see the following in the logs when the consumer no longer has any pending ack:

```
TEST_ACTIVE_WORK_ITEMS consumer ack pending 0
TEST_WORKQUEUE stream size 130000
```

### Server and client version

Server: NATS 2.10.17 / 2.10.18. Also tested with latest commit in main; same issue.
Client: NATS go client 1.36.0.

### Host environment

Ubuntu 20.04.6 LTS.

### Steps to reproduce

The script that reproduces this behavior is below. It expects NATS to be at `127.0.0.1:4222`, so a local NATS just needs to be spun up with `nats-server -js`. The script will then delete/recreate the test streams (TEST_WORKQUEUE and TEST_ACTIVE_WORK_ITEMS), and the issue can be verified from the logs as described above.

```go
package main

import (
	"encoding/json"
	"errors"
	"fmt"
	"strconv"
	"sync"
	"time"

	"github.com/nats-io/nats.go"
)

type ActiveWorkItem struct {
	ID     string
	Expiry time.Time
}

// Delete and recreate TEST_WORKQUEUE and TEST_ACTIVE_WORK_ITEMS.
func resetStreams(js nats.JetStreamContext) {
	_ = js.DeleteStream("TEST_WORKQUEUE")
	_ = js.DeleteStream("TEST_ACTIVE_WORK_ITEMS")
	_, err := js.AddStream(&nats.StreamConfig{
		Name:              "TEST_ACTIVE_WORK_ITEMS",
		Discard:           nats.DiscardOld,
		MaxMsgsPerSubject: 1,
		Subjects:          []string{"TEST_ACTIVE_WORK_ITEMS.>"},
	})
	if err != nil {
		panic(err)
	}
	_, err = js.AddStream(&nats.StreamConfig{
		Name:                 "TEST_WORKQUEUE",
		Discard:              nats.DiscardNew,
		MaxMsgsPerSubject:    1,
		DiscardNewPerSubject: true,
		Retention:            nats.WorkQueuePolicy,
		Subjects:             []string{"TEST_WORKQUEUE.>"},
	})
	if err != nil {
		panic(err)
	}
}

// Populate TEST_WORKQUEUE stream with 130000 items.
// Each item is a sequential number from 0 to 129999.
func populateWorkqueue(js nats.JetStreamContext) {
	NUM_MSGS := 130000
	var wg sync.WaitGroup
	for i := 0; i < NUM_MSGS; i++ {
		wg.Add(1)
		i := i
		go func() {
			defer wg.Done()
			itemID := fmt.Sprintf("%d", i)
			js.Publish(fmt.Sprintf("TEST_WORKQUEUE.%s", itemID), []byte(itemID))
		}()
	}
	wg.Wait()

	streamInfo, err := js.StreamInfo("TEST_WORKQUEUE")
	if err != nil {
		panic(err)
	}
	fmt.Println("Finished populating TEST_WORKQUEUE", streamInfo.State.Msgs)
}

// Launch a consumer for TEST_ACTIVE_WORK_ITEMS stream.
//
// This consumer works as follows:
// * If the active work item hasn't expired, it's rescheduled using NakWithDelay.
// * Otherwise, the consumer publishes it back to the TEST_WORKQUEUE stream and ACKs it in TEST_ACTIVE_WORK_ITEMS.
func startActiveWorkItemsConsumer(js nats.JetStreamContext) {
	if _, err := js.AddConsumer("TEST_ACTIVE_WORK_ITEMS", &nats.ConsumerConfig{
		Durable:       "testactiveworkitemsconsumer",
		AckPolicy:     nats.AckExplicitPolicy,
		MaxAckPending: -1,
		MaxWaiting:    20000,
		AckWait:       30 * time.Second,
	}); err != nil {
		panic(err)
	}
	sub, err := js.PullSubscribe("TEST_ACTIVE_WORK_ITEMS.>", "testactiveworkitemsconsumer", nats.BindStream("TEST_ACTIVE_WORK_ITEMS"))
	if err != nil {
		panic(err)
	}
	go func() {
		for {
			// Get a batch of at most 200 messages.
			msgs, err := sub.Fetch(200)
			// If there is any error, go back and retry the fetch.
			if err != nil {
				if !errors.Is(err, nats.ErrTimeout) {
					fmt.Println("active work items consumer: unexpected error fetching msgs", err)
				}
				continue
			}
			// Process the batch.
			var wg sync.WaitGroup
			for _, msg := range msgs {
				msg := msg
				wg.Add(1)
				// Process an individual message
				go func() {
					defer wg.Done()
					var workItem ActiveWorkItem
					err = json.Unmarshal(msg.Data, &workItem)
					if err != nil {
						panic(err)
					}
					now := time.Now()
					// If the work item has not expired, nak it with the respective delay.
					if workItem.Expiry.After(now) {
						_ = msg.NakWithDelay(workItem.Expiry.Sub(now))
						// Even if we get an error, just return; in either case, item should be eventually retried.
						return
					}
					// Work item has expired, so publish this item's ID back to TEST_WORKQUEUE.
					_, err = js.Publish(fmt.Sprintf("TEST_WORKQUEUE.%s", workItem.ID), []byte(workItem.ID))
					switch {
					case err == nil || errors.Is(err, nats.ErrMaxMessages):
						// Either the item has been published to TEST_WORKQUEUE or it already existed there,
						// so we just need to ack this item.
						if err = msg.AckSync(); err != nil {
							return
						}
					// If we get any error other than ErrMaxMessages, then just return and let the message be redelivered.
					default:
						return
					}
				}()
			}
			// Wait for batch to finish.
			wg.Wait()
		}
	}()
}

// Fetch items from the workqueue and publish them to the active work items stream, removing them from the workqueue.
func processWorkqueueItems(js nats.JetStreamContext) {
	if _, err := js.AddConsumer("TEST_WORKQUEUE", &nats.ConsumerConfig{
		Durable:       "testworkqueueconsumer",
		AckPolicy:     nats.AckExplicitPolicy,
		MaxWaiting:    20000,
		MaxAckPending: -1,
	}); err != nil {
		panic(err)
	}

	workqueueSub, err := js.PullSubscribe("TEST_WORKQUEUE.>", "testworkqueueconsumer", nats.BindStream("TEST_WORKQUEUE"))
	if err != nil {
		panic(err)
	}

	NUM_MSGS := 100000
	var wg sync.WaitGroup
	for i := 0; i < NUM_MSGS; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			msgs, err := workqueueSub.Fetch(1)
			if err != nil {
				fmt.Println("error fetching from workqueue", err)
				return
			}
			itemID, err := strconv.Atoi(string(msgs[0].Data))
			if err != nil {
				panic(err)
			}
			// Publish item to TEST_ACTIVE_WORK_ITEMS stream with an expiry time.
			workItem := ActiveWorkItem{ID: string(msgs[0].Data), Expiry: time.Now().Add(30 * time.Second)}
			data, err := json.Marshal(workItem)
			if err != nil {
				panic(err)
			}
			_, err = js.Publish(fmt.Sprintf("TEST_ACTIVE_WORK_ITEMS.%d", itemID), data)
			if err != nil {
				fmt.Println("error publishing to TEST_ACTIVE_WORK_ITEMS", workItem, err)
				return
			}
			// Update expiry time and republish item to TEST_ACTIVE_WORK_ITEMS stream.
			workItem.Expiry = time.Now().Add(30 * time.Second)
			data, err = json.Marshal(workItem)
			if err != nil {
				panic(err)
			}
			_, err = js.Publish(fmt.Sprintf("TEST_ACTIVE_WORK_ITEMS.%d", itemID), data)
			if err != nil {
				fmt.Println("error publishing to TEST_ACTIVE_WORK_ITEMS", workItem, err)
				return
			}
			// Ack item to remove it from TEST_WORKQUEUE.
			err = msgs[0].AckSync()
			if err != nil {
				fmt.Println("error acking msg from workqueue", err)
				return
			}
		}()
		// Pause every 20000 items.
		if (i > 0) && (i%20000 == 0) {
			wg.Wait()
		}
	}
	wg.Wait()
}

func main() {
	var nc *nats.Conn
	var js nats.JetStreamContext
	nc, err := nats.Connect("127.0.0.1")
	if err != nil {
		panic(err)
	}
	defer nc.Close()
	if js, err = nc.JetStream(); err != nil {
		panic(err)
	}

	resetStreams(js)

	populateWorkqueue(js)

	startActiveWorkItemsConsumer(js)

	processWorkqueueItems(js)

	// Print info on a loop for monitoring stream and consumer state.
	for {
		time.Sleep(5 * time.Second)
		consumerInfo, err := js.ConsumerInfo("TEST_ACTIVE_WORK_ITEMS", "testactiveworkitemsconsumer")
		if err != nil {
			panic(err)
		}

		fmt.Println("TEST_ACTIVE_WORK_ITEMS consumer ack pending", consumerInfo.NumAckPending)

		streamInfo, err := js.StreamInfo("TEST_WORKQUEUE")
		if err != nil {
			panic(err)
		}

		fmt.Println("TEST_WORKQUEUE stream size", streamInfo.State.Msgs)
	}
}
```

Pull request:
Fix 'checkSkipFirstBlock' with '_EMPTY_' filter
As mentioned in https://github.com/nats-io/nats-server/issues/5705#issuecomment-2254268218, an issue was introduced starting from 2.10.17.

A call to `fs.numFilteredPending(filter, &ss)` was replaced with `nbi := fs.checkSkipFirstBlock(filter, wc)`. But the latter wouldn't handle the case where `filter == _EMPTY_` from a call to `LoadNextMsg`, whereas the former would.

Resolves #5705

Signed-off-by: Maurice van Veen <github@mauricevanveen.com>

Patch:
diff --git a/server/filestore.go b/server/filestore.go
index b034ac77..6c2a902f 100644
--- a/server/filestore.go
+++ b/server/filestore.go
@@ -2610,6 +2610,11 @@ func (fs *fileStore) FilteredState(sseq uint64, subj string) SimpleState {
 // This is used to see if we can selectively jump start blocks based on filter subject and a floor block index.
 // Will return -1 if no matches at all.
 func (fs *fileStore) checkSkipFirstBlock(filter string, wc bool) (int, int) {
+	if filter == _EMPTY_ {
+		filter = fwcs
+		wc = true
+	}
+
 	start, stop := uint32(math.MaxUint32), uint32(0)
 	if wc {
 		fs.psim.Match(stringToBytes(filter), func(_ []byte, psi *psi) {


Test patch:
diff --git a/server/filestore_test.go b/server/filestore_test.go
index 371b7256..0542781e 100644
--- a/server/filestore_test.go
+++ b/server/filestore_test.go
@@ -7349,6 +7349,28 @@ func TestFileStoreCheckSkipFirstBlockBug(t *testing.T) {
 	require_NoError(t, err)
 }
 
+// https://github.com/nats-io/nats-server/issues/5705
+func TestFileStoreCheckSkipFirstBlockEmptyFilter(t *testing.T) {
+	sd := t.TempDir()
+	fs, err := newFileStore(
+		FileStoreConfig{StoreDir: sd, BlockSize: 128},
+		StreamConfig{Name: "zzz", Subjects: []string{"foo.*.*"}, Storage: FileStorage})
+	require_NoError(t, err)
+	defer fs.Stop()
+
+	msg := []byte("hello")
+	// Create 4 blocks, each block holds 2 msgs
+	for i := 0; i < 4; i++ {
+		fs.StoreMsg("foo.22.bar", nil, msg)
+		fs.StoreMsg("foo.22.baz", nil, msg)
+	}
+	require_Equal(t, fs.numMsgBlocks(), 4)
+
+	nbi, lbi := fs.checkSkipFirstBlock(_EMPTY_, false)
+	require_Equal(t, nbi, 0)
+	require_Equal(t, lbi, 3)
+}
+
 ///////////////////////////////////////////////////////////////////////////
 // Benchmarks
 ///////////////////////////////////////////////////////////////////////////