fix(nats): prevent KV scan consumer leaks

[dr-hoseyn]

Summary

• replace full-bucket kv.keys() scans with prefix-scoped metadata watchers
• always stop and explicitly delete the temporary JetStream consumer on both success and failure
• preserve the in-memory KV behavior used by tests and single-process fallbacks

This prevents large shared KV buckets from turning each per-node queue lookup into a full scan. It also avoids abandoned consumers accumulating after request timeouts, which otherwise creates a feedback loop of higher NATS load, sync delays, and subscription timeouts.

Type of change

• Bug fix
• New feature
• Breaking change
• Refactor / cleanup
• Documentation
• Tests / CI

Checklist

• I tested the change locally or explained why it cannot be tested.
• I added or updated tests for behavior changes.
• I updated documentation, translations, or examples if needed. (Not applicable)
• I checked database migrations when models or schema changed. (No schema changes)
• I did not include secrets, tokens, private keys, or unrelated changes.

Testing

• python -m uv run --frozen pytest tests/test_nats_node_memory.py tests/test_node_manager_sync.py tests/test_connect_concurrency.py tests/test_record_usages.py -q — 24 passed
• python -m ruff check app/nats/kv_cas.py tests/test_nats_node_memory.py — passed
• validated against a live multi-worker deployment: queues drained, transient consumer count returned to zero, and subscription requests completed successfully

Screenshots

Not applicable.

Notes for reviewers

KeyValue.Watcher.stop() only unsubscribes the client; nats-py leaves the ephemeral consumer until its inactivity threshold. The explicit consumer deletion in finally makes cleanup immediate and is covered for both normal completion and exceptions.

Summary by CodeRabbit

• Bug Fixes
  • Improved key listing for NATS key-value stores by applying prefix filtering server-side.
  • Excluded deleted entries from key listings.
  • Ensured watcher resources are cleaned up reliably, including when errors occur.
• Tests
  • Added coverage for filtering, cleanup, and error handling during key listing.

[coderabbitai]

No actionable comments were generated in the recent review. 🎉

ℹ️ Recent review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: CHILL

Plan: Pro Plus

Run ID: 79b2ff45-621d-465e-969f-1b9b43af0125

📥 Commits

Reviewing files that changed from the base of the PR and between 3331421 and 120ddd4.

📒 Files selected for processing (2)

• app/nats/kv_cas.py
• tests/test_nats_node_memory.py

---

Walkthrough

kv_list_keys now uses a prefix-filtered watcher for real NATS KeyValue instances. It collects active keys and cleans up the watcher and consumer after success or failure. Tests cover filtering, results, cleanup, and error propagation.

Changes

NATS KV key listing

Layer / File(s)	Summary
Filtered watcher lifecycle app/nats/kv_cas.py	The real NATS KeyValue path uses a prefix-filtered, metadata-only watcher. It collects non-delete entries and cleans up the watcher and consumer on success or error.
Watcher lifecycle validation tests/test_nats_node_memory.py	Mock fixtures and tests verify watcher options, collected keys, shutdown, consumer deletion, and iteration error propagation.

Estimated code review effort: 3 (Moderate) | ~20 minutes

Merge Risk: ⚪ Minimal · up to 120dd

This change narrows KV lookups and cleans up temporary consumers to prevent scan overhead and consumer accumulation; no actionable merge-blocking risk remains beyond normal checks and review.

Sequence Diagram(s)

sequenceDiagram
  participant kv_list_keys
  participant KeyValue
  participant JetStream
  kv_list_keys->>KeyValue: start prefix-filtered watcher
  KeyValue->>JetStream: create metadata-only consumer
  JetStream-->>kv_list_keys: provide KV updates
  kv_list_keys->>KeyValue: stop watcher
  kv_list_keys->>JetStream: delete consumer

Loading

Suggested reviewers: m03ed

Poem

A rabbit watched the key stream glow,
Through filtered paths the updates flow.
On error, too, the cleanup’s bright—
The watcher stops before the night.
The consumer hops away.

🚥 Pre-merge checks | ✅ 5

✅ Passed checks (5 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title clearly summarizes the main change: preventing temporary NATS KV scan consumer leaks.
Docstring Coverage	✅ Passed	No functions found in the changed files to evaluate docstring coverage. Skipping docstring coverage check.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands.}

[dr-hoseyn]

@coderabbitai review

[coderabbitai]

✅ Action performed

Review finished.

Note: CodeRabbit is an incremental review system and does not re-review already reviewed commits. This command is applicable only when automatic reviews are paused.