Add consumer_limits support to StreamConfig

[caspervonb]

Partially resolves #760

[Copilot]

Pull Request Overview

This PR adds support for NATS JetStream stream consumer limits, a feature introduced in nats-server 2.10.0. Consumer limits allow streams to define default constraints for newly created consumers.

Key Changes

• Added StreamConsumerLimits dataclass to represent consumer limits configuration with inactive_threshold and max_ack_pending fields
• Integrated consumer limits into StreamConfig as an optional field
• Added comprehensive test coverage for various consumer limits scenarios

Reviewed Changes

Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.

File	Description
nats/src/nats/js/api.py	Implements StreamConsumerLimits dataclass with serialization/deserialization logic and adds consumer_limits field to StreamConfig
nats/tests/test_js.py	Adds test cases covering full limits, partial limits (only one field set), and consumer inheritance behavior

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

The as_dict method has a bug: it unconditionally sets inactive_threshold even when it's None. The _to_nanoseconds method converts None to 0 (line 85), so when inactive_threshold is None, it will be sent as 0 to the server instead of being omitted.

This breaks the test case on line 4946 where only max_ack_pending is set - the server will receive inactive_threshold=0 instead of it being omitted.

The fix should conditionally set inactive_threshold only when it's not None:

def as_dict(self) -> Dict[str, object]:
    result = super().as_dict()
    if self.inactive_threshold is not None:
        result["inactive_threshold"] = self._to_nanoseconds(self.inactive_threshold)
    return result

This pattern is consistent with how other classes handle optional nanosecond fields (e.g., see ConsumerConfig.as_dict() lines 581-588).

Suggested change

	result["inactive_threshold"] = self._to_nanoseconds(self.inactive_threshold)
	if self.inactive_threshold is not None:
	result["inactive_threshold"] = self._to_nanoseconds(self.inactive_threshold)