Conversation
…cies, dynamic batching by default, driver lifecycle fixes (v2.3.0)
There was a problem hiding this comment.
Pull request overview
This PR introduces a typed “policy object” layer for Messenger drivers (readers/publishers) to normalize legacy flat kwargs into immutable dataclasses, while also updating driver lifecycle management and adding extensive unit + NATS integration tests.
Changes:
- Added
DeliveryPolicy,ErrorPolicy/RetryPolicy, andBatchPolicy(frozen dataclasses) plus normalization/conflict detection. - Updated
MsgReader/MsgPublisherto accept policy objects (and keep legacy kwargs working), including dynamic batch sizing and retry pacing. - Added new policy test suites (unit + live NATS) and introduced driver lifecycle safety improvements (deregister on close, Messenger close safety net), plus minor infra/docs/version updates.
Reviewed changes
Copilot reviewed 15 out of 15 changed files in this pull request and generated 3 comments.
Show a summary per file
| File | Description |
|---|---|
| tests/test_policies.py | New unit tests covering policy semantics, validation, normalization conflicts, defaults, and lifecycle behavior. |
| tests/test_policies_nats.py | New NATS integration tests validating end-to-end policy-driven reads/publishes and deprecation behavior. |
| tests/conftest.py | Updates NATS testcontainer import path. |
| serverish/messenger/policies.py | Implements policy dataclasses, enums, and legacy-kwargs normalization with conflict detection. |
| serverish/messenger/msg_single_read.py | Deprecates get_singlereader() and adjusts single_read() to avoid deprecated factory. |
| serverish/messenger/msg_single_pub.py | Deprecates get_singlepublisher() and adjusts single_publish() to avoid deprecated factory. |
| serverish/messenger/msg_reader.py | Integrates delivery/error/batch policies; adds dynamic batching and retry pacing; refines “emptied” semantics. |
| serverish/messenger/msg_publisher.py | Integrates error/retry policies for ack retries; adds deprecation warning on publish-before-open; updates factory to accept kwargs. |
| serverish/messenger/msg_progress_pub.py | Updates factory to accept kwargs (policy passthrough). |
| serverish/messenger/msg_journal_pub.py | Updates factory to accept kwargs (policy passthrough). |
| serverish/messenger/messenger.py | Adds close() safety net; factory kwargs passthrough; deprecates single factories; driver deregistration/re-registration logic. |
| serverish/messenger/init.py | Re-exports policy symbols and UNLIMITED. |
| serverish/connection/connection.py | Switches DNS check from deprecated aiodns API to getaddrinfo. |
| pyproject.toml | Bumps package version to 2.3.0. |
| docs/design/POLICIES.md | Documents dynamic batching default and damping behavior. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| # getaddrinfo, not the deprecated gethostbyname (removed in aiodns 4 line) | ||
| await resolver.getaddrinfo(h, family=socket.AF_INET, type=socket.SOCK_STREAM) |
There was a problem hiding this comment.
Not the case for the aiodns line we depend on (>=3.2,<5; 4.0.4 in dev): DNSResolver.getaddrinfo(self, host, family=AF_UNSPEC, port=None, proto=0, type=0, flags=0) — port is optional. Verified empirically: test_dns exercises this path and passes on aiodns 4.0.4. No change needed.
| warnings.warn( | ||
| f"Publishing on a publisher that was never opened ({self}). Open it explicitly " | ||
| f"(await pub.open() or 'async with pub:') - publishing on a closed publisher " | ||
| f"will become an error in serverish 3.0; for one-shot use single_publish()", | ||
| DeprecationWarning, stacklevel=2) |
There was a problem hiding this comment.
Good catch — fixed. The publisher now tracks _was_opened and the warning says never-opened vs closed accordingly; added test_publish_on_closed_publisher_warns_accurately.
| dynamic = (self.batching.dynamic if self.batching.dynamic is not None | ||
| else self.batch_defaults.dynamic) | ||
| if not dynamic: | ||
| return self.batching.max_batch or self.static_batch_default | ||
| max_batch = self.batching.max_batch or self.batch_defaults.max_batch | ||
| rate = self._consumption_rate() |
There was a problem hiding this comment.
Partially agreed and fixed: in static mode (BatchPolicy(dynamic=False)) the legacy mutable reader.batch attribute is honored again (policy max_batch wins when set); covered by a test. In the new dynamic default the attribute is intentionally inert — pull sizing follows the measured consumption rate. Note batch was never public API in a release: it was a hard-coded self.batch = 100 through 2.2.x.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 17 out of 17 changed files in this pull request and generated 2 comments.
Suppressed comments (6)
serverish/messenger/msg_reader.py:482
st.ncounts every state-machine loop, not retry attempts (for example, auto-opening the reader increments it before the first error). Consequently,RetryPolicy(attempts=1)can allow zero retries, and successful reopen/continue steps also consume the retry budget. Track a dedicated retry-attempt counter and increment it only when the WAIT branch schedules a retry.
if not self._retry.keeps_retrying(st.n, time.monotonic() - st.error_since):
serverish/messenger/policies.py:242
- The exponentiation occurs before the
max_delayclamp, so an unlimited retry policy eventually raisesOverflowErrordespite having a finite cap. This affects the reader default during a sufficiently long outage; handle overflow by returning the finite cap (while preserving zero-delay behavior).
delay = (self.delay or 0.0) * (self.backoff or 1.0) ** attempt
max_delay = self.max_delay
if max_delay is not None and max_delay != UNLIMITED:
delay = min(delay, max_delay)
return delay
serverish/messenger/policies.py:4
- This newly added reference is broken in the resulting tree because the same PR deletes
docs/design/POLICIES.md; several other new comments and tests also reference it. Either retain the finalized design document or update all references to an existing document.
objects (design: docs/design/POLICIES.md):
serverish/messenger/policies.py:322
- This public policy docstring still says the driver default is static with batch 100, but this PR changes
MsgReader.batch_defaultsto dynamic (initial 2, five-second target, maximum 10,000). Update it so API documentation does not contradict runtime behavior.
None = driver default (today: static, max_batch=100).
doc/messenger.md:86
- This example publishes without opening the journal publisher, so it immediately triggers the deprecation warning introduced by this PR and demonstrates the pattern the preceding section tells users to avoid. Open it explicitly in the example.
journal = get_journalpublisher('tic.journal.mytelescope.pipeline')
await journal.info('Calibration started') # never raises by default (on_error=LOG)
serverish/messenger/policies.py:333
- The integer policy fields only receive range checks, so values such as
max_batch=1.5orinitial_batch=Trueare accepted and can make_next_batch_size()return a non-integer fetch count. Validate these fields as positive integers, consistently withDeliverFromSeqandRetryPolicy.attempts.
def __post_init__(self):
if self.max_batch is not None and self.max_batch < 1:
raise ValueError("BatchPolicy.max_batch must be >= 1")
if self.initial_batch is not None and self.initial_batch < 1:
raise ValueError("BatchPolicy.initial_batch must be >= 1")
| kwargs['raise_on_publish_error'] = self.errors.on_error is OnError.RAISE | ||
| if (self.errors is not None and self.errors.retry is not None | ||
| and self.errors.retry.attempts is not None | ||
| and self.errors.retry.attempts is not UNLIMITED): |
There was a problem hiding this comment.
Agreed — real bug, fixed. is not UNLIMITED replaced with != UNLIMITED (CPython: float('inf') is math.inf → False, so a caller's RetryPolicy(attempts=float('inf')) passed validation by value but slipped past the identity check into int(inf) → OverflowError). policies.py already used equality everywhere; this was the only identity check. Regression test added (test_publisher_unlimited_attempts_by_value).
| self.reader._last_message_time = time.monotonic() | ||
| if len(self.reader.messages) == 0: | ||
| self.reader._consume_stamps.append(time.monotonic()) # feeds dynamic batch sizing | ||
| if len(self.reader.messages) == 0 and not self.reader._server_pending: |
There was a problem hiding this comment.
Disagreed — this is deliberate graceful degradation, not an oversight. _server_pending is None means the server never told us anything (non-JetStream message, exotic setup), and in that mode it stays None forever. Requiring an explicit 0 would make wait_for_empty() never fire there — turning degraded metadata into a hang, which is strictly worse than falling back to the pre-2.3 local-drain semantics. So the contract is: server said 0 → confirmed empty; server said N>0 → not empty; server silent → legacy behavior. It's the same degradation policy fetch_available uses for its end-of-data detection, and the inline comment states it ('num_pending == 0 or unknown'). On any healthy JetStream path num_pending is always present, so the strict contract holds exactly where it can.
v2.3.0: Policy objects for Messenger drivers
Implements
docs/design/POLICIES.md— typed, grouped driver configurationwith full backward compatibility and isolation from the raw NATS API.
Direct answer to user feedback: publishers obtained via
get_publisher()were unconfigurable, and driver parameters had accreted chaotically.
Policies
DeliveryPolicyvariants —DeliverAll/Last/New/LastPerSubject/ FromTime(start)/FromSeq(seq)+until(FOREVER|END_OF_DATA).Invalid combinations are unrepresentable.
deliver_policyis dual-typed:the string form (
'last') is permanent sugar, not deprecated.ErrorPolicywith nestedRetryPolicy(attempts/delay/backoff/ max_delay/total_timeout, explicitUNLIMITED = math.inf;Nonealwaysmeans "driver default"). Both the reader's WAIT-mode reconnect pacing and
the publisher's ack retries now run on RetryPolicy, with per-driver
defaults declared as class attributes.
BatchPolicy— single class, explicitdynamicflag;max_batchmeans the same in both modes, so mode-neutral callers benefit from
default changes. Driver default is now dynamic:
initial_batch=2,max_time_in_memory=5 s,max_batch=10 000, growth damped to ≤ 8× theprevious pull (a two-sample rate estimate can be wildly optimistic).
Static behaviour:
BatchPolicy(dynamic=False, max_batch=…).kwarg raises
PolicyConflict. All policies are frozen dataclasses(immutable → safe shared defaults, lesson from MsgCallbackSubscriber: class-level default Event() is shared between instances #34).
get_publisher/get_journalpublisher/get_progresspublisheraccept
**kwargs(incl.error_policy);get_readeracceptserror_policy/batch_policy.Behaviour fixes flushed out by the dynamic default
wait_for_empty()/_emptiednow means "the server has nothingmore" (
num_pending == 0), not "local read-ahead drained" — it firedprematurely whenever the backlog exceeded one pull (also pre-existing
with static batch and backlog > 100).
Driver lifecycle (open/close review)
ever created and nothing deregistered them — a service creating ad-hoc
readers (oca_monitor pattern) accumulated them for the process lifetime.
close()now deregisters from the tree (keeping the parent reference soreopen works);
open()re-registers.Messenger.close()closes drivers left open (with awarning) — a forgotten reader would leak a server-side consumer.
ensure_opendocumented: restores the closed state after the call.Deprecations (Phase 3 of the design lands later; these are the early ones)
get_singlepublisher()/get_singlereader()warn — usesingle_publish()/single_read()(one-shot) orget_publisher()/get_reader(); ofp holds 8 long-lived single-publishers, each publishre-opening the driver.
(will become an error in 3.0; paves the way for
open()gaining realsemantics like early stream validation).
Root-cause warning fixes
testcontainers.community.natsimport (conftest);getaddrinfo()instead of removed-in-aiodns-4gethostbyname().Tests
41 new tests (policy semantics, normalization conflicts, retry math,
dynamic sizing + damping, driver defaults, lifecycle deregistration,
deprecations, end-to-end policy-driven reads on live NATS).
Full suite: 347 passed, 8 skipped, 1 xfailed — the pre-existing tests
pass unchanged, which is the backward-compatibility proof.