Flush messages with `nack` on subscriber close

[brent-statsig]

Hey!

Problem

When closing the connection, there is no easy way to nack all messages remaining in the queue. This leads to our P99 delivery time of messages to inflate. It is fairly common for a pod to be pre-empted, which means there is a fairly short window to close down processing, and it is not realistic to wait until all processing is complete.

In the event of a connection closing, it is extremely difficult to ensure that all messages are nack without building my own version of a lease manager to track messages.

I noticed in LeaseManager.close, it wipes the local state. Can this just be updated to nack everything, and then wipe?

/**
   * Removes ALL messages from inventory.
   * @private
   */
  clear(): void {
    const wasFull = this.isFull();

    this._pending = [];
    this._messages.clear();
    this.bytes = 0;

    if (wasFull) {
      process.nextTick(() => this.emit('free'));
    }

    this._cancelExtension();
  }

[feywind]

@brent-statsig Yeah, this is something we're planning to address. Right now there isn't really a consistent way to deal with queued ack/modack/nack messages on close. There are a few other issues here asking for similar things, and I need to go back and link all of those at some point.