Deadlock on PublishMsgAsync

[Zach-Johnson]

Observed behavior

I have code like this:

   futures := make([]jetstream.PubAckFuture, len(b))

   for i, msg := range b {
		jsMsg := nats.NewMsg(subj)

		msgBytes, err := msg.AsBytes()
		if err != nil {
			return err
		}

		jsMsg.Data = msgBytes

		future, err := js.PublishMsgAsync(jsMsg)
		if err != nil {
			return err
		}

		futures[i] = future
	}

	timeOut := time.After(5 * time.Second)

	select {
	case <-js.PublishAsyncComplete():
		for _, future := range futures {
			select {
			case <-future.Ok():
			case err := <-future.Err():
				return err
			case <-timeOut: // This will deadlock without this case here
				return errors.New("async publish complete, but futures did not return")
			}
		}
	case <-ctx.Done():
		return ctx.Err()
	case <-timeOut:
		return errors.New("did not receive async publish complete signal")
	}

It dead locks when reading the result of the futures if you remove the time out case.

This happens because there are ID collisions in the map that tracks acks. I updated the code that updates the map with a panic

func (js *jetStream) registerPAF(id string, paf *pubAckFuture) (int, int) {
	js.publisher.Lock()
	if js.publisher.acks == nil {
		js.publisher.acks = make(map[string]*pubAckFuture)
	}
	if _, ok := js.publisher.acks[id]; ok {
		panic(fmt.Sprintf("collision occurred on id: %s", id))
	}
	js.publisher.acks[id] = paf
	np := len(js.publisher.acks)
	maxpa := js.publisher.asyncPublisherOpts.maxpa
	js.publisher.Unlock()
	return np, maxpa
}

and I hit the panic occasionally.

I'm able to consistently reproduce this, although it takes a long time (up to an hour sometimes). I am using a batch size of 10k messages.

Expected behavior

No dead lock or some error condition to indicate a misconfiguration in the producer client.

Server and client version

server 2.10.25
go sdk v1.39.0

Host environment

No response

Steps to reproduce

No response