Add nats-jetstream package #734

caspervonb · 2025-09-23T10:04:49Z

This adds a new modern (following the newer simplified interface) JetStream package under the name nats-jetstream which implements the following:

Depends on #732

aricart

Not quite sure why we are going to look at the schemas and validate them... But LGTM

nats-jetstream/src/nats/jetstream/api/client.py

caspervonb

Some nits

nats-jetstream/src/nats/jetstream/message.py

nats-jetstream/src/nats/jetstream/stream.py

nats-jetstream/src/nats/jetstream/consumer/__init__.py

caspervonb

More nits

nats-jetstream/src/nats/jetstream/stream.py

caspervonb · 2025-09-23T16:23:21Z

nats-jetstream/src/nats/jetstream/__init__.py

+        self,
+        stream_name: str,
+        name: str,
+        durable_name: str | None = None,


Pack into kwargs

Suggested change

durable_name: str | None = None,

caspervonb · 2025-09-23T16:23:33Z

nats-jetstream/src/nats/jetstream/__init__.py

+        self,
+        stream_name: str,
+        name: str,
+        durable_name: str | None = None,


Pack into kwargs

Suggested change

durable_name: str | None = None,

wallyqs · 2025-09-23T16:35:06Z

nats-jetstream/src/nats/jetstream/util.py

+    Returns:
+        A unique inbox subject string in the format "_INBOX.{uuid}"
+    """
+    return f"_INBOX.{uuid.uuid4().hex}"


what happened to the nuid?

wallyqs

Do you have:

Clustering based test cases
Cluster reconnect test cases resuming fetch

?

philpennock

Reviewed through end of the src/nats/jetstream/api/ folder, need to resume from the consumer/ sibling to that when I next look at this again. Sending what I've looked at so far though. (Will be around on Thursday to resume).

philpennock · 2025-09-23T23:46:47Z

nats-jetstream/examples/pull_consume_callback.py

+    while True:
+        await asyncio.sleep(0.5)
+        if nc.status != ClientStatus.CONNECTED:
+            continue


Since this is example code for people to start from, and people do Interesting Things when copy/pasting without full understanding, I think it's worth a small comment:

if nc.status != ClientStatus.CONNECTED: # NB: this is only safe because of the sleep above; beware busy loops continue

Yeah example got stale, need new ones.

philpennock · 2025-09-23T23:47:46Z

nats-jetstream/examples/pull_consume_callback.py

+        try:
+            await js.publish("FOO.TEST1", f"msg {i}".encode())
+        except Exception as err:
+            print("pub error: ", err)


grumble (unix stdio pedant) : , file=sys.stderr, flush=True) (and import sys above)

nats-jetstream/examples/pull_consume_iterator.py

philpennock · 2025-09-23T23:54:19Z

nats-jetstream/schemas/jetstream/api/v1/account_info_response.json

@@ -0,0 +1,24 @@
+{
+  "$schema": "http://json-schema.org/draft-07/schema#",
+  "$id": "https://nats.io/schemas/jetstream/api/v1/account_info_response.json",


Uh, are we expecting every client library for each language to hold a copy of these? That seems ... sub-optimal. Shouldn't there be one primary authoritative location, designed to be pulled in via git submodules or whichever other repo-linking poison folks prefer?

(I'm assuming no point reviewing the schemas in this PR so skipping past them.)

Agreed, those schemas live and are maintained in jsm.go, it'd be hard to keep it up to date across many repos.

These are pulled from there, just with a little bit of filtering unwanted files. I'd do a submodule if we had nats-io/schema or something.

I guess there is a task to fetch them from jsm.go?

Yeah tools/fetch_schemas.py

philpennock · 2025-09-23T23:56:02Z

nats-jetstream/src/nats/jetstream/api/__init__.py

+    StreamState,
+    SubjectTransform,
+    Tier,
+    LostStreamData,


philpennock · 2025-09-24T00:07:39Z

nats-jetstream/src/nats/jetstream/api/client.py

+def check_response(
+    data: Any, expected_type: type[ResponseT]
+) -> tuple[bool, set[str] | None, set[str] | None]:
+    # TODO check for missing and unknown keys


For clarity: are you expecting this TODO to be done before this PR moves out of draft, or for some later milestone?

philpennock · 2025-09-24T00:10:12Z

nats-jetstream/src/nats/jetstream/api/client.py

+
+        consumer_name = consumer_config.get("name")
+        if not consumer_name:
+            raise ValueError("name is required")


This would be more friendly if all 3 were collected, and then if not (a and b and c) used to return a ValueError listing all the missing fields at once.

(nb: this is an opinion and not a blocking issue, happy to be ignored on this)

philpennock · 2025-09-24T03:34:07Z

nats-jetstream/src/nats/jetstream/api/client.py

+        mirror = kwargs.get("mirror")
+        subjects = kwargs.get("subjects")
+        if mirror is not None and subjects:
+            raise ValueError("Cannot specify both mirror and subjects")


I look at logic like this and can't help but think "Isn't this schema enforcement? If we have all these .json schema files, why aren't the client functions (at least for infrequent ops like stream create) just validating the config against the schema file instead of implementing policy checks manually?"

philpennock · 2025-09-24T03:43:53Z

nats-jetstream/src/nats/jetstream/api/client.py

+        response_type: type[ResponseT],
+        raise_on_error: Literal[True] = True,


This, and the same in the next @overload, is missing the timeout parameter; the implementation is assignable to these function signatures so it should work, but it looks like this means that type-checkers should reject attempts to actually provide the timeout parameter. Have I misunderstood? Is this intentional? Or is it just a forgotten parameter for the signature?

philpennock · 2025-09-24T04:05:12Z

nats-jetstream/src/nats/jetstream/api/client.py

+        raise_on_error: bool = True,
+        timeout: float = 5.0,
+    ) -> ResponseT | ErrorResponse:
+        request_id = str(uuid.uuid4())[:8]


This is reading 16 bytes from /dev/urandom every time, just to return the representation of the first 4 bytes. Aside from making me twitch (which is probably amusing 😉), that's also going to be a device read on every JSON request.

The secrets module was introduced in Python 3.6; secrets.token_hex(4) will be more efficient, but still hit /dev/urandom. But the only goal is a distinct ID, there's no defense against other clients knowing the ID, right? Presumably not using an incrementing integer to avoid threading issues, but should then be using a client-side cache to prevent re-use, so just rolling the dice here (but the die has a lot of sides). I'd be tempted to use an in-process PRNG seeded once and just roll with that.

For now, just use secrets.token_hex(4) please and remember to benchmark to see if this ever matters with lots of small JSON requests.

philpennock

Reviewed the rest, except I skimmed the test files just enough to confirm they were limited to behavior tests, not reflective integrity tests.

nats-jetstream/src/nats/jetstream/consumer/__init__.py

philpennock · 2025-09-25T19:25:53Z

nats-jetstream/src/nats/jetstream/consumer/__init__.py

+    """Protocol for a batch of messages retrieved from a JetStream consumer."""
+
+    def __aiter__(self) -> AsyncIterator[Message]:
+        """Return self as an async iterator."""


Here and in MessageStream below, is this really self?

nats-jetstream/src/nats/jetstream/consumer/pull.py

philpennock · 2025-09-25T19:33:37Z

nats-jetstream/src/nats/jetstream/consumer/pull.py

+                            raise StopAsyncIteration
+                        case _:
+                            self._error = Exception(
+                                f"Status {raw_msg.status.code}: {raw_msg.status.description or ''}"


minor nit: can we use or '(missing description)' to make it clearer if this exception is ever seen that there was something expected that's missing?

I'm thinking we shouldn't swallow unexpected errors.

nats-jetstream/src/nats/jetstream/message.py

nats-jetstream/src/nats/jetstream/stream.py

philpennock · 2025-09-25T21:33:21Z

nats-jetstream/src/nats/jetstream/stream.py

+            try:
+                headers_bytes = base64.b64decode(message["hdrs"])
+                headers_str = headers_bytes.decode("utf-8")
+                if headers_str.startswith("NATS/1.0\r\n"):


Should there be an exception raised if the headers don't start with this?

philpennock · 2025-09-25T21:37:00Z

nats-jetstream/tools/fetch_schemas.py

+def main():
+    """Main entry point."""
+    # Get the root directory of our project
+    root_dir = Path(__file__).parent.parent


This is fragile to current location in the git repo. If you're going to invoke git anyway for this rare maintenance operation, then git rev-parse --show-toplevel is your friend here.

Signed-off-by: Casper Beyer <[email protected]>

- Add validation to reject both max_messages and max_bytes simultaneously - Set batch size to 1,000,000 when using max_bytes for better throughput - Add heartbeat timer pause/resume on disconnect/reconnect - All three fixes address the non-compliance issues identified in ADR-37

- Remove test that used both max_messages and max_bytes together - Add test validating rejection of both parameters - Fix threshold_bytes test to use only max_bytes per ADR-37

Signed-off-by: Casper Beyer <[email protected]>

caspervonb force-pushed the add-nats-jetstream-package branch from cc17acd to b383f0b Compare September 23, 2025 10:39

aricart approved these changes Sep 23, 2025

View reviewed changes

nats-jetstream/src/nats/jetstream/api/client.py Outdated Show resolved Hide resolved

caspervonb mentioned this pull request Sep 23, 2025

Add nats-server package #731

Merged

caspervonb commented Sep 23, 2025

View reviewed changes

wallyqs reviewed Sep 23, 2025

View reviewed changes

oliverlambson mentioned this pull request Sep 24, 2025

Resolve #682 #683

Closed

3 tasks

philpennock reviewed Sep 24, 2025

View reviewed changes

philpennock reviewed Sep 25, 2025

View reviewed changes

caspervonb force-pushed the add-nats-client-package branch 4 times, most recently from 7acfd1f to 2e13fe5 Compare October 7, 2025 15:02

caspervonb force-pushed the add-nats-jetstream-package branch 4 times, most recently from 1da8646 to b6b8a1a Compare October 7, 2025 15:38

caspervonb force-pushed the add-nats-client-package branch 2 times, most recently from 3cd0ce1 to 171ad83 Compare October 8, 2025 20:16

caspervonb added 10 commits October 8, 2025 22:43

Add nats-jetstream package

6d19ac8

Signed-off-by: Casper Beyer <[email protected]>

Remove is_active property from MessageHandler

324aa15

Add type annotations for MessageHandler attributes

43ccf65

Add type annotations for PullMessageBatch attributes

2e8f08e

Add type annotations to PullMessageStream attributes

b09af62

Add type annotations for PullConsumer attributes

0df512c

Rename batch parameter to max_messages in consume method

d811201

Add keyword-only separator to consume method signatures

d908a7b

Remove redundant expires parameter from consume

a2f69e5

Move __all__ definition to end of message.py

1a679ac

caspervonb added 26 commits October 20, 2025 17:49

Fix skipped tests

a47c7c2

Signed-off-by: Casper Beyer <[email protected]>

Add tests for multiple filter_subjects

3e35f47

Signed-off-by: Casper Beyer <[email protected]>

Refine Pull consumer status code handling

eba8a6f

Signed-off-by: Casper Beyer <[email protected]>

Add threshold params to pull message stream

13d092e

Signed-off-by: Casper Beyer <[email protected]>

Add heartbeat monitoring to pull consumer and stream

e9f2097

Signed-off-by: Casper Beyer <[email protected]>

Handle 409 and pending headers in pull consumer

07e65d9

Signed-off-by: Casper Beyer <[email protected]>

Compute message size and use in pull consumer

650df40

Signed-off-by: Casper Beyer <[email protected]>

Accept ConsumerConfig in update_consumer

df891bf

Signed-off-by: Casper Beyer <[email protected]>

Validate ConsumerConfig and use in update_consumer

2d30ae1

Signed-off-by: Casper Beyer <[email protected]>

Make all api client request methods positional and use request dicts

516bb5d

Signed-off-by: Casper Beyer <[email protected]>

Apply code style and formatting fixes

4076d94

Signed-off-by: Casper Beyer <[email protected]>

Use Stream API for consumer pause/resume in tests

e6b61bb

Signed-off-by: Casper Beyer <[email protected]>

Use RFC3339 strings for timestamps in API types

691a39c

Signed-off-by: Casper Beyer <[email protected]>

Convert ts fields to datetime

3135f85

Signed-off-by: Casper Beyer <[email protected]>

Use datetime for stream and consumer timestamps

4bbe30f

Signed-off-by: Casper Beyer <[email protected]>

Use timedelta for duration fields

7832e77

Signed-off-by: Casper Beyer <[email protected]>

Fix timedelta tests

4173324

Signed-off-by: Casper Beyer <[email protected]>

Add stream republish, transform, and metadata tests

d7c97ac

Signed-off-by: Casper Beyer <[email protected]>

Run ruff format

db9fec7

Signed-off-by: Casper Beyer <[email protected]>

Update tests to enforce ADR-37 semantics

acec368

- Remove test that used both max_messages and max_bytes together - Add test validating rejection of both parameters - Fix threshold_bytes test to use only max_bytes per ADR-37

Fix unused variable in test_consumer_info_not_paused_initially

9c7dd6b

Standardize UTF-8 string quoting in Message.size

384f202

Signed-off-by: Casper Beyer <[email protected]>

Retry publish on NoResponders errors

1950cc3

Signed-off-by: Casper Beyer <[email protected]>

Add create_or_update_consumer to JetStream and Stream

ba1e69b

Signed-off-by: Casper Beyer <[email protected]>

Overhaul error handling

185c9f9

Signed-off-by: Casper Beyer <[email protected]>

caspervonb force-pushed the add-nats-jetstream-package branch from 1274414 to 185c9f9 Compare October 22, 2025 06:17

caspervonb mentioned this pull request Oct 22, 2025

TimeoutError on publish during a server restart #751

Open

caspervonb force-pushed the add-nats-client-package branch from a3fc895 to 71eb800 Compare October 22, 2025 11:46

caspervonb mentioned this pull request Oct 30, 2025

Fix opt_start_time usage and other datetime fields #535

Closed

		response_type: type[ResponseT],
		raise_on_error: Literal[True] = True,

Add nats-jetstream package #734

Are you sure you want to change the base?

Add nats-jetstream package #734

Uh oh!

Conversation

caspervonb commented Sep 23, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

aricart left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

caspervonb left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

caspervonb left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

wallyqs left a comment

Choose a reason for hiding this comment

Uh oh!

philpennock left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

philpennock left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

caspervonb commented Sep 23, 2025 •

edited

Loading