Backport sourcing-consumer config for ADR-60 - #937
Conversation
ADR-60 lets clients pre-create a push-durable consumer and pass it via StreamSource.consumer when sourcing or mirroring from a workqueue or interest stream. Without the field the server returns SOURCE_DURABLE_CONSUMER_CFG_INVALID.
|
Claude finished @caspervonb's task in 2m 5s —— View job PR Review: Backport sourcing-consumer config for ADR-60
SummaryThe ADR-60 core changes ( Out-of-scope / untested changes — The diff also includes ADR-50 atomic-batch-publish additions that are unrelated to the PR description:
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:
|
![synadia-claude-reviewer[bot]](https://avatars.githubusercontent.com/u/132956691?s=60&v=4){kind=small}
|
@claude please review again |
|
Claude finished @caspervonb's task in 2m 9s —— View job PR Review: Backport sourcing-consumer config for ADR-60
ADR-50 batch-publish changes are still presentThe 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:
BATCH_COMMIT = "Nats-Batch-Commit"
BATCH_ID = "Nats-Batch-Id"
BATCH_SEQUENCE = "Nats-Batch-Sequence"
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 — ADR-60 changes look correct
TestsThe five
No issues with the test logic. SummaryThe ADR-60 additions are correct and well-tested. The only blocker is that the ADR-50 batch-publish changes ( |
…sumer-config # Conflicts: # nats/tests/test_js.py
|
PTAL @claude |
|
Claude finished @caspervonb's task in 1m 58s —— View job PR Review: Backport sourcing-consumer config for ADR-60
Scope is now cleanThe previous reviews flagged ADR-50 batch-publish additions (
|
| 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.
The server requires the pre-created sourcing consumer to use the flow_control ack policy; without it the source fails at runtime with "stream source consumer requires flow control ack policy" (10217). Expose the policy on the enum and exercise the full sourcing flow from a workqueue stream against a 2.14+ server.
|
PTAL @claude — the branch has moved since the last review: merged |
|
Claude finished @caspervonb's task in 2m 24s —— View job PR Review: Backport sourcing-consumer config for ADR-60
What's new since the last reviewTwo additions on top of the already-approved ADR-60 core (
|
| Check | What it validates |
|---|---|
cinfo.config.ack_policy == AckPolicy.FLOW_CONTROL |
Server round-trips the ack policy correctly |
consumer.name == "C" / consumer.deliver_subject == "deliver.up" |
StreamConsumerSource deserialized from add_stream response |
Poll loop → info.state.messages == 3 |
Messages actually flow through the sourced stream |
info.sources[0].error is None |
No sourcing error reported by the server |
pytest.skip on server < 2.14 |
Correct feature gating |
Minor nit — potential None dereference:
assert info.sources[0].error is NoneStreamInfo.sources is typed Optional[List[StreamSourceInfo]]. If the server response ever omits the sources field, this raises TypeError rather than a descriptive assertion failure. A guard would be safer:
assert info.sources is not None and len(info.sources) == 1
assert info.sources[0].error is NoneIn practice this won't happen for a sourcing stream, so it's not a blocker — just a robustness note.
Summary
The AckPolicy.FLOW_CONTROL addition is correct, well-documented, and properly used in the test. The end-to-end test is a solid integration check that covers the full ADR-60 sourcing flow. The one minor nit (unguarded info.sources[0]) is non-blocking. This looks good to merge.
renerocksai
left a comment
renerocksai
left a comment
There was a problem hiding this comment.
Please bear in mind, I am fairly new to nats-py and might be wrong in my assessments. The review focused on idiomatic / good-quality async Python code; I haven't evaluated NATS-specific design choices.
Nothing blocking came up. Two small tightenings — hedged inline, take or leave.
Sourcing consumers are acknowledged by the sourcing server through flow control messages rather than an ack reply per message, so the auto-ack wrapper sent a pointless ack and marked each message acked, making a later legitimate ack raise MsgAlreadyAckdError. Compare by value while here so a policy that arrived from the server as a plain string is matched too.
2 participants
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)) andAckPolicy.FLOW_CONTROL, which the server requires on that consumer — without it the source is accepted but fails at runtime withstream 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).