Add create_or_update_stream to JetStream

[caspervonb]

The JetStream context exposes create_or_update_consumer but no stream equivalent, so callers have to branch on existence for the most common stream operation. Adds an idempotent upsert that attempts an update first and falls back to create when the stream does not yet exist, propagating any other error and returning a Stream handle like create_stream.

[synadia-claude-reviewer]

Claude finished @caspervonb's task in 2m 23s —— View job

---

PR Review

• Gather context
• Read PR diff and relevant source files
• Analyze for correctness, asyncio patterns, type annotations, API consistency
• Post review findings

---

Summary

The implementation is clean and correct. create_or_update_stream fills a real API gap — the symmetry with create_or_update_consumer is good, and the method signature/overload pattern mirrors create_stream exactly.

What looks good:

• Update-first strategy is correct and efficient for the common "stream already exists" case
• config_dict = config.to_request() is computed once and safely reused across both API calls (unpacking doesn't mutate the dict)
• Returns Stream (not StreamInfo) — consistent with create_stream, different from update_stream. This is the right call.
• Tests cover all four meaningful cases: absent stream, present stream, StreamConfig object, missing name
• Error handling: StreamNotFoundError is caught, all other errors propagate — matching the documented behaviour

One thing to decide: There's a TOCTOU race under concurrent access — see the inline comment on line 503. Two clients simultaneously hitting a non-existent stream will both get StreamNotFoundError from stream_update, both attempt stream_create, and the second will surface StreamNameAlreadyInUseError rather than succeeding silently. For typical single-client stream setup this is fine. If you want true concurrent idempotency, the fix is a three-step try: update → create → update-again-on-name-collision. Either choice is reasonable; just worth making intentionally.

No asyncio issues, no resource leaks, no type annotation problems beyond what already exists in the codebase.

[synadia-claude-reviewer]

Under concurrent access, two clients can both receive StreamNotFoundError from stream_update and then race on stream_create. The second caller would propagate StreamNameAlreadyInUseError rather than succeeding as expected for an "idempotent upsert."

For most single-client setups this is fine, but if truly idempotent behavior under concurrent access is needed, the StreamNameAlreadyInUseError in the fallback stream_create could be caught and resolved with a final stream_update:

try:
    response = await self._api.stream_update(config.name, **config_dict)
except StreamNotFoundError:
    try:
        response = await self._api.stream_create(config.name, **config_dict)
    except StreamNameAlreadyInUseError:
        response = await self._api.stream_update(config.name, **config_dict)

If the current behaviour (propagate StreamNameAlreadyInUseError) is intentional, the docstring's "any other error is propagated" already covers it — just worth a conscious decision either way.

[caspervonb]

Review comments addressed, PTAL @claude

[synadia-claude-reviewer]

Claude finished @caspervonb's task in 1m 33s —— View job

---

Re-review: Race condition addressed correctly ✓

• Gather context
• Read updated diff and source files
• Analyze race condition fix
• Post review findings

---

The update-first → create → update-on-collision three-step pattern is correctly implemented and StreamNameAlreadyInUseError is already present in the import list at line 27, so nothing is missing there.

Implementation (__init__.py lines 500–508):

try:
    response = await self._api.stream_update(config.name, **config_dict)
except StreamNotFoundError:
    try:
        response = await self._api.stream_create(config.name, **config_dict)
    except StreamNameAlreadyInUseError:
        # Lost a create race with a concurrent caller; the stream
        # exists now, so updating keeps the operation idempotent.
        response = await self._api.stream_update(config.name, **config_dict)

This is exactly the pattern suggested in the prior review — clean, correct.

Race test (test_stream.py lines 130–154):

The test correctly exercises the specific race window:

1. Creates the stream at server level (so it exists on the server)
2. Patches stream_update to raise StreamNotFoundError once, simulating the window where the stream appeared after our first update attempt
3. The fallback stream_create then hits StreamNameAlreadyInUseError against the server
4. The final stream_update succeeds with the new config

Both assertions are meaningful: update_calls == 2 verifies the recovery path was taken, and set(stream.info.config.subjects) == {"FOO.*", "BAR.*"} verifies the final state is correct (not stale from the first pre-existing config).

Minor nit (no action needed): The docstring still says "any other error is propagated" — technically StreamNameAlreadyInUseError is now also swallowed in the fallback path. But since the method contract is simply "idempotent upsert," the docstring remains accurate at the level of user intent. No change needed.

This PR looks good to me.