JetStream: the value for max pending async messages can be exceeded

[ygabuev]

Observed behavior

When using the simplified JetStream client in Go, it is possible to exceed the limit of max pending async messages that is specified by the WithPublishAsyncMaxPending option.

Expected behavior

The limit specified by WithPublishAsyncMaxPending should not be exceeded in practice.

Server and client version

NATS server: 2.10.12
NATS Go client: 1.34.1

Host environment

No response

Steps to reproduce

The following code

package main

import (
	"context"
	"fmt"
	"time"

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

const (
	Subject = "foo.bar.baz"
	Stream  = "FOO"
)

func main() {
	ctx := context.Background()
	nc, err := nats.Connect(nats.DefaultURL)
	if err != nil {
		panic(err)
	}

	js, err := jetstream.New(
		nc,
		jetstream.WithPublishAsyncMaxPending(1),
	)
	if err != nil {
		panic(err)
	}

	_, _ = js.CreateStream(ctx, jetstream.StreamConfig{
		Name:     Stream,
		Subjects: []string{Subject},
	})
	defer js.DeleteStream(ctx, Stream)

	go func() {
		for {
			fmt.Printf("currently pending: %d\n", js.PublishAsyncPending())
			time.Sleep(50 * time.Millisecond)
		}
	}()

	for i := 0; i < 5; i++ {
		go func(i int) {
			_, err := js.PublishAsync(Subject, []byte{})
			if err != nil {
				fmt.Printf("%d  %v\n", i, err)
			}
		}(i)
	}

	time.Sleep(200 * time.Millisecond)
}

generates the following output for me:

currently pending: 0
currently pending: 4
currently pending: 4
currently pending: 4
2  nats: stalled with too many outstanding async published messages
3  nats: stalled with too many outstanding async published messages
0  nats: stalled with too many outstanding async published messages
1  nats: stalled with too many outstanding async published messages
currently pending: 0