Backport sourcing-consumer config for ADR-60

[caspervonb]

ADR-60 lets clients pre-create a push-durable consumer for sourcing/mirroring from workqueue or interest streams. This adds StreamSource.consumer (StreamConsumerSource(name, deliver_subject)) and AckPolicy.FLOW_CONTROL, which the server requires on that consumer — without it the source is accepted but fails at runtime with stream source consumer requires flow control ack policy (10217).

Includes an end-to-end test sourcing from a workqueue stream (skipped on servers older than 2.14).

[synadia-claude-reviewer]

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

---

PR Review: Backport sourcing-consumer config for ADR-60

• Gather PR diff and context
• Review api.py changes
• Review test_js.py changes
• Post findings

---

Summary

The ADR-60 core changes (StreamConsumerSource + StreamSource.consumer) are correct and well-tested. One issue worth addressing:

Out-of-scope / untested changes — The diff also includes ADR-50 atomic-batch-publish additions that are unrelated to the PR description:

• Three new Header enum values: BATCH_COMMIT, BATCH_ID, BATCH_SEQUENCE
• Two new PubAck fields: batch_id, batch_size
• PubAck.from_response and PubAck.as_dict overrides (remapping batch→batch_id, count→batch_size)

None of these have test coverage. They should either be described in the PR body or split into a separate PR to keep scope clear.

ADR-60 changes are sound:

• StreamConsumerSource dataclass is minimal and correct; required fields (name, deliver_subject) match the server's expectation.
• StreamSource.consumer serialises and deserialises correctly via the existing Base._convert / Base.as_dict machinery.
• Tests cover serialisation, deserialisation, absent-consumer case, and round-trip equality — good coverage.

[caspervonb]

@claude please review again

[synadia-claude-reviewer]

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

---

PR Review: Backport sourcing-consumer config for ADR-60

• Gather PR diff and context
• Review api.py changes
• Review test_js.py changes
• Post findings

---

ADR-50 batch-publish changes are still present

The previous review flagged ADR-50 additions as out-of-scope, and you noted they were moved to #938 — but the current diff on this branch still includes them:

nats/src/nats/js/api.py lines 26–28 — three Header enum values:

BATCH_COMMIT = "Nats-Batch-Commit"
BATCH_ID = "Nats-Batch-Id"
BATCH_SEQUENCE = "Nats-Batch-Sequence"

nats/src/nats/js/api.py lines 169–187 — two PubAck fields and two method overrides:

batch_id: Optional[str] = None
batch_size: Optional[int] = None

@classmethod
def from_response(cls, resp) -> PubAck:
    # remaps batch→batch_id, count→batch_size
    ...

def as_dict(self) -> Dict[str, object]:
    # remaps batch_id→batch, batch_size→count
    ...

None of these have test coverage. If #938 is intended to carry this, it looks like the rebase may not have cleanly excluded them from this branch. Worth double-checking — git log --oneline origin/main..HEAD or comparing the branch tip against main directly should confirm whether they're meant to be here.

---

ADR-60 changes look correct

StreamConsumerSource dataclass (api.py line 208–219):

• Minimal and correct. Inherits Base, exposes name: str and deliver_subject: str as required fields matching the server expectation.
• The docstring accurately describes the use case and required fields.

StreamSource.consumer field (api.py line 230):

• Optional[StreamConsumerSource] = None — correct typing and default.

from_response deserialization (api.py line 236):

• cls._convert(resp, "consumer", StreamConsumerSource) is correct. _convert handles the absent-field case by setting it to None and the present-field case by calling StreamConsumerSource.from_response(data).

as_dict serialization:

• No explicit handling needed: Base.as_dict() line 150 already calls .as_dict() on any field value that is a Base instance, so consumer serialises correctly without custom code in StreamSource.as_dict().

Tests

The five StreamConsumerSourceTest cases cover exactly the right surface:

Test	What it checks
test_stream_source_as_dict_with_consumer	Serialisation → correct nested dict
test_stream_source_as_dict_without_consumer	Absent consumer → key not present in dict
test_stream_source_from_response_with_consumer	Deserialisation → StreamConsumerSource instance
test_stream_source_from_response_without_consumer	Absent consumer → field is None
test_stream_source_consumer_round_trip	JSON encode → decode → equality

No issues with the test logic.

---

Summary

The ADR-60 additions are correct and well-tested. The only blocker is that the ADR-50 batch-publish changes (Header enum values, PubAck fields/overrides) appear to still be on this branch despite the claim they were moved to #938. If the rebase that was supposed to remove them didn't land, dropping those commits or cherry-picking just the ADR-60 commit would clean this up.

[caspervonb]

PTAL @claude

[synadia-claude-reviewer]

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

---

PR Review: Backport sourcing-consumer config for ADR-60

• Gather PR diff and context
• Review api.py changes
• Review test_js.py changes
• Post findings

---

Scope is now clean

The previous reviews flagged ADR-50 batch-publish additions (Header enum values, PubAck fields/overrides) as out-of-scope. Those changes are no longer in this PR's diff — they've since landed in origin/main via #922. The current diff is purely ADR-60.

---

api.py — Looks correct

StreamConsumerSource dataclass (lines 230–241):

• Inherits Base, exposes the two fields the server requires: name: str and deliver_subject: str.
• No optional fields, so Base.as_dict() serialises both unconditionally — no custom override needed.
• Docstring accurately describes the ADR-60 use case and the server requirement.

StreamSource.consumer field (line 252):

• Optional[StreamConsumerSource] = None — correct typing and default.

StreamSource.from_response addition (line 258):

• cls._convert(resp, "consumer", StreamConsumerSource) follows the same pattern as external and subject_transforms on the lines above it. The _convert helper handles both the absent-field case (None) and the present-field case (calls StreamConsumerSource.from_response(data)).

Serialisation via Base.as_dict():

• No explicit handling needed in StreamSource.as_dict(). The base as_dict() already calls .as_dict() on any field value that is a Base instance (line 169 of api.py), so the nested consumer dict is produced automatically. The absent-consumer case is also handled: None values are skipped (line 163).

---

test_js.py — Good coverage

The five StreamConsumerSourceTest cases cover exactly the right surface:

Test	What it checks
test_stream_source_as_dict_with_consumer	Serialisation → correct nested dict
test_stream_source_as_dict_without_consumer	Absent consumer → key not present in dict
test_stream_source_from_response_with_consumer	Deserialisation → StreamConsumerSource instance
test_stream_source_from_response_without_consumer	Absent consumer → field is None
test_stream_source_consumer_round_trip	JSON encode → decode → equality

No issues with the test logic.

---

Summary

The ADR-60 additions are correct, consistent with the existing ExternalStream/SubjectTransform patterns, and well-tested. No further changes needed from my side — this looks good to merge.