feat(queue): give every NATS enqueue a deduplication id (CLO-4772)

[levivannoort]

The utopia-php/monorepo finding of CLO-4772 — Epic 2 of the NATS queue migration readiness audit. The other six findings are appwrite/cloud.

Stacked on #173 (Epic 1) — merge that first; this branch is based on it.

---

The problem

Broker\Nats::enqueue published with no Nats-Msg-Id, and the work stream carried no duplicate window, so the header would have been ignored even if it had been sent.

The default request timeout is 5s, which makes the ambiguous publish real: the server can store a message and still leave the client waiting past its deadline for the ack. A caller retrying that had no way to say "this is the same message", so the copy was indistinguishable downstream from a genuine second one — on queues that bill people.

The change

Every publish carries its pid as Nats-Msg-Id, and the work stream is created with a duplicateWindow (default 120s, JetStream's own default). PubAck.duplicate is counted rather than discarded — it is the only signal that deduplication is doing anything, since a quiet success reads identically whether or not anything was collapsed.

The judgment call: a random pid per call cannot dedupe a caller that retries enqueue() itself, because that call mints a fresh one. Rather than guess an identity on the caller's behalf — two identical stats increments are both real work, and collapsing them would lose one — the broker takes an optional messageId closure:

new Nats($source, messageId: fn(array $payload): string => $payload['eventId']);

Callers that supply it get their own retries deduplicated; everyone else keeps today's semantics exactly. This is the seam Epic 2's cloud findings need — the Stripe and Resend idempotency keys both want a stable id derived from the message.

Publisher::enqueue is untouched: adding a parameter there would break every external implementer.

Security note: the returned id is rejected if empty or carrying CR, LF or NUL. It becomes a header value, Headers does not police what it is given, and a CRLF would end the header block early and inject the rest into the frame. Worth a second opinion on whether the validation belongs in Headers::set instead — I kept it at the point where the caller-supplied value enters, and left nats alone.

Clamping: the window is capped at the queue's jobTtl where one is set. JetStream refuses a stream whose duplicate window outlives its max age, and is right to — an id cannot be recognised as a duplicate of a message the stream has already discarded. Caught by testJobTtlExpiresUnackedMessages, which failed on the first attempt.

Verification

Check	Result
Unit	81 pass (79 baseline + 2)
E2E NatsBrokerTest	25 pass against a real JetStream server (1 pre-existing ext-swoole error)
Pint / PHPStan / Rector	clean
bin/monorepo validate	all packages valid

The dedupe test was checked against the unfixed code: removing msgId from the publish stores 2 messages instead of 1.

testWithoutAStableIdTwoIdenticalPayloadsRemainTwoMessages pins the boundary of the guarantee so it cannot drift silently.

[levivannoort]

Superseded by #173, which now carries this commit. Closing as part of collapsing the stack.

Why the stack could not go green: with the base set to fix/nats-connection-death-detection, origin/$base...HEAD only saw packages/queue, so changed was queue alone. queue was therefore not in dependents(changed) and ran registry-resolved — against released nats ^1.0 and pools ^2.0, neither of which has Connection::tick() or Pool::maintain(). phpstan failed on exactly those four calls. The workflow's own comment describes this hazard; it just is not detected when the sibling changed in an ancestor branch rather than in the PR's own range.

On #173, base main, changed is {nats, pools, queue}, so queue lands in the linked set and resolves its siblings from the checkout.

This commit merged cleanly — no resolution needed.

Branch feat/queue-nats-idempotent-enqueue is left in place; delete at your discretion.