fix(queue): stop a failed ack pinning its message in flight

[levivannoort]

Closes #196.

Broker\Nats::commit() acked and then dropped the message from $inFlight:

$jsMessage->ackSync();
unset($this->inFlight[$pid]);

ackSync() is a request-reply, so a transient failure raises rather than returning a verdict — and that is precisely when the unset was skipped.

Nothing else clears the entry. reject() is the only other place that unsets, and Adapter::runPhases() deliberately does not reject after a commit failure — the work is done, so a NAK there is a duplicate guaranteed rather than a duplicate risked. The entry is left with no owner, and a long-lived worker accumulates one pinned JetStreamMessage per failed ack.

The fix

The map records that this instance owes an ack for the message, not that one succeeded, so the unset belongs in a finally. The throw still propagates, because runPhases() reporting it is the behaviour that decided not to reject.

Verification

NatsBrokerTest::testAnAckThatFailsDoesNotPinTheMessageInFlight closes the connection under the ack — the cheapest stand-in for the transient failure this guards — and asserts the map is clear afterwards. Confirmed red before the fix and green after:

in flight after receive: yes
commit threw: Utopia\NATS\Exception\ConnectionException
still in flight after the failed ack: YES -> leak     # before
still in flight after the failed ack: no  -> cleared  # after

The new test is in the e2e tier, which needs a JetStream server and cannot run on a host without ext-swoole (tests/e2e.sh boots Swoole workers), so it rests on CI — the red/green above was taken against a real server via the same steps the test performs. bin/monorepo check queue (pint, phpstan, rector) passes, and the unit tier is unchanged at 100 tests.

Found by a review sweep over main while resolving #192; not introduced by it. Two other findings from that sweep are filed as #197 and #198.