Fix premature KV watcher initial completion

[d3vv3]

Summary

• derive a fixed minimum initial-delivery boundary from the consumer-info snapshot
• count all callback deliveries, including ignored delete and purge markers
• emit the initial None marker only after reaching that boundary and while the current message reports num_pending == 0
• add deterministic regression tests for both premature-completion modes

Problem

KeyValue.keys() and history() stop at the first None produced by watch(). Two independent mechanisms can emit that marker before the initial replay actually completed:

1. Stale / non-monotonic per-message pending. The watcher emits None whenever an individual message reports metadata.num_pending == 0. Under concurrent KV writes that value can reach zero and then increase again. In a three-node NATS 2.14.3 trace, revision 1996 reported pending 0 and caused None; revisions 1997 and 1998 then arrived behind it with pending 10 and 9. Both keys were acknowledged and live before watcher creation, so keys() returned a partial result.
2. In-flight deliveries at setup. The setup path compares server num_pending with the local subscription delivery counter, which does not include messages the consumer has already sent but that are still in flight. This is the case described in kv.watch/history race can emit initial None too early and cause spurious NoKeysError #842.

Why not a count-only boundary

A pure delivery-count boundary (delivered.consumer_seq + num_pending at consumer creation, emit None after that many callbacks) fixes both races above but introduces an identity-displacement error with history=1: if key A is still pending and a writer updates A while inserting a new key C, the pending A revision is erased and C takes its slot in the frozen count. The count is then satisfied before the new A revision arrives, and A is dropped — the same user-visible bug.

Fix

The consumer-info snapshot boundary is a reliable minimum: at least that many deliveries must arrive before initial data can be complete. The per-message num_pending == 0 signal is a necessary drain condition but can fire early. The patch therefore requires both:

• the callback delivery count has reached the frozen snapshot boundary, and
• the current message reports num_pending == 0.

The count guard filters stale early zeros; the pending guard drains displaced updates. Liveness is no worse than the released code, which also required a num_pending == 0 message to terminate. The watcher API and continued live updates after None are unchanged.

Related: #842

Verification

• test_watch_initial_marker_ignores_transient_zero_pending fails against released nats-py 2.15.0 (marker emitted at the stale zero) and passes with this patch
• test_watch_initial_marker_waits_for_displaced_update fails against a count-only boundary and passes with this patch
• 23 legacy KV tests and 27 subtests pass against NATS 2.14.3
• Ruff and formatting checks pass
• the original three-node churn reproducer fails immediately with nats-py 2.14.0; the patched package completed the instrumented churn workload without omitting a continuously-live key

[caspervonb]

This fixes the reported early-completion races, but the frozen delivery boundary can become permanently unreachable. With include_history=True, a consumer can snapshot 3 pending revisions, then a concurrent kv.purge() rollup collapses them to one marker. That marker arrives with num_pending == 0, but _received == 1 never reaches _pending == 3, so None is never emitted and history() hangs. I verified NATS Server 2.14 changes pending from 3 to 1 in this case, and a targeted regression times out. Please reconcile legitimate boundary shrinkage or add a safe bounded fallback, plus a regression test.

[d3vv3]

Thanks for the feedback! Good catch

[Y0-L0]

Your updated code looks great!

It fixes the additional edge case we identified in the initial version.
My reproducer script from the initial version no longer surfaces any bugs. 🚀

Details on the edge-case:

This was only present in the initial version of this PR. Not the current version!

The boundary is a fixed count: at watcher creation consumer_info() reports N existing keys, and the initial None marker is emitted once N messages have been delivered.
But which N messages get delivered isn't pinned to the keys that existed at creation. Consider a bucket with two keys, and concurrent writes during the initial replay:

Bucket at watcher creation: A@seq1, B@seq2 → num_pending = 2, boundary frozen at 2
During the initial drain, concurrently:
PUT C (new key) → seq 3
PUT A (update) → seq 4 (with history=1, A@seq1 is discarded; A moves to seq 4)

Delivery is ascending stream sequence:
recv #1: B (seq 2)
recv #2: C (seq 3) → count reaches 2 == boundary → None emitted here
recv #3: A (seq 4) → arrives AFTER None, treated as a live update

keys() returns {B, C} and drops A, a key that existed the entire time.

The count is correct (2 → 3 keys), but the boundary counts deliveries, not identities: the newly inserted key C (lower seq) fills a slot, while the updated pre-existing key A
(higher seq) is pushed past the boundary. Any insert-races-update where the new key gets a lower sequence than the update reproduces this.

[d3vv3]

Quick note on the red checks - I think they're not from this PR. The nats-key-value job can't find pytest, and because that matrix has no fail-fast: false, it takes the others down with it. Same failure on #1001. The jobs that actually cover this change are green. Happy to send a small separate PR for it if you want.
Also, while reading the new workspace layout I noticed nats-key-value has the same problem this PR fixes - keys() and history() still stop at the first message reporting zero pending, and the watcher's boundary collapses on a stale zero. I have a small reproducer that fails on main.
I've left it out here since it's a different design and the semantics in that package are yours to call.

[caspervonb]

On the checks, yeah uv had a change in semantics, and wasn't pinned.