From 97ac4cf3400e21cfbd22eba9606fb0ab9e98f9f5 Mon Sep 17 00:00:00 2001 From: majkelx Date: Fri, 7 Aug 2026 16:50:28 +0200 Subject: [PATCH 1/5] Policy objects for Messenger drivers: Delivery/Error/Retry/Batch policies, dynamic batching by default, driver lifecycle fixes (v2.3.0) --- docs/design/POLICIES.md | 9 +- pyproject.toml | 2 +- serverish/connection/connection.py | 3 +- serverish/messenger/__init__.py | 3 + serverish/messenger/messenger.py | 82 +++++- serverish/messenger/msg_journal_pub.py | 4 +- serverish/messenger/msg_progress_pub.py | 4 +- serverish/messenger/msg_publisher.py | 65 ++++- serverish/messenger/msg_reader.py | 145 ++++++++-- serverish/messenger/msg_single_pub.py | 9 +- serverish/messenger/msg_single_read.py | 7 +- serverish/messenger/policies.py | 342 ++++++++++++++++++++++++ tests/conftest.py | 2 +- tests/test_policies.py | 298 +++++++++++++++++++++ tests/test_policies_nats.py | 126 +++++++++ 15 files changed, 1056 insertions(+), 45 deletions(-) create mode 100644 serverish/messenger/policies.py create mode 100644 tests/test_policies.py create mode 100644 tests/test_policies_nats.py diff --git a/docs/design/POLICIES.md b/docs/design/POLICIES.md index 8fd163f..016640b 100644 --- a/docs/design/POLICIES.md +++ b/docs/design/POLICIES.md @@ -108,7 +108,14 @@ Slow consumer → small pulls (bounded memory & residency); fast consumer on a high-RTT link → pulls grow toward `max_batch` (fewer round trips — see the production profiling: 34 k msgs over ~290 ms RTT, pull size 100→1000 = 3.3× faster). `num_pending` (server-stamped) additionally caps the request -at what actually exists. +at what actually exists, and growth is damped to ≤ 8× the previous pull — +a two-sample rate estimate can be wildly optimistic, and the damping turns +a potential memory spike into a geometric ramp-up (2→16→128→…). + +**Driver default (decided 2026-08-07): dynamic** — +`BatchPolicy(dynamic=True, initial_batch=2, max_time_in_memory=5.0, +max_batch=10_000)`. Static behaviour is pinned explicitly with +`BatchPolicy(dynamic=False, max_batch=…)` (default static size: 100). ### 1.4 Defaults strategy diff --git a/pyproject.toml b/pyproject.toml index 7694a7e..584f831 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "serverish" -version = "2.2.1" +version = "2.3.0" description = "helpers for server alike projects" authors = ["Mikołaj Kałuszyński", "MMME team"] readme = "README.md" diff --git a/serverish/connection/connection.py b/serverish/connection/connection.py index dd6a1a3..568d9c5 100644 --- a/serverish/connection/connection.py +++ b/serverish/connection/connection.py @@ -62,7 +62,8 @@ def is_ip(cls, host) -> bool: async def _check_host(resolver, h): import aiodns try: - await resolver.gethostbyname(h, socket.AF_INET) + # getaddrinfo, not the deprecated gethostbyname (removed in aiodns 4 line) + await resolver.getaddrinfo(h, family=socket.AF_INET, type=socket.SOCK_STREAM) return None except aiodns.error.DNSError as e: try: diff --git a/serverish/messenger/__init__.py b/serverish/messenger/__init__.py index 75c7db6..82e324f 100644 --- a/serverish/messenger/__init__.py +++ b/serverish/messenger/__init__.py @@ -1,4 +1,7 @@ from .messenger import Messenger +from .policies import (UNLIMITED, BatchPolicy, DeliverAll, DeliverFromSeq, DeliverFromTime, + DeliverLast, DeliverLastPerSubject, DeliverNew, DeliveryPolicy, + ErrorPolicy, OnError, OnMissed, PolicyConflict, RetryPolicy, Until) from .msgvalidator import MsgValidator, DataValidator, MetaValidator from .msg_publisher import get_publisher from .msg_reader import get_reader diff --git a/serverish/messenger/messenger.py b/serverish/messenger/messenger.py index d626ef8..ded25b2 100644 --- a/serverish/messenger/messenger.py +++ b/serverish/messenger/messenger.py @@ -155,6 +155,17 @@ async def schedule_open(self, host: str | None = None, port: int | None = None) async def close(self): + # Safety net: close drivers that are still open. The tree owns their + # lifecycle - a forgotten reader would otherwise leak a server-side + # consumer until its inactive_threshold expires. + for child in list(self.children_by_name.values()): + if isinstance(child, MsgDriver) and child.is_open: + log.warning(f"Messenger.close(): driver {child} left open - closing it now; " + f"close drivers explicitly (async with / await close())") + try: + await child.close() + except Exception as e: + log.warning(f"Error closing {child} on Messenger.close(): {e}") if self.conn is not None: await self.connection.disconnect() self.conn = None @@ -323,17 +334,19 @@ async def purge(self, subject: str) -> None: await js.purge_stream(stream, subject=subject) @staticmethod - def get_publisher(subject: str) -> "MsgPublisher": + def get_publisher(subject: str, **kwargs) -> "MsgPublisher": """Returns a publisher for a given subject Args: subject (str): subject to publish to + kwargs: additional MsgPublisher arguments, e.g. error_policy (ErrorPolicy), + raise_on_publish_error, ack_timeout_retries Returns: MsgPublisher: message publisher """ from serverish.messenger.msg_publisher import MsgPublisher - return MsgPublisher(subject=subject, parent=Messenger()) + return MsgPublisher(subject=subject, parent=Messenger(), **kwargs) @staticmethod def get_reader(subject: str, @@ -363,7 +376,8 @@ def get_reader(subject: str, # Extract MsgReader-specific parameters reader_params = {} consumer_cfg = {} - reader_param_names = {'nowait', 'error_behavior', 'on_missed_messages'} + reader_param_names = {'nowait', 'error_behavior', 'on_missed_messages', + 'error_policy', 'batch_policy'} for key, value in kwargs.items(): if key in reader_param_names: @@ -380,7 +394,13 @@ def get_reader(subject: str, @staticmethod def get_singlepublisher(subject): - """Returns a signle-publisher for a given subject + """Returns a single-publisher for a given subject + + .. deprecated:: 2.3 + The single publisher exists only to back the `single_publish()` + convenience function - use that for one-shot publishing, or + `get_publisher()` for repeated publishing (holding long-lived + single-publishers re-opens the driver on every publish). Args: subject (str): subject to publish to @@ -389,6 +409,11 @@ def get_singlepublisher(subject): MsgSinglePublisher: a publisher for the given subject """ + import warnings + warnings.warn( + "get_singlepublisher() is deprecated: use single_publish() for one-shot " + "publishing or get_publisher() for repeated publishing", + DeprecationWarning, stacklevel=2) from serverish.messenger.msg_single_pub import MsgSinglePublisher return MsgSinglePublisher(subject=subject, parent=Messenger()) @@ -396,6 +421,18 @@ def get_singlepublisher(subject): def get_singlereader(subject, deliver_policy='last', **kwargs): + """Returns a single-reader for a given subject + + .. deprecated:: 2.3 + The single reader exists only to back the `single_read()` + convenience function - use that for one-shot reads, or + `get_reader()` for iteration. + """ + import warnings + warnings.warn( + "get_singlereader() is deprecated: use single_read() for one-shot reads " + "or get_reader() for iteration", + DeprecationWarning, stacklevel=2) from serverish.messenger.msg_single_read import MsgSingleReader return MsgSingleReader(subject=subject, parent=Messenger(), @@ -458,11 +495,12 @@ def callback(rpc: Rpc): parent=Messenger()) @staticmethod - def get_progresspublisher(subject) -> 'MsgProgressPublisher': + def get_progresspublisher(subject, **kwargs) -> 'MsgProgressPublisher': """Returns a progress tracking publisher for a given subject Args: subject (str): subject to report progress to + kwargs: additional MsgPublisher arguments (e.g. error_policy) Returns: MsgProgressPublisher: a publisher for the given subject @@ -471,7 +509,7 @@ def get_progresspublisher(subject) -> 'MsgProgressPublisher': from serverish.messenger.msg_progress_pub import MsgProgressPublisher return MsgProgressPublisher(subject=subject, - parent=Messenger()) + parent=Messenger(), **kwargs) @staticmethod def get_progressreader(subject, @@ -497,11 +535,12 @@ def get_progressreader(subject, **kwargs) @staticmethod - def get_journalpublisher(subject) -> 'MsgJournalPublisher': + def get_journalpublisher(subject, **kwargs) -> 'MsgJournalPublisher': """Returns a journal publisher for a given subject Args: subject (str): subject to publish to + kwargs: additional MsgPublisher arguments (e.g. error_policy) Returns: MsgJournalPublisher: a publisher for the given subject @@ -509,7 +548,7 @@ def get_journalpublisher(subject) -> 'MsgJournalPublisher': """ from serverish.messenger.msg_journal_pub import MsgJournalPublisher return MsgJournalPublisher(subject=subject, - parent=Messenger()) + parent=Messenger(), **kwargs) @staticmethod def get_journalreader(subject, @@ -712,16 +751,41 @@ def connection(self) -> ConnectionJetStream: return self.messenger.connection async def open(self) -> None: + # (re-)register in the parent tree - close() deregisters, and drivers + # may be reopened (single publishers, KV one-shots, @ensure_open) + if self.parent is not None and self not in getattr(self.parent, 'children_names', {}): + self.parent.ensure_parenting(self) self.is_open = True async def close(self) -> None: self.is_open = False + # Deregister from the parent collector: the tree holds strong + # references, so closed drivers would otherwise be retained for the + # process lifetime (a service creating ad-hoc readers accumulates + # them without bound). The driver keeps its own parent reference so + # it can be reopened. + parent = self.parent + if parent is not None and self in getattr(parent, 'children_names', {}): + parent.remove_child(self) + self.parent = parent # remove_child cleared it; keep for reopen @staticmethod def ensure_open(func): + """Decorator: runs the method on a temporarily opened driver. + + If the driver is already open, the method just runs. If it is + closed, the driver is opened for the duration of the call and + **closed again afterwards** - the driver's open/closed state is + restored, NOT left open. This enables one-shot usage + (e.g. `kv_get`/`kv_put`, single publishers) without ceremony. + + For repeated operations do not rely on this per-call open/close + churn - open the driver explicitly (``async with driver:`` or + ``await driver.open()``). + """ @functools.wraps(func) async def wrapper(driver: MsgDriver, *args, **kwargs): - # If not open, use context manager + # If not open, use context manager (opens, runs, closes back) if not driver.is_open: async with driver: return await func(driver, *args, **kwargs) diff --git a/serverish/messenger/msg_journal_pub.py b/serverish/messenger/msg_journal_pub.py index c46f344..50c8ae0 100644 --- a/serverish/messenger/msg_journal_pub.py +++ b/serverish/messenger/msg_journal_pub.py @@ -174,7 +174,7 @@ def health_status(self) -> dict: return status -def get_journalpublisher(subject) -> MsgJournalPublisher: +def get_journalpublisher(subject, **kwargs) -> MsgJournalPublisher: """Returns a publisher for a given subject Args: @@ -184,4 +184,4 @@ def get_journalpublisher(subject) -> MsgJournalPublisher: MsgJournalPublisher: a publisher for the given subject """ - return Messenger.get_journalpublisher(subject) + return Messenger.get_journalpublisher(subject, **kwargs) diff --git a/serverish/messenger/msg_progress_pub.py b/serverish/messenger/msg_progress_pub.py index 4143f8f..ebb1e1c 100644 --- a/serverish/messenger/msg_progress_pub.py +++ b/serverish/messenger/msg_progress_pub.py @@ -350,7 +350,7 @@ def health_status(self) -> dict: return status -def get_progresspublisher(subject) -> MsgProgressPublisher: +def get_progresspublisher(subject, **kwargs) -> MsgProgressPublisher: """Returns a progress tracking publisher for a given subject Args: @@ -360,5 +360,5 @@ def get_progresspublisher(subject) -> MsgProgressPublisher: MsgProgressPublisher: a publisher for the given subject """ - return Messenger.get_progresspublisher(subject) + return Messenger.get_progresspublisher(subject, **kwargs) diff --git a/serverish/messenger/msg_publisher.py b/serverish/messenger/msg_publisher.py index 95dacf4..8656c9c 100644 --- a/serverish/messenger/msg_publisher.py +++ b/serverish/messenger/msg_publisher.py @@ -1,6 +1,8 @@ from __future__ import annotations +import asyncio import time +import warnings import jsonschema import nats.errors @@ -10,6 +12,7 @@ from serverish.base import MessengerNotConnected, MessengerPublishAckTimeout from serverish.messenger import Messenger from serverish.messenger.messenger import MsgDriver, log +from serverish.messenger.policies import UNLIMITED, ErrorPolicy, OnError, RetryPolicy class MsgPublisher(MsgDriver): @@ -45,14 +48,43 @@ class MsgPublisher(MsgDriver): doc="Retries when JetStream ack does not arrive in time; " "safe thanks to Nats-Msg-Id deduplication") - def __init__(self, **kwargs) -> None: + #: Effective ack-retry pacing (docs/design/POLICIES.md §1.4: driver defaults + #: live at the driver). Overridable via error_policy=ErrorPolicy(retry=...). + retry_defaults = RetryPolicy(attempts=2, delay=0.0, backoff=1.0, + max_delay=0.0, total_timeout=UNLIMITED) + + def __init__(self, error_policy: ErrorPolicy | None = None, **kwargs) -> None: + # --- policy normalization (docs/design/POLICIES.md) --- + self.errors: ErrorPolicy | None = ErrorPolicy.from_kwargs( + error_policy, + raise_on_publish_error=kwargs.get('raise_on_publish_error'), + ack_timeout_retries=kwargs.get('ack_timeout_retries')) + if self.errors is not None and self.errors.on_error is not None: + if self.errors.on_error not in (OnError.RAISE, OnError.LOG): + raise ValueError("Publishers support OnError.RAISE or OnError.LOG only") + 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): + kwargs['ack_timeout_retries'] = int(self.errors.retry.attempts) + # --- end policy normalization --- # Health monitoring fields self._publish_count: int = 0 self._error_count: int = 0 self._last_publish_time: float | None = None self._last_error: Exception | None = None + self._warned_unopened: bool = False super().__init__(**kwargs) + @property + def _ack_retry(self) -> RetryPolicy: + """Resolved ack-retry pacing: error_policy.retry wins; otherwise the + legacy `ack_timeout_retries` param (kept live - it may be set after + construction) fills the attempts slot of the driver defaults.""" + if self.errors is not None and self.errors.retry is not None: + return self.errors.retry.resolved(self.retry_defaults) + return RetryPolicy(attempts=self.ack_timeout_retries).resolved(self.retry_defaults) + async def publish(self, data: dict | None = None, meta: dict | None = None, **kwargs) -> dict: """Publishes a messages to publisher subject @@ -66,6 +98,13 @@ async def publish(self, data: dict | None = None, meta: dict | None = None, **kw Raises nats errors if the message could not be published until `raise_on_publish_error` is set to True otherwise logs the error, and returns the message with the `error` tag and `status` set to the error message. """ + if not self.is_open and not self._warned_unopened: + self._warned_unopened = True + 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) msg = self.messenger.create_msg(data, meta) bdata = self.messenger.encode(msg) try: @@ -121,19 +160,27 @@ async def _publish_with_ack_retries(self, bdata: bytes, headers: dict, msg_id: s Raises: MessengerPublishAckTimeout: no ack after all retries """ - for attempt in range(self.ack_timeout_retries + 1): + retry = self._ack_retry + start = time.monotonic() + attempt = 0 + while True: try: await self.connection.js.publish(self.subject, bdata, headers=headers, **kwargs) return except nats.errors.TimeoutError as e: - if attempt < self.ack_timeout_retries: + if retry.keeps_retrying(attempt, time.monotonic() - start): + delay = retry.delay_for(attempt) log.warning(f"No JetStream ack for message {msg_id} on '{self.subject}' " - f"(attempt {attempt + 1}/{self.ack_timeout_retries + 1}), retrying - " - f"deduplicated by Nats-Msg-Id") + f"(attempt {attempt + 1}), retrying" + + (f" in {delay:.1f}s" if delay else "") + + " - deduplicated by Nats-Msg-Id") + if delay: + await asyncio.sleep(delay) + attempt += 1 continue raise MessengerPublishAckTimeout( f"No JetStream ack for message {msg_id} on '{self.subject}' " - f"after {self.ack_timeout_retries + 1} attempts. The message MAY have been " + f"after {attempt + 1} attempts. The message MAY have been " f"delivered - a missing ack often means a starved client event loop, " f"not a delivery failure") from e @@ -166,14 +213,16 @@ def health_status(self) -> dict: } -def get_publisher(subject) -> MsgPublisher: +def get_publisher(subject, **kwargs) -> MsgPublisher: """Returns a publisher for a given subject Args: subject (str): subject to publish to + kwargs: additional MsgPublisher arguments, e.g. error_policy (ErrorPolicy), + raise_on_publish_error, ack_timeout_retries Returns: Publisher: a publisher for the given subject """ - return Messenger.get_publisher(subject) + return Messenger.get_publisher(subject, **kwargs) diff --git a/serverish/messenger/msg_reader.py b/serverish/messenger/msg_reader.py index b6cf9e7..9b5f7e9 100644 --- a/serverish/messenger/msg_reader.py +++ b/serverish/messenger/msg_reader.py @@ -24,6 +24,8 @@ from serverish.base.fifoset import FifoSet from serverish.messenger import Messenger from serverish.messenger.messenger import MsgDriver +from serverish.messenger.policies import (UNLIMITED, BatchPolicy, DeliveryPolicy, ErrorPolicy, + OnError, RetryPolicy) log = logging.getLogger(__name__.rsplit('.')[-1]) @@ -45,6 +47,9 @@ class MsgReader(MsgDriver): nowait (bool): if True, read_next will return immediately if no messages are available and will finish iteration error_behavior (str): on serious error (e.g. disconnection), one of 'RAISE', 'FINISH', 'WAIT' on_missed_messages (str): on missed messages (e.g. during broken connection), one of 'SKIP', 'REPLAY' + error_policy (ErrorPolicy): typed alternative to error_behavior/on_missed_messages, adds RetryPolicy + control over WAIT-mode reconnect pacing; conflicts with the flat kwargs (docs/design/POLICIES.md) + batch_policy (BatchPolicy): pull-size control - static bound or dynamic (consumption-rate based) sizing """ @@ -74,10 +79,26 @@ class MsgReader(MsgDriver): doc="If True, read_next will return immediately if no messages are available and will " "finish iteration") + #: Effective WAIT-mode reconnect pacing (docs/design/POLICIES.md §1.4: + #: driver defaults live at the driver). Overridable per reader via + #: error_policy=ErrorPolicy(retry=RetryPolicy(...)). + retry_defaults = RetryPolicy(attempts=UNLIMITED, delay=0.2, backoff=1.5, + max_delay=15.0, total_timeout=UNLIMITED) + #: Effective batching defaults: dynamic sizing driven by the measured + #: consumption rate. Starts tiny (2), keeps a prefetched message at most + #: ~5 s in client memory, never requests more than 10k. Callers pin + #: static behaviour explicitly with BatchPolicy(dynamic=False, ...). + batch_defaults = BatchPolicy(dynamic=True, initial_batch=2, + max_time_in_memory=5.0, max_batch=10_000) + #: Pull size for static mode when the policy does not name one. + static_batch_default = 100 + def __init__(self, subject, parent = None, deliver_policy = 'all', opt_start_time = None, consumer_cfg=None, + error_policy: ErrorPolicy | None = None, + batch_policy: BatchPolicy | None = None, **kwargs) -> None: if parent is None: parent = Messenger() @@ -86,8 +107,48 @@ def __init__(self, subject, parent = None, } if consumer_cfg is not None: consumer_cfg_defaults.update(consumer_cfg) - # pull request size == read-ahead buffer bound; becomes BatchPolicy (docs/design/POLICIES.md) - self.batch = 100 + + # --- policy normalization (docs/design/POLICIES.md) --- + # `deliver_policy` is dual-typed: a string ('last', ...) is permanent + # sugar, a DeliveryPolicy object is the full form. Satellites + # (opt_start_time, opt_start_seq, nowait) combine only with the string. + self.delivery: DeliveryPolicy = DeliveryPolicy.from_value( + deliver_policy, + opt_start_time=opt_start_time, + opt_start_seq=consumer_cfg_defaults.get('opt_start_seq'), + nowait=kwargs.get('nowait')) + delivery_params = self.delivery.to_reader_params() + deliver_policy = delivery_params['deliver_policy'] + opt_start_time = delivery_params.get('opt_start_time', opt_start_time) + if 'opt_start_seq' in delivery_params: + consumer_cfg_defaults['opt_start_seq'] = delivery_params['opt_start_seq'] + if 'nowait' in delivery_params: + kwargs['nowait'] = delivery_params['nowait'] + + self.errors: ErrorPolicy | None = ErrorPolicy.from_kwargs( + error_policy, + error_behavior=kwargs.get('error_behavior'), + on_missed_messages=kwargs.get('on_missed_messages')) + if self.errors is not None: + if self.errors.on_error is not None: + if self.errors.on_error is OnError.LOG: + raise ValueError("Readers do not support OnError.LOG - use RAISE, FINISH or WAIT") + kwargs['error_behavior'] = self.errors.on_error.value + if self.errors.on_missed is not None: + kwargs['on_missed_messages'] = self.errors.on_missed.value + self._retry: RetryPolicy = ((self.errors.retry if self.errors and self.errors.retry + else RetryPolicy()).resolved(self.retry_defaults)) + + if batch_policy is not None and not isinstance(batch_policy, BatchPolicy): + raise ValueError(f"batch_policy must be a BatchPolicy, got {type(batch_policy).__name__}") + self.batching: BatchPolicy = batch_policy or BatchPolicy() + # legacy attribute: static-mode pull size / read-ahead bound + self.batch = self.batching.max_batch or self.static_batch_default + self._consume_stamps: deque[float] = deque(maxlen=32) # recent consumption timestamps + self._server_pending: int | None = None # last num_pending reported by the server + self._last_batch: int | None = None # previous dynamic pull size (growth damping) + # --- end policy normalization --- + self.messages = deque() self.pull_subscription: JetStreamContext.PullSubscription | None = None self.push_subscription: JetStreamContext.PushSubscription | None = None @@ -108,17 +169,9 @@ def __init__(self, subject, parent = None, super().__init__(subject=subject, parent=parent, deliver_policy=deliver_policy, opt_start_time=opt_start_time, consumer_cfg=consumer_cfg_defaults, **kwargs) - # Validate deliver_policy ↔ start-marker consistency up front so - # callers get a clear error at construction time rather than an - # opaque NATS 400 error buried inside the read loop. - if self.deliver_policy == 'by_start_time' and self.opt_start_time is None: - raise ValueError( - "deliver_policy='by_start_time' requires opt_start_time to be set" - ) - if self.deliver_policy == 'by_start_sequence' and self.consumer_cfg.get('opt_start_seq') is None: - raise ValueError( - "deliver_policy='by_start_sequence' requires opt_start_seq to be set in consumer_cfg" - ) + # deliver_policy ↔ start-marker consistency is validated up front by + # DeliveryPolicy.from_value above, so callers get a clear error at + # construction time rather than an opaque NATS 400 in the read loop. log.debug(f"Created {self}") @@ -168,6 +221,7 @@ class EndIterationException(_LoopException): pass log: List[str] = field(default_factory=list) start_time: datetime = field(default_factory=datetime.now) error: Exception = None + error_since: float = field(default_factory=time.monotonic) # start of current error streak last_consumer_check: float = field(default_factory=time.monotonic) consumer_check_interval: float = 10.0 # Check consumer health every 10 seconds @@ -217,7 +271,13 @@ async def pop_msg(self) -> None: # Update health monitoring stats self.reader._message_count += 1 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: + # "emptied" means the SERVER has nothing more for us + # (num_pending == 0 or unknown), not merely that the + # local read-ahead drained - a pull smaller than the + # backlog (dynamic batching, backlog > static batch) + # must not signal wait_for_empty() prematurely. self.reader._emptied.set() except Exception as e: # Error in logging, can be ignored @@ -295,8 +355,9 @@ async def read_batch(self) -> None: # For fetch_available, timeout is network latency budget (not message wait time) # Server responds immediately with no_wait=True, we just account for slow networks fetch_timeout = 2.0 # Reduced - should be quick with no_wait - log.debug(self.fmt(f"Pulling {self.reader.batch} messages with timeout {fetch_timeout}s")) - new_msgs = await self.reader.fetch_available(batch=self.reader.batch, timeout=fetch_timeout) + batch = self.reader._next_batch_size() + log.debug(self.fmt(f"Pulling {batch} messages with timeout {fetch_timeout}s")) + new_msgs = await self.reader.fetch_available(batch=batch, timeout=fetch_timeout) log.debug(self.fmt(f"Pulled {len(new_msgs)} messages")) # If no messages were available (got 404 from server or timeout), decide what to do @@ -396,6 +457,8 @@ def logput(self, msg: str) -> None: raise MessengerReaderStopped except st.ErrorException as e: # some error st.logput(f'{e.task}-err') + if st.error is None: + st.error_since = time.monotonic() # new error streak begins st.error = e.error self._last_error = e.error # Track for health monitoring # Fatal NATS API errors (4xx): the same request will never @@ -416,7 +479,13 @@ def logput(self, msg: str) -> None: log.error(st.fmt(f"finishing iteration on error: {e.error}")) raise MessengerReaderStopped case 'WAIT': - wait_time = min(0.2 + st.n/5.0, 15.0) + if not self._retry.keeps_retrying(st.n, time.monotonic() - st.error_since): + log.error(st.fmt(f"retry budget exhausted " + f"(attempts={self._retry.attempts}, " + f"total_timeout={self._retry.total_timeout}s), " + f"raising: {e.error}")) + raise e.error + wait_time = self._retry.delay_for(st.n) log.warning(st.fmt(f"read_next error, (retry in {wait_time:.1f}s): {e.error}")) await asyncio.sleep(wait_time) case _: # should not be reached @@ -539,6 +608,8 @@ def log_backstop(): needed -= 1 num_pending = self._num_pending(msg) + if num_pending is not None: + self._server_pending = num_pending # feeds dynamic batch sizing if num_pending == 0: # Server-confirmed end of data - no reason to wait for the # closing status (which may trail by a network round trip, @@ -556,6 +627,46 @@ def log_backstop(): return msgs + def _next_batch_size(self) -> int: + """Pull size for the next request. + + Static mode: the configured constant. Dynamic mode (the default): + sized so that a prefetched message waits at most `max_time_in_memory` + seconds at the measured consumption rate, damped to at most 8x the + previous pull (a two-sample rate estimate can be wildly optimistic - + the damping turns a potential memory spike into a geometric ramp-up), + clamped to `max_batch` and capped by the server-stamped num_pending + (+1 to keep probing for new data). + """ + 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() + if rate is None: + batch = self.batching.initial_batch or self.batch_defaults.initial_batch + else: + target = self.batching.max_time_in_memory or self.batch_defaults.max_time_in_memory + batch = int(rate * target) + if self._last_batch is not None: + batch = min(batch, self._last_batch * 8) + if self._server_pending is not None: + batch = min(batch, self._server_pending + 1) + batch = max(1, min(batch, max_batch)) + self._last_batch = batch + return batch + + def _consumption_rate(self) -> float | None: + """Messages/s the consumer actually takes, None until enough data""" + stamps = self._consume_stamps + if len(stamps) < 2: + return None + span = stamps[-1] - stamps[0] + if span <= 0: + return None + return (len(stamps) - 1) / span + @staticmethod def _num_pending(msg: Msg) -> int | None: """Returns server-side `num_pending` from JetStream message metadata diff --git a/serverish/messenger/msg_single_pub.py b/serverish/messenger/msg_single_pub.py index 5c4a7e7..0d21b4b 100644 --- a/serverish/messenger/msg_single_pub.py +++ b/serverish/messenger/msg_single_pub.py @@ -23,6 +23,10 @@ async def publish(self, data: dict | None = None, meta: dict | None = None, **kw def get_singlepublisher(subject) -> MsgSinglePublisher: """Returns a single-publisher for a given subject + .. deprecated:: 2.3 + Use `single_publish()` for one-shot publishing or `get_publisher()` + for repeated publishing. + Args: subject (str): subject to publish to @@ -30,7 +34,7 @@ def get_singlepublisher(subject) -> MsgSinglePublisher: MsgSinglePublisher: a publisher for the given subject """ - return Messenger.get_singlepublisher(subject) + return Messenger.get_singlepublisher(subject) # emits the DeprecationWarning async def single_publish(subject, data: dict | None = None, meta: dict | None = None, **kwargs) -> dict: @@ -43,5 +47,6 @@ async def single_publish(subject, data: dict | None = None, meta: dict | None = kwargs: additional arguments to pass to the connection """ - pub = get_singlepublisher(subject) + # constructed directly, not via the deprecated factory + pub = MsgSinglePublisher(subject=subject, parent=Messenger()) return await pub.publish(data, meta=meta, **kwargs) diff --git a/serverish/messenger/msg_single_read.py b/serverish/messenger/msg_single_read.py index 9e2e98d..185ff38 100644 --- a/serverish/messenger/msg_single_read.py +++ b/serverish/messenger/msg_single_read.py @@ -72,6 +72,9 @@ def get_singlereader(subject: str, **kwargs) -> 'MsgSingleReader': """Returns a single value reader for a given subject + .. deprecated:: 2.3 + Use `single_read()` for one-shot reads or `get_reader()` for iteration. + Args: subject (str): subject to read from deliver_policy: deliver policy, in this context 'last' is most useful @@ -114,5 +117,7 @@ async def single_read(subject: str, print("No data published yet") """ - reader = get_singlereader(subject, deliver_policy, **kwargs) + # constructed directly, not via the deprecated factory + reader = MsgSingleReader(subject=subject, parent=Messenger(), + deliver_policy=deliver_policy, **kwargs) return await reader.read(wait=wait) \ No newline at end of file diff --git a/serverish/messenger/policies.py b/serverish/messenger/policies.py new file mode 100644 index 0000000..e065493 --- /dev/null +++ b/serverish/messenger/policies.py @@ -0,0 +1,342 @@ +"""Policy objects for Messenger drivers. + +Policies group related driver configuration into typed, immutable value +objects (design: docs/design/POLICIES.md): + +- `DeliveryPolicy` variants — where a reader starts and when it stops, +- `ErrorPolicy` (+ nested `RetryPolicy`) — what to do when things break, +- `BatchPolicy` — how much a reader prefetches. + +All policies are frozen dataclasses: immutable, hashable, safe to share as +class-level defaults. Two uniform rules apply everywhere: + +- ``None`` always and only means "driver default" — each driver documents + its effective defaults; +- "no limit" is always the explicit `UNLIMITED`, never a hidden meaning + of None. + +Backward compatibility: the flat kwargs used across the ecosystem +(``deliver_policy='last'``, ``opt_start_time=...``, ``nowait=True``, +``error_behavior=...``, ``raise_on_publish_error=...``) keep working — +they are normalized into policies in one place (`resolve_delivery`, +`resolve_error_policy`). Passing a policy object together with one of its +satellite kwargs is an error. +""" +from __future__ import annotations + +import math +from dataclasses import dataclass, field, replace +from datetime import datetime +from enum import Enum + +UNLIMITED = math.inf +"""Explicit "no limit" marker for numeric policy fields. + +Deliberately NOT named FOREVER: `Until.FOREVER` is a different concept +("keep following live data"). +""" + + +class Until(Enum): + """When a reader finishes iterating.""" + FOREVER = 'forever' # live subscription, never finishes (legacy: nowait=False) + END_OF_DATA = 'end_of_data' # stop when the server confirms drained (legacy: nowait=True) + + +class OnError(Enum): + """Reaction to a serious driver error (e.g. disconnection, no ack).""" + RAISE = 'RAISE' # re-raise to the caller + FINISH = 'FINISH' # silently finish iteration (readers) + WAIT = 'WAIT' # wait for recovery, keep retrying (readers) + LOG = 'LOG' # log and continue, tag the message (publishers) + + +class OnMissed(Enum): + """Reaction to messages missed during a broken connection (readers).""" + SKIP = 'SKIP' + REPLAY = 'REPLAY' + + +class PolicyConflict(ValueError): + """Raised when a policy object is combined with one of its satellite kwargs.""" + pass + + +# -------------------------------------------------------------------------- +# DeliveryPolicy — variants as types: invalid combinations are unrepresentable +# -------------------------------------------------------------------------- + +@dataclass(frozen=True) +class DeliveryPolicy: + """Base of delivery policy variants - where to start, when to stop. + + Use one of the concrete variants: `DeliverAll`, `DeliverLast`, + `DeliverNew`, `DeliverLastPerSubject`, `DeliverFromTime(start)`, + `DeliverFromSeq(seq)`. + """ + until: Until | None = field(default=None, kw_only=True) + + _policy_str: str = field(default='', init=False, repr=False, compare=False) + + @classmethod + def from_value(cls, value: 'str | DeliveryPolicy', + opt_start_time: datetime | None = None, + opt_start_seq: int | None = None, + nowait: bool | None = None) -> 'DeliveryPolicy': + """Normalizes legacy flat kwargs or a policy object into a policy object. + + The string form is permanent API sugar (``'last'`` == `DeliverLast()`), + not deprecated. Satellite kwargs (`opt_start_time`, `opt_start_seq`, + `nowait`) are accepted only with the string form. + + Raises: + PolicyConflict: a policy object was combined with a satellite kwarg + ValueError: unknown policy string or missing start marker + """ + if isinstance(value, DeliveryPolicy): + for name, sat in (('opt_start_time', opt_start_time), + ('opt_start_seq', opt_start_seq), + ('nowait', nowait)): + if sat is not None: + raise PolicyConflict( + f"Both a DeliveryPolicy object and legacy '{name}' given - " + f"express everything in the policy object") + return value + until = None if nowait is None else (Until.END_OF_DATA if nowait else Until.FOREVER) + match value: + case 'all': + policy = DeliverAll() + case 'last': + policy = DeliverLast() + case 'new': + policy = DeliverNew() + case 'last_per_subject': + policy = DeliverLastPerSubject() + case 'by_start_time': + if opt_start_time is None: + raise ValueError("deliver_policy='by_start_time' requires opt_start_time to be set") + policy = DeliverFromTime(opt_start_time) + case 'by_start_sequence': + if opt_start_seq is None: + raise ValueError("deliver_policy='by_start_sequence' requires opt_start_seq " + "to be set in consumer_cfg") + policy = DeliverFromSeq(opt_start_seq) + case _: + raise ValueError(f"Unknown deliver policy {value!r}") + if until is not None: + policy = replace(policy, until=until) + return policy + + def to_reader_params(self) -> dict: + """Translates the policy into MsgReader's canonical parameters. + + This (together with `MsgReader._create_consumer_cfg`) is the single + point where serverish delivery semantics map onto the NATS API. + """ + params: dict = {'deliver_policy': self._policy_str} + if self.until is not None: + params['nowait'] = self.until is Until.END_OF_DATA + return params + + +@dataclass(frozen=True) +class DeliverAll(DeliveryPolicy): + """Deliver all messages retained in the stream.""" + _policy_str: str = field(default='all', init=False, repr=False, compare=False) + + +@dataclass(frozen=True) +class DeliverLast(DeliveryPolicy): + """Deliver the last message of the subject, then live data.""" + _policy_str: str = field(default='last', init=False, repr=False, compare=False) + + +@dataclass(frozen=True) +class DeliverNew(DeliveryPolicy): + """Deliver only messages published after subscribing.""" + _policy_str: str = field(default='new', init=False, repr=False, compare=False) + + +@dataclass(frozen=True) +class DeliverLastPerSubject(DeliveryPolicy): + """Deliver the last message of each subject matching a wildcard (snapshot).""" + _policy_str: str = field(default='last_per_subject', init=False, repr=False, compare=False) + + +@dataclass(frozen=True) +class DeliverFromTime(DeliveryPolicy): + """Deliver messages published at or after `start` (tz-aware datetime, + or - legacy - an ISO string passed through to the server).""" + start: datetime | str = None # type: ignore[assignment] # required, validated below + _policy_str: str = field(default='by_start_time', init=False, repr=False, compare=False) + + def __post_init__(self): + if not isinstance(self.start, (datetime, str)): + raise ValueError("DeliverFromTime requires a datetime (or ISO string) 'start'") + + def to_reader_params(self) -> dict: + params = super().to_reader_params() + params['opt_start_time'] = self.start + return params + + +@dataclass(frozen=True) +class DeliverFromSeq(DeliveryPolicy): + """Deliver messages starting from stream sequence `seq`.""" + seq: int = None # type: ignore[assignment] # required, validated below + _policy_str: str = field(default='by_start_sequence', init=False, repr=False, compare=False) + + def __post_init__(self): + if not isinstance(self.seq, int) or isinstance(self.seq, bool) or self.seq < 1: + raise ValueError("DeliverFromSeq requires a positive int 'seq'") + + def to_reader_params(self) -> dict: + params = super().to_reader_params() + params['opt_start_seq'] = self.seq + return params + + +# -------------------------------------------------------------------------- +# ErrorPolicy / RetryPolicy +# -------------------------------------------------------------------------- + +@dataclass(frozen=True) +class RetryPolicy: + """How to retry a failing operation. + + All fields default to None = "driver default". `UNLIMITED` is the + explicit no-limit marker for `attempts` and `total_timeout`. + """ + attempts: int | float | None = None # 0 = no retry; UNLIMITED = keep retrying + delay: float | None = None # initial inter-attempt delay [s] + backoff: float | None = None # delay multiplier per attempt (>= 1) + max_delay: float | None = None # cap for the growing delay [s] + total_timeout: float | None = None # overall time budget [s]; UNLIMITED = none + + def __post_init__(self): + if self.attempts is not None and self.attempts != UNLIMITED and ( + not isinstance(self.attempts, int) or self.attempts < 0): + raise ValueError("RetryPolicy.attempts must be a non-negative int or UNLIMITED") + if self.backoff is not None and self.backoff < 1: + raise ValueError("RetryPolicy.backoff must be >= 1") + for name in ('delay', 'max_delay', 'total_timeout'): + v = getattr(self, name) + if v is not None and v != UNLIMITED and v < 0: + raise ValueError(f"RetryPolicy.{name} must be non-negative") + + def resolved(self, defaults: 'RetryPolicy') -> 'RetryPolicy': + """Returns a copy with None fields filled from `defaults` (driver defaults).""" + return RetryPolicy(*(own if own is not None else default + for own, default in zip( + (self.attempts, self.delay, self.backoff, + self.max_delay, self.total_timeout), + (defaults.attempts, defaults.delay, defaults.backoff, + defaults.max_delay, defaults.total_timeout)))) + + def delay_for(self, attempt: int) -> float: + """Delay before retry number `attempt` (0-based), on a fully resolved policy.""" + 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 + + def keeps_retrying(self, attempt: int, elapsed: float) -> bool: + """Whether retry number `attempt` (0-based) is allowed, on a fully resolved policy.""" + attempts = self.attempts if self.attempts is not None else 0 + if attempts != UNLIMITED and attempt >= attempts: + return False + total = self.total_timeout + if total is not None and total != UNLIMITED and elapsed >= total: + return False + return True + + +@dataclass(frozen=True) +class ErrorPolicy: + """What to do when things break - common to readers and publishers. + + Readers use `on_error` RAISE/FINISH/WAIT and `on_missed`; publishers use + `on_error` RAISE/LOG. `retry` paces WAIT-mode reconnects (readers) and + ack-timeout retries (publishers). None = driver default. + """ + on_error: OnError | None = None + on_missed: OnMissed | None = None + retry: RetryPolicy | None = None + + @classmethod + def from_kwargs(cls, error_policy: 'ErrorPolicy | None', + error_behavior: str | None = None, + on_missed_messages: str | None = None, + raise_on_publish_error: bool | None = None, + ack_timeout_retries: int | None = None) -> 'ErrorPolicy | None': + """Normalizes legacy flat kwargs or a policy object into a policy object. + + Returns None when nothing at all was specified (pure driver defaults). + + Raises: + PolicyConflict: policy object combined with a satellite kwarg + """ + satellites = {'error_behavior': error_behavior, + 'on_missed_messages': on_missed_messages, + 'raise_on_publish_error': raise_on_publish_error, + 'ack_timeout_retries': ack_timeout_retries} + given = {k: v for k, v in satellites.items() if v is not None} + if error_policy is not None: + if not isinstance(error_policy, ErrorPolicy): + raise ValueError(f"error_policy must be an ErrorPolicy, got {type(error_policy).__name__}") + if given: + raise PolicyConflict( + f"Both error_policy and legacy {sorted(given)} given - " + f"express everything in the policy object") + return error_policy + if not given: + return None + on_error = None + if error_behavior is not None: + on_error = OnError(error_behavior) + elif raise_on_publish_error is not None: + on_error = OnError.RAISE if raise_on_publish_error else OnError.LOG + on_missed = OnMissed(on_missed_messages) if on_missed_messages is not None else None + retry = RetryPolicy(attempts=ack_timeout_retries) if ack_timeout_retries is not None else None + return cls(on_error=on_error, on_missed=on_missed, retry=retry) + + +# -------------------------------------------------------------------------- +# BatchPolicy +# -------------------------------------------------------------------------- + +@dataclass(frozen=True) +class BatchPolicy: + """How much a reader prefetches per pull - and how that adapts. + + `max_batch` means the same thing in both modes (static: the pull size; + dynamic: its upper bound), so callers expressing only the bound stay + neutral on mode and benefit when the driver default flips to dynamic. + + Dynamic mode (`dynamic=True`): pull size follows the consumer's measured + consumption rate so a prefetched message waits at most + `max_time_in_memory` seconds client-side, clamped to `max_batch` and + capped by the server-stamped num_pending. + + None = driver default (today: static, max_batch=100). + """ + max_batch: int | None = None + dynamic: bool | None = None + initial_batch: int | None = None # dynamic only: size of the first pull + max_time_in_memory: float | None = None # dynamic only: target residency [s] + + 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") + if self.max_time_in_memory is not None and self.max_time_in_memory <= 0: + raise ValueError("BatchPolicy.max_time_in_memory must be positive") + if self.dynamic is False and (self.initial_batch is not None + or self.max_time_in_memory is not None): + raise ValueError("BatchPolicy: initial_batch/max_time_in_memory require dynamic mode, " + "but dynamic=False was given") + if (self.initial_batch is not None and self.max_batch is not None + and self.initial_batch > self.max_batch): + raise ValueError("BatchPolicy.initial_batch must not exceed max_batch") diff --git a/tests/conftest.py b/tests/conftest.py index 9b5df0c..6bb2e95 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -14,7 +14,7 @@ import pytest import pytest_asyncio -from testcontainers.nats import NatsContainer +from testcontainers.community.nats import NatsContainer from serverish.messenger import Messenger diff --git a/tests/test_policies.py b/tests/test_policies.py new file mode 100644 index 0000000..5a0f761 --- /dev/null +++ b/tests/test_policies.py @@ -0,0 +1,298 @@ +"""Unit tests for policy objects (docs/design/POLICIES.md) - no NATS needed.""" +import dataclasses +import datetime + +import pytest + +from serverish.messenger import (UNLIMITED, BatchPolicy, DeliverAll, DeliverFromSeq, + DeliverFromTime, DeliverLast, DeliverLastPerSubject, + DeliverNew, DeliveryPolicy, ErrorPolicy, OnError, + OnMissed, PolicyConflict, RetryPolicy, Until) + +UTC = datetime.timezone.utc + + +# --- DeliveryPolicy ------------------------------------------------------- + +def test_from_value_string_sugar(): + assert DeliveryPolicy.from_value('all') == DeliverAll() + assert DeliveryPolicy.from_value('last') == DeliverLast() + assert DeliveryPolicy.from_value('new') == DeliverNew() + assert DeliveryPolicy.from_value('last_per_subject') == DeliverLastPerSubject() + + +def test_from_value_with_satellites(): + t = datetime.datetime.now(UTC) + p = DeliveryPolicy.from_value('by_start_time', opt_start_time=t, nowait=True) + assert p == DeliverFromTime(t, until=Until.END_OF_DATA) + p = DeliveryPolicy.from_value('by_start_sequence', opt_start_seq=42) + assert p == DeliverFromSeq(42) + + +def test_from_value_passes_objects_through(): + p = DeliverLast(until=Until.END_OF_DATA) + assert DeliveryPolicy.from_value(p) is p + + +def test_from_value_object_plus_satellite_conflicts(): + with pytest.raises(PolicyConflict): + DeliveryPolicy.from_value(DeliverLast(), nowait=True) + with pytest.raises(PolicyConflict): + DeliveryPolicy.from_value(DeliverAll(), opt_start_time=datetime.datetime.now(UTC)) + + +def test_from_value_missing_start_markers(): + with pytest.raises(ValueError, match="requires opt_start_time"): + DeliveryPolicy.from_value('by_start_time') + with pytest.raises(ValueError, match="requires opt_start_seq"): + DeliveryPolicy.from_value('by_start_sequence') + with pytest.raises(ValueError, match="Unknown deliver policy"): + DeliveryPolicy.from_value('sometimes') + + +def test_variant_validation(): + with pytest.raises(ValueError): + DeliverFromTime(None) + with pytest.raises(ValueError): + DeliverFromSeq(0) + with pytest.raises(ValueError): + DeliverFromSeq(True) # bool is not a sequence number + + +def test_to_reader_params(): + t = datetime.datetime.now(UTC) + assert DeliverLast().to_reader_params() == {'deliver_policy': 'last'} + assert DeliverFromTime(t, until=Until.END_OF_DATA).to_reader_params() == { + 'deliver_policy': 'by_start_time', 'opt_start_time': t, 'nowait': True} + assert DeliverFromSeq(7).to_reader_params() == { + 'deliver_policy': 'by_start_sequence', 'opt_start_seq': 7} + + +def test_policies_are_frozen(): + with pytest.raises(dataclasses.FrozenInstanceError): + DeliverLast().until = Until.FOREVER + with pytest.raises(dataclasses.FrozenInstanceError): + RetryPolicy().attempts = 5 + with pytest.raises(dataclasses.FrozenInstanceError): + BatchPolicy().max_batch = 10 + + +# --- RetryPolicy ---------------------------------------------------------- + +def test_retry_validation(): + with pytest.raises(ValueError): + RetryPolicy(attempts=-1) + with pytest.raises(ValueError): + RetryPolicy(attempts=1.5) + with pytest.raises(ValueError): + RetryPolicy(backoff=0.5) + with pytest.raises(ValueError): + RetryPolicy(delay=-1) + RetryPolicy(attempts=UNLIMITED, total_timeout=UNLIMITED) # explicit no-limit is fine + + +def test_retry_resolved_fills_only_none(): + defaults = RetryPolicy(attempts=2, delay=1.0, backoff=2.0, max_delay=8.0, total_timeout=UNLIMITED) + r = RetryPolicy(attempts=5).resolved(defaults) + assert r == RetryPolicy(attempts=5, delay=1.0, backoff=2.0, max_delay=8.0, total_timeout=UNLIMITED) + + +def test_retry_delay_for_backoff_and_cap(): + r = RetryPolicy(attempts=UNLIMITED, delay=1.0, backoff=2.0, max_delay=5.0, total_timeout=UNLIMITED) + assert [r.delay_for(n) for n in range(4)] == [1.0, 2.0, 4.0, 5.0] + + +def test_retry_keeps_retrying(): + r = RetryPolicy(attempts=2, delay=0, backoff=1, max_delay=0, total_timeout=10.0) + assert r.keeps_retrying(0, elapsed=0) + assert r.keeps_retrying(1, elapsed=9.9) + assert not r.keeps_retrying(2, elapsed=0) # attempts exhausted + assert not r.keeps_retrying(0, elapsed=11.0) # time budget exhausted + unlimited = RetryPolicy(attempts=UNLIMITED, total_timeout=UNLIMITED) + assert unlimited.keeps_retrying(10 ** 6, elapsed=10.0 ** 6) + + +# --- ErrorPolicy ---------------------------------------------------------- + +def test_error_policy_from_legacy_kwargs(): + p = ErrorPolicy.from_kwargs(None, error_behavior='FINISH', on_missed_messages='REPLAY') + assert p == ErrorPolicy(on_error=OnError.FINISH, on_missed=OnMissed.REPLAY) + p = ErrorPolicy.from_kwargs(None, raise_on_publish_error=False, ack_timeout_retries=5) + assert p.on_error is OnError.LOG + assert p.retry.attempts == 5 + assert ErrorPolicy.from_kwargs(None) is None # nothing given -> pure driver defaults + + +def test_error_policy_conflicts(): + with pytest.raises(PolicyConflict): + ErrorPolicy.from_kwargs(ErrorPolicy(on_error=OnError.RAISE), error_behavior='WAIT') + with pytest.raises(PolicyConflict): + ErrorPolicy.from_kwargs(ErrorPolicy(), ack_timeout_retries=1) + + +# --- BatchPolicy ---------------------------------------------------------- + +def test_batch_validation(): + with pytest.raises(ValueError): + BatchPolicy(max_batch=0) + with pytest.raises(ValueError): + BatchPolicy(dynamic=False, max_time_in_memory=1.0) + with pytest.raises(ValueError): + BatchPolicy(dynamic=False, initial_batch=10) + with pytest.raises(ValueError): + BatchPolicy(dynamic=True, initial_batch=200, max_batch=100) + BatchPolicy(max_batch=500) # mode-neutral: just the bound + BatchPolicy(dynamic=True, max_batch=1000, max_time_in_memory=0.5) + + +# --- driver integration (construction only, no NATS) ----------------------- + +def test_reader_accepts_delivery_object(): + from serverish.messenger import get_reader + t = datetime.datetime.now(UTC) + reader = get_reader('test.policies.unit', deliver_policy=DeliverFromTime(t, until=Until.END_OF_DATA)) + assert reader.deliver_policy == 'by_start_time' + assert reader.opt_start_time == t + assert reader.nowait is True + assert reader.delivery == DeliverFromTime(t, until=Until.END_OF_DATA) + + +def test_reader_legacy_kwargs_still_normalize(): + from serverish.messenger import get_reader + reader = get_reader('test.policies.unit', deliver_policy='last', nowait=True) + assert reader.delivery == DeliverLast(until=Until.END_OF_DATA) + assert reader.nowait is True + + +def test_reader_policy_satellite_conflict(): + from serverish.messenger import get_reader + with pytest.raises(PolicyConflict): + get_reader('test.policies.unit', deliver_policy=DeliverLast(), nowait=True) + with pytest.raises(PolicyConflict): + get_reader('test.policies.unit', deliver_policy=DeliverAll(), + error_policy=ErrorPolicy(on_error=OnError.WAIT), error_behavior='RAISE') + + +def test_reader_rejects_log_on_error(): + from serverish.messenger import get_reader + with pytest.raises(ValueError, match="OnError.LOG"): + get_reader('test.policies.unit', error_policy=ErrorPolicy(on_error=OnError.LOG)) + + +def test_reader_retry_resolution(): + from serverish.messenger import get_reader + reader = get_reader('test.policies.unit', + error_policy=ErrorPolicy(retry=RetryPolicy(attempts=3, delay=0.5))) + assert reader._retry.attempts == 3 + assert reader._retry.delay == 0.5 + assert reader._retry.max_delay == 15.0 # filled from driver defaults + + +def test_reader_batch_defaults_are_dynamic(): + """Driver default: dynamic sizing, initial 2, max 10k (docs/design/POLICIES.md).""" + from serverish.messenger import get_reader + reader = get_reader('test.policies.unit') + assert reader._next_batch_size() == 2 # initial pull is tiny + # a mode-neutral policy (just the bound) inherits the dynamic default + bounded = get_reader('test.policies.unit', batch_policy=BatchPolicy(max_batch=500)) + assert bounded._next_batch_size() == 2 + + +def test_reader_static_batch_from_policy(): + from serverish.messenger import get_reader + reader = get_reader('test.policies.unit', batch_policy=BatchPolicy(dynamic=False, max_batch=7)) + assert reader._next_batch_size() == 7 + plain_static = get_reader('test.policies.unit', batch_policy=BatchPolicy(dynamic=False)) + assert plain_static._next_batch_size() == 100 # static default size + + +def test_reader_dynamic_batch_sizing(): + from serverish.messenger import get_reader + reader = get_reader('test.policies.unit', + batch_policy=BatchPolicy(dynamic=True, max_batch=1000, + initial_batch=50, max_time_in_memory=2.0)) + # no consumption data yet -> initial batch + assert reader._next_batch_size() == 50 + # consumer measured at ~100 msg/s -> 100 * 2.0s = 200 (within 8x damping of 50) + reader._consume_stamps.extend(i * 0.01 for i in range(11)) + assert reader._next_batch_size() == 200 + # server says only 20 pending -> capped at 21 (probe for new data) + reader._server_pending = 20 + assert reader._next_batch_size() == 21 + # max_batch clamp wins over huge rate (damping released for the check) + reader._server_pending = None + reader._last_batch = None + reader._consume_stamps.clear() + reader._consume_stamps.extend(i * 0.0001 for i in range(11)) + assert reader._next_batch_size() == 1000 + + +def test_reader_dynamic_growth_damping(): + """A burst-inflated rate estimate must ramp geometrically, not spike.""" + from serverish.messenger import get_reader + reader = get_reader('test.policies.unit') # defaults: initial 2, max 10k + assert reader._next_batch_size() == 2 + # absurdly fast consumption measured -> still at most 8x per step + reader._consume_stamps.extend(i * 0.0001 for i in range(11)) + assert reader._next_batch_size() == 16 + assert reader._next_batch_size() == 128 + assert reader._next_batch_size() == 1024 + + +def test_driver_error_defaults(): + """Per-driver defaults documented at the driver (POLICIES.md §1.4).""" + from serverish.messenger import get_publisher, get_journalpublisher + from serverish.messenger.msg_reader import MsgReader + from serverish.messenger.msg_publisher import MsgPublisher + assert get_publisher('test.policies.unit').raise_on_publish_error is True + assert get_journalpublisher('test.policies.unit').raise_on_publish_error is False + assert MsgPublisher.retry_defaults.attempts == 2 + assert MsgReader.retry_defaults.attempts == UNLIMITED + assert MsgReader.retry_defaults.max_delay == 15.0 + + +async def test_driver_close_deregisters_from_tree(): + """Closed drivers must leave the Messenger tree (no unbounded accumulation), + and reopening must re-register them.""" + from serverish.messenger import Messenger, get_publisher + m = Messenger() + pub = get_publisher('test.policies.unit.lifecycle') + assert pub in m.children_names + await pub.close() + assert pub not in m.children_names + assert pub.parent is m # kept for reopen + await pub.open() + assert pub in m.children_names + await pub.close() + assert pub not in m.children_names + + +def test_publisher_error_policy(): + from serverish.messenger import get_publisher + pub = get_publisher('test.policies.unit', + error_policy=ErrorPolicy(on_error=OnError.LOG, + retry=RetryPolicy(attempts=0))) + assert pub.raise_on_publish_error is False + assert pub._ack_retry.attempts == 0 + with pytest.raises(ValueError, match="RAISE or"): + get_publisher('test.policies.unit', error_policy=ErrorPolicy(on_error=OnError.WAIT)) + with pytest.raises(PolicyConflict): + get_publisher('test.policies.unit', error_policy=ErrorPolicy(on_error=OnError.RAISE), + raise_on_publish_error=False) + + +def test_publisher_legacy_ack_retries_stay_live(): + from serverish.messenger import get_publisher + pub = get_publisher('test.policies.unit') + pub.ack_timeout_retries = 7 # legacy post-construction mutation + assert pub._ack_retry.attempts == 7 + + +# --- deprecations ---------------------------------------------------------- + +def test_single_factories_deprecated(): + from serverish.messenger import get_singlepublisher, get_singlereader + with pytest.warns(DeprecationWarning, match="single_publish"): + get_singlepublisher('test.policies.unit') + with pytest.warns(DeprecationWarning, match="single_read"): + get_singlereader('test.policies.unit') diff --git a/tests/test_policies_nats.py b/tests/test_policies_nats.py new file mode 100644 index 0000000..e84acfb --- /dev/null +++ b/tests/test_policies_nats.py @@ -0,0 +1,126 @@ +"""Integration tests for policies against a live NATS (see test_policies.py for units).""" +import asyncio +import datetime +import warnings + +import pytest + +from serverish.messenger import (BatchPolicy, DeliverFromTime, DeliverLastPerSubject, + ErrorPolicy, OnError, RetryPolicy, Until, + get_publisher, get_reader) +from serverish.messenger.msg_single_pub import single_publish +from serverish.messenger.msg_single_read import single_read + +UTC = datetime.timezone.utc + + +@pytest.mark.nats +async def test_reader_with_delivery_object(messenger, unique_subject): + """End-to-end read driven entirely by a DeliveryPolicy object.""" + pub = get_publisher(unique_subject) + await pub.open() + await messenger.purge(unique_subject) + for i in range(3): + await pub.publish(data={'i': i}) + await asyncio.sleep(0.1) + cutoff = datetime.datetime.now(UTC) + for i in range(3, 6): + await pub.publish(data={'i': i}) + await pub.close() + + reader = get_reader(unique_subject, + deliver_policy=DeliverFromTime(cutoff, until=Until.END_OF_DATA)) + received = [data['i'] async for data, meta in reader] + await reader.close() + assert received == [3, 4, 5] + + +@pytest.mark.nats +async def test_snapshot_with_policy_and_dynamic_batch(messenger, unique_subject): + """tcsctl-style snapshot: last_per_subject + dynamic batching, end to end.""" + for k in range(4): + pub = get_publisher(f"{unique_subject}.k{k}") + await pub.open() + for revision in range(2): + await pub.publish(data={'k': k, 'rev': revision}) + await pub.close() + await asyncio.sleep(0.1) + + reader = get_reader(f"{unique_subject}.>", + deliver_policy=DeliverLastPerSubject(until=Until.END_OF_DATA), + batch_policy=BatchPolicy(dynamic=True, max_batch=500)) + received = [data async for data, meta in reader] + await reader.close() + assert len(received) == 4 + assert all(data['rev'] == 1 for data in received) + + +@pytest.mark.nats +async def test_publisher_retry_policy_paces_ack_retries(messenger, unique_subject): + """ErrorPolicy.retry drives the ack-retry loop (attempts honored).""" + import nats.errors + pub = get_publisher(unique_subject, + error_policy=ErrorPolicy(retry=RetryPolicy(attempts=2, delay=0.0))) + await pub.open() + js = pub.connection.js + original_publish = js.publish + calls = [] + + async def flaky(*args, **kwargs): + calls.append(1) + if len(calls) <= 2: + raise nats.errors.TimeoutError + return await original_publish(*args, **kwargs) + + js.publish = flaky + try: + await pub.publish(data={'v': 1}) + finally: + js.publish = original_publish + await pub.close() + assert len(calls) == 3 # 1 original + 2 retries from the policy + + +@pytest.mark.nats +async def test_publish_on_unopened_publisher_warns_once(messenger, unique_subject): + pub = get_publisher(unique_subject) + with pytest.warns(DeprecationWarning, match="never opened"): + await pub.publish(data={'v': 1}) + with warnings.catch_warnings(): + warnings.simplefilter("error") # second publish must NOT warn again + await pub.publish(data={'v': 2}) + await pub.close() + + +@pytest.mark.nats +async def test_opened_publisher_does_not_warn(messenger, unique_subject): + async with get_publisher(unique_subject) as pub: + with warnings.catch_warnings(): + warnings.simplefilter("error", DeprecationWarning) + await pub.publish(data={'v': 1}) + + +@pytest.mark.nats +async def test_messenger_close_closes_open_drivers(messenger, nats_server, unique_subject): + """Messenger.close() is the lifecycle safety net: drivers left open are + closed (a forgotten reader would leak a server-side consumer).""" + from serverish.messenger import Messenger + reader = get_reader(unique_subject, deliver_policy='last') + await reader.open() + try: + await Messenger().close() + assert reader.is_open is False + finally: + await Messenger().close() + await Messenger().open(host=nats_server['host'], port=nats_server['port']) + + +@pytest.mark.nats +async def test_single_helpers_do_not_emit_deprecation(messenger, unique_subject): + """single_publish/single_read are the blessed API - internal use of the + single drivers must not trigger the factory DeprecationWarnings.""" + with warnings.catch_warnings(): + warnings.simplefilter("error", DeprecationWarning) + await single_publish(unique_subject, data={'v': 7}) + data, meta = await single_read(unique_subject) + assert data == {'v': 7} From fbcda728d6d1af6c849a9120799078182a0caf96 Mon Sep 17 00:00:00 2001 From: majkelx Date: Fri, 7 Aug 2026 17:37:15 +0200 Subject: [PATCH 2/5] Doc messenger.md --- README.md | 14 +- doc/messenger.md | 270 ++++++++++++++++++++++++++++++ docs/design/POLICIES.md | 185 -------------------- serverish/messenger/messenger.py | 3 +- serverish/messenger/msg_reader.py | 4 +- 5 files changed, 286 insertions(+), 190 deletions(-) create mode 100644 doc/messenger.md delete mode 100644 docs/design/POLICIES.md diff --git a/README.md b/README.md index 61bb955..1afdbc8 100644 --- a/README.md +++ b/README.md @@ -5,10 +5,10 @@ * Component management * Singletons * Connections management and diagnostics -* NATS based Messenger +* NATS based Messenger — publishers, readers, KV buckets, RPC ([user guide with examples](doc/messenger.md)) * Live documents - auto-updating configuration from NATS -See [`doc` directory](doc/) for more documentation. +See the [Messenger user guide](doc/messenger.md) and the [`doc` directory](doc/) for more documentation. ## Optional (extras) dependencies The following extras are available: @@ -28,6 +28,16 @@ this will install `nats-py` package. - Custom JetStream API replacing nats-py private internals access ## Changes +* 2.3 Policy objects (`DeliveryPolicy`, `ErrorPolicy`/`RetryPolicy`, `BatchPolicy`) for typed driver + configuration with backward-compatible string/kwarg forms; dynamic pull batching by default; + driver lifecycle fixes (closed drivers leave the Messenger tree); deprecates + `get_singlepublisher`/`get_singlereader` and publishing on never-opened publishers. + See the [Messenger user guide](doc/messenger.md). +* 2.2 Deterministic end-of-data for `nowait` readers (server-stamped `num_pending`, no fixed waits); + publish ack-timeout handling with `Nats-Msg-Id` dedup and transparent retries; common + `ServerishError` exception base. **Use 2.2.1+** — 2.2.0 is yanked (silent message loss, #38). +* 2.1 JetStream KV bucket support: `MsgKvStore`/`MsgKvReader`/`MsgKvSubscriber` with full + envelope semantics and one-shot `kv_get`/`kv_put`. * 1.7 Subscription reliability: reader auto-recovers from NATS disconnects and consumer expiry, RPC responder auto-resubscribes, `health_status` property on all drivers (reader, publisher, RPC responder, connection) with reconnect counts, error tracking, slow consumer detection. diff --git a/doc/messenger.md b/doc/messenger.md new file mode 100644 index 0000000..17a3829 --- /dev/null +++ b/doc/messenger.md @@ -0,0 +1,270 @@ +# Messenger — user guide + +NATS/JetStream messaging for serverish services: publishers, readers, +KV buckets, RPC. All examples assume `serverish[messenger]` installed and +a JetStream stream covering your subjects (streams are configured on the +server, not created by the library). + +## Connecting + +```python +from serverish.messenger import Messenger + +msg = Messenger(name='my-service') # name becomes meta.sender +async with msg.context(host='nats.example', port=4222): + ... # publish / read here +``` + +Long-running services usually call `await msg.open(...)` at startup and +`await msg.close()` at shutdown instead of the context manager. +`Messenger.close()` also closes any drivers left open (and warns about them). + +## Publishing + +### Repeated publishing + +The usual pattern: the publisher is a member of your service/component, +opened at startup and closed at shutdown. + +```python +from serverish.messenger import get_publisher + +class MyService: + def __init__(self): + self.status_pub = get_publisher('svc.status.my-service') + + async def open(self): + await self.status_pub.open() + + async def on_job_update(self, jobs: int): + await self.status_pub.publish(data={'state': 'running', 'jobs': jobs}) + + async def close(self): + await self.status_pub.close() +``` + +For a locally scoped publisher there is a context-manager shortcut: + +```python +async with get_publisher('svc.status.my-service') as pub: + await pub.publish(data={'state': 'done'}) +``` + +Either way, open publishers explicitly — publishing on a never-opened +publisher works but emits a deprecation warning and will become an error in 3.0. + +### One-shot + +```python +from serverish.messenger import single_publish + +await single_publish('svc.status.my-service', data={'state': 'done'}) +``` + +### Errors and retries + +By default a failed publish raises. Missing JetStream acks are retried +transparently (safe: every publish carries a globally unique `Nats-Msg-Id`, +so the server deduplicates); when retries are exhausted you get +`MessengerPublishAckTimeout` — note the message *may* have been delivered. + +```python +from serverish.messenger import ErrorPolicy, OnError, RetryPolicy, get_publisher + +# non-critical telemetry: log errors instead of raising, retry acks harder +pub = get_publisher('telemetry.conditions.mysensor', + error_policy=ErrorPolicy(on_error=OnError.LOG, + retry=RetryPolicy(attempts=5, delay=0.1))) +``` + +### Journal (human-readable log over NATS) + +```python +from serverish.messenger import get_journalpublisher + +journal = get_journalpublisher('tic.journal.mytelescope.pipeline') +await journal.info('Calibration started') # never raises by default (on_error=LOG) +``` + +## Reading + +Readers are async iterators yielding `(data, meta)` pairs. + +### Follow a subject live + +```python +from serverish.messenger import get_reader + +async with get_reader('telemetry.weather.davis', deliver_policy='last') as reader: + async for data, meta in reader: # last known value first, then live + print(data['measurements']['temperature']) +``` + +### Drain and stop (batch processing) + +`DeliverAll(until=END_OF_DATA)` — read everything the server has, then stop. +End-of-data is server-confirmed (`num_pending == 0`), there is no fixed waiting. + +```python +from serverish.messenger import DeliverAll, Until, get_reader + +reader = get_reader('tic.status.zb08.fits.pipeline.raw', + deliver_policy=DeliverAll(until=Until.END_OF_DATA)) +async for data, meta in reader: + process(data) +# loop simply ends when the subject is drained +``` + +The legacy spelling `get_reader(subject, deliver_policy='all', nowait=True)` +does the same and stays supported. + +### Time window (e.g. last 48 h of telemetry) + +```python +from datetime import datetime, timedelta, timezone +from serverish.messenger import DeliverFromTime, Until, get_reader + +start = datetime.now(timezone.utc) - timedelta(hours=48) +reader = get_reader('telemetry.power.>', + deliver_policy=DeliverFromTime(start, until=Until.END_OF_DATA)) +async for data, meta in reader: + ... +``` + +### Snapshot of many subjects (dashboard pattern) + +One message per subject — the latest — then stop: + +```python +from serverish.messenger import DeliverLastPerSubject, Until, get_reader + +reader = get_reader('svc.status.>', + deliver_policy=DeliverLastPerSubject(until=Until.END_OF_DATA)) +snapshot = {meta['nats']['subject']: data async for data, meta in reader} +``` + +### Callback instead of iteration + +```python +from serverish.messenger import get_callbacksubscriber + +def on_status(data, meta): + update_ui(data) # return False to stop the subscription + +sub = get_callbacksubscriber('tic.status.zb08.dome.shutterstatus', deliver_policy='last') +await sub.open() +await sub.subscribe(on_status) +... +await sub.close() +``` + +### One-shot read + +```python +from serverish.messenger import single_read + +data, meta = await single_read('tic.config.observatory') +``` + +## Policies + +Grouped, typed configuration. Strings like `'last'` are permanent sugar for +the simple cases; policy objects unlock the rest. `None` fields always mean +"driver default"; "no limit" is the explicit `UNLIMITED`. + +| kwarg | accepts | controls | +|---|---|---| +| `deliver_policy` | `'all'/'last'/'new'/'last_per_subject'` or `DeliverAll/Last/New/LastPerSubject/FromTime(start)/FromSeq(seq)` (+ `until`) | where reading starts, when it stops | +| `error_policy` | `ErrorPolicy(on_error, on_missed, retry=RetryPolicy(...))` | reaction to failures; retry pacing (reader reconnects, publisher acks) | +| `batch_policy` | `BatchPolicy(max_batch, dynamic, initial_batch, max_time_in_memory)` | reader prefetch sizing | + +Combining a policy object with its legacy satellite kwarg (e.g. +`DeliverLast()` + `nowait=True`) raises `PolicyConflict` — express everything +in the object. + +### Batching: memory vs round trips + +Default is **dynamic**: pulls start tiny and follow your actual consumption +rate so prefetched messages wait at most ~5 s in memory, never more than +10 000 per pull. Slow consumers get small pulls automatically; fast bulk +readers over high-latency links ramp up (each pull costs a network round trip). + +```python +from serverish.messenger import BatchPolicy, get_reader + +# constrained device: keep at most 50 messages in memory, ~1 s residency +reader = get_reader(subject, batch_policy=BatchPolicy(max_batch=50, max_time_in_memory=1.0)) + +# pin the pre-2.3 static behaviour +reader = get_reader(subject, batch_policy=BatchPolicy(dynamic=False, max_batch=100)) +``` + +## KV buckets + +Key-value store over JetStream. Values are regular serverish envelopes. + +```python +from serverish.messenger import get_kvstore, kv_get, kv_put + +async with get_kvstore('observatory-state', create_bucket=True) as store: + await store.put('dome.position', {'az': 123.5}) + data, meta = await store.get('dome.position') + keys = await store.keys() + +data, meta = await kv_get('observatory-state', 'dome.position') # one-shot +``` + +Watch keys for changes (iterator or callback): + +```python +from serverish.messenger import get_kvreader + +async with get_kvreader('observatory-state', key='dome.>') as watcher: + async for data, meta in watcher: # current values first, then changes + if data is None: # delete/purge marker + print('deleted:', meta['kv']['key']) +``` + +## RPC + +Request/reply over core NATS (not JetStream — use a non-stream subject): + +```python +from serverish.messenger import get_rpcresponder, request + +def handler(rpc): + rpc.set_response(data={'sum': rpc.data['a'] + rpc.data['b']}) + +responder = get_rpcresponder('svc.rpc.my-service.add') +await responder.open() +await responder.register_function(handler) + +data, meta = await request('svc.rpc.my-service.add', data={'a': 2, 'b': 3}) +``` + +## Messages and errors + +Every message is an envelope: your `data` plus auto-filled `meta` +(`id`, `sender`, `ts` — 7-element UTC list, `message_type`, `tags`; after +delivery also `meta['nats']` with `seq`, `subject`, `num_pending`, ...). + +All library exceptions derive from `ServerishError`: + +```python +from serverish.base import ServerishError, MessengerReaderStopped, MessengerPublishAckTimeout + +try: + await pub.publish(data=payload) +except MessengerPublishAckTimeout: + ... # no ack - message MAY have been delivered (dedup makes re-publish safe) +except ServerishError: + ... # anything else from serverish +``` + +## Lifecycle summary + +- Prefer `async with` for every driver; or `open()`/`close()` explicitly. +- A reader left open keeps a server-side consumer alive — close it. +- `Messenger.close()` closes forgotten drivers and disconnects. +- One-shot helpers (`single_publish`, `single_read`, `kv_get`, `kv_put`) + manage open/close for you. diff --git a/docs/design/POLICIES.md b/docs/design/POLICIES.md deleted file mode 100644 index 016640b..0000000 --- a/docs/design/POLICIES.md +++ /dev/null @@ -1,185 +0,0 @@ -# Design: Policy objects for Messenger drivers - -Status: DRAFT for discussion (2026-08-06) -Goals: grouped, typed configuration (policies) • backward compatibility for -the existing ecosystem • isolation from raw NATS API • per-driver defaults. - -Grounded in a usage scan of all serverish consumers (TOI, oca-fits-proc, -halina9000, ocabox-tcs, oca_monitor, pms, ocabox): -- `deliver_policy` **strings** + `opt_start_time` are used everywhere → hard compat surface; -- `error_behavior` / `on_missed_messages` / `raise_on_publish_error` are used **nowhere** → free to restructure; -- ofp holds 8 long-lived `get_singlepublisher` objects publishing repeatedly → confirms single-publisher misuse; -- `ensure_open` has no external users. - -## 1. Policy taxonomy - -All policies are **frozen dataclasses** (immutable → safe as shared defaults; -no mutable-default traps, cf. #34). They live in -`serverish/messenger/policies.py` and are exported from `serverish.messenger`. - -### 1.1 DeliveryPolicy — where to start, when to stop (readers) - -Variants as types — invalid combinations are unrepresentable (no more -"by_start_time requires opt_start_time" runtime checks): - -```python -DeliverAll() # deliver_policy='all' -DeliverLast() # 'last' -DeliverNew() # 'new' -DeliverLastPerSubject() # 'last_per_subject' -DeliverFromTime(start: datetime) # 'by_start_time' + opt_start_time -DeliverFromSeq(seq: int) # 'by_start_sequence' + opt_start_seq -``` - -Common field on the base class: - -```python -until: Until = Until.FOREVER # FOREVER — live subscription (today: nowait=False) - # END_OF_DATA — stop when server confirms drained (today: nowait=True) -``` - -Driver-specific stop conditions (progress reader's `stop_when_done`, -document reader's `initial_wait`) stay as driver params — they are not -delivery semantics. Translation to NATS consumer config happens in ONE -place (`DeliveryPolicy.to_consumer_cfg()`), which is the isolation point. - -### 1.2 ErrorPolicy — what to do when things break (readers + publishers) - -```python -UNLIMITED = math.inf # explicit "no limit"; None NEVER means infinity - # (deliberately NOT named FOREVER — Until.FOREVER is a - # different concept: "keep following live data") - -@dataclass(frozen=True) -class RetryPolicy: - attempts: int | float | None = None # 0 = no retry; UNLIMITED = keep retrying - delay: float | None = None # initial inter-attempt delay [s] - backoff: float | None = None # delay multiplier per attempt - max_delay: float | None = None - total_timeout: float | None = None # UNLIMITED = never give up on time budget - -@dataclass(frozen=True) -class ErrorPolicy: - on_error: OnError | None = None # RAISE | FINISH | WAIT (readers) / RAISE | LOG (publishers) - on_missed: OnMissed | None = None # SKIP | REPLAY (readers) - retry: RetryPolicy | None = None -``` - -Uniform rule (all policies, cf. §1.4): **`None` always and only means -"driver default"**. Unlimited is always the explicit `UNLIMITED`, never an -implicit meaning of None — e.g. today's WAIT-mode reader default becomes -`RetryPolicy(attempts=UNLIMITED, delay=0.2, max_delay=15, total_timeout=UNLIMITED)`, -stated in the driver, not hidden in a None. - -Absorbs: `error_behavior`, `on_missed_messages` (readers), -`raise_on_publish_error`, `ack_timeout_retries` (publishers). The reader's -WAIT-mode reconnect pacing (`min(0.2 + n/5, 15)`) becomes its default -RetryPolicy; the publisher's ack retries become -`RetryPolicy(attempts=2, delay=0)`. - -### 1.3 BatchPolicy — how much to prefetch (readers) - -```python -@dataclass(frozen=True) -class BatchPolicy: - max_batch: int | None = None # pull-size cap; the ONLY size a static - # policy needs, the upper bound of a dynamic one - dynamic: bool | None = None # None → driver default (today: False) - initial_batch: int | None = None # dynamic only: size of the first pull - max_time_in_memory: float | None = None # dynamic only: target residency of a - # prefetched message [s] -``` - -One class, explicit `dynamic` flag — not two classes and not a `batch` vs -`initial/max` field split. Rationale: `max_batch` means the same thing in -both modes (static: the pull size; dynamic: its upper bound), so the common -user expresses only the bound (`BatchPolicy(max_batch=500)`) and stays -**neutral on mode** — when the driver default later flips to -`dynamic=True`, those callers get the improvement for free. Only someone -who explicitly writes `dynamic=False` pins the static behaviour. Two -separate classes (or an `int` sugar) would force every caller to pick a -mode at write-time and freeze the ecosystem on static. Setting -`initial_batch`/`max_time_in_memory` together with `dynamic=False` is a -validation error. - -Dynamic mode: the reader measures the consumer's actual consumption rate -and sizes the next pull to `clamp(rate × max_time_in_memory, 1, max_batch)`. -Slow consumer → small pulls (bounded memory & residency); fast consumer on a -high-RTT link → pulls grow toward `max_batch` (fewer round trips — see the -production profiling: 34 k msgs over ~290 ms RTT, pull size 100→1000 = -3.3× faster). `num_pending` (server-stamped) additionally caps the request -at what actually exists, and growth is damped to ≤ 8× the previous pull — -a two-sample rate estimate can be wildly optimistic, and the damping turns -a potential memory spike into a geometric ramp-up (2→16→128→…). - -**Driver default (decided 2026-08-07): dynamic** — -`BatchPolicy(dynamic=True, initial_batch=2, max_time_in_memory=5.0, -max_batch=10_000)`. Static behaviour is pinned explicitly with -`BatchPolicy(dynamic=False, max_batch=…)` (default static size: 100). - -### 1.4 Defaults strategy - -Policy fields default to **None = "driver decides"**. Each driver documents -its effective defaults (e.g. `MsgJournalPublisher` → `on_error=LOG`; -`MsgPublisher` → `on_error=RAISE`; `MsgReader` → `until=FOREVER`, -fixed batch 100). This keeps one policy type usable across drivers with -different sensibilities, exactly as today's per-class param defaults. - -## 2. Backward compatibility & kwarg model - -**One kwarg per policy group, dual-typed, with scalar sugar that is -permanent — only the satellite kwargs get deprecated.** Nothing ships -pre-deprecated: the simple spellings are first-class API forever. - -| kwarg | permanent forms | absorbs (satellites, deprecated in Phase 3) | -|---|---|---| -| `deliver_policy` | `str` sugar (`'last'`, `'all'`, ...) or `DeliveryPolicy` object | `opt_start_time`, `nowait`, `opt_start_seq` (via consumer_cfg) | -| `batch_policy` | `BatchPolicy` object only — deliberately **no `int` sugar**: a scalar would mean "static forever" scattered across projects, blocking the planned flip of the driver default to `dynamic=True` | — (new; the pull size was never public API) | -| `error_policy` | `ErrorPolicy` object (no meaningful scalar) | `error_behavior`, `on_missed_messages`, `raise_on_publish_error`, `ack_timeout_retries` | - -- Scalar sugar maps 1:1 to a policy object (`'last'` ≡ `DeliverLast()`, - `batch=500` ≡ `BatchPolicy(batch=500)`) — one normalization helper - (`resolve_policies(**kwargs)`), used by all factories and drivers. -- Passing both a policy object and one of its satellite kwargs is an error. -- Satellite kwargs stay **silent** in Phase 1 (they are all over the - ecosystem); Phase 3 adds DeprecationWarning; removal not before 3.0. -- `consumer_cfg` passthrough stays as the documented, explicit escape hatch - to raw NATS — isolation with a pressure valve. -- Note: a flat `batch=int` reader param was prototyped on - `fix/fetch-available-eager-finish` (it produced the WAN profiling numbers - above) and then withdrawn before merge — batch control ships only as - `BatchPolicy`, in this design's PR. - -## 3. API cleanups riding along - -1. **Single publisher/reader off the public API**: the only legitimate use - is one-shot convenience — which `single_publish()` / `single_read()` - already provide. `get_singlepublisher` / `get_singlereader` - (static + module) get a DeprecationWarning pointing to `single_publish` - / `single_read` (one-shot) or `get_publisher` / `get_reader` (repeated). - Classes become internal. Migration note for ofp: its 8 long-lived - single-publishers should be plain `get_publisher` (publish works the - same, without per-publish open/close churn). -2. **Publish on a closed publisher**: one-time-per-instance - DeprecationWarning ("open the publisher or use `async with`"). Keeps - working; paves the way for `open()` gaining real semantics (e.g. early - stream-existence validation) without surprising anyone later. -3. **`ensure_open` docs**: explicitly document the restore-previous-state - semantics ("if the driver was closed, it is closed again after the - call") — the current name suggests it leaves the driver open. Internal - audit: used only by driver implementations (incl. KV one-shots); no - external users. Keep, document, and prefer explicit - `async with` in examples so the temporary-open pattern doesn't spread. - -## 4. Rollout - -- **Phase 1**: `policies.py` (Delivery/Error/Retry/Batch), normalization - layer, dual-typed `deliver_policy`, driver defaults as policies, - deprecations from §3. All existing tests must pass unchanged (that is the - compat proof), plus new policy tests. -- **Phase 2**: adaptive BatchPolicy (consumption-rate estimator in - `fetch_available`, uses `num_pending`). -- **Phase 3** (separate, later): DeprecationWarning on legacy flat kwargs; - removal not before a major version bump. - -Out of scope here: loop-lag detector (#36), KV document reader. diff --git a/serverish/messenger/messenger.py b/serverish/messenger/messenger.py index ded25b2..76e42f8 100644 --- a/serverish/messenger/messenger.py +++ b/serverish/messenger/messenger.py @@ -27,6 +27,7 @@ from serverish.base.idmanger import gen_id, gen_uid from serverish.base.manageable import Manageable from serverish.messenger.msgvalidator import MsgValidator +from serverish.messenger.policies import DeliveryPolicy from serverish.base.singleton import Singleton from serverish.base import MessengerCannotConnect @@ -350,7 +351,7 @@ def get_publisher(subject: str, **kwargs) -> "MsgPublisher": @staticmethod def get_reader(subject: str, - deliver_policy='all', + deliver_policy: str | DeliveryPolicy = 'all', opt_start_time=None, **kwargs) -> 'MsgReader': """Returns a reader for a given subject diff --git a/serverish/messenger/msg_reader.py b/serverish/messenger/msg_reader.py index 9b5f7e9..757e075 100644 --- a/serverish/messenger/msg_reader.py +++ b/serverish/messenger/msg_reader.py @@ -94,7 +94,7 @@ class MsgReader(MsgDriver): static_batch_default = 100 def __init__(self, subject, parent = None, - deliver_policy = 'all', + deliver_policy: str | DeliveryPolicy = 'all', opt_start_time = None, consumer_cfg=None, error_policy: ErrorPolicy | None = None, @@ -1042,7 +1042,7 @@ def __str__(self): def get_reader(subject: str, - deliver_policy='all', + deliver_policy: str | DeliveryPolicy = 'all', opt_start_time=None, **kwargs) -> 'MsgReader': """Returns a subscription for a given subject, manages single subscription From 30cfd412de06ee833dbf04db9beed1f8ae75e3da Mon Sep 17 00:00:00 2001 From: majkelx Date: Fri, 7 Aug 2026 17:47:27 +0200 Subject: [PATCH 3/5] minor in response to review --- serverish/messenger/msg_publisher.py | 13 ++++++++++--- serverish/messenger/msg_reader.py | 4 +++- tests/test_policies.py | 2 ++ tests/test_policies_nats.py | 14 +++++++++++++- 4 files changed, 28 insertions(+), 5 deletions(-) diff --git a/serverish/messenger/msg_publisher.py b/serverish/messenger/msg_publisher.py index 8656c9c..31306d3 100644 --- a/serverish/messenger/msg_publisher.py +++ b/serverish/messenger/msg_publisher.py @@ -74,8 +74,13 @@ def __init__(self, error_policy: ErrorPolicy | None = None, **kwargs) -> None: self._last_publish_time: float | None = None self._last_error: Exception | None = None self._warned_unopened: bool = False + self._was_opened: bool = False super().__init__(**kwargs) + async def open(self) -> None: + self._was_opened = True + await super().open() + @property def _ack_retry(self) -> RetryPolicy: """Resolved ack-retry pacing: error_policy.retry wins; otherwise the @@ -100,10 +105,12 @@ async def publish(self, data: dict | None = None, meta: dict | None = None, **kw """ if not self.is_open and not self._warned_unopened: self._warned_unopened = True + state = "closed" if self._was_opened else "never-opened" 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()", + f"Publishing on a {state} publisher ({self}). Open it explicitly " + f"(await pub.open() or 'async with pub:') - publishing on a publisher " + f"that is not open will become an error in serverish 3.0; " + f"for one-shot use single_publish()", DeprecationWarning, stacklevel=2) msg = self.messenger.create_msg(data, meta) bdata = self.messenger.encode(msg) diff --git a/serverish/messenger/msg_reader.py b/serverish/messenger/msg_reader.py index 757e075..4646304 100644 --- a/serverish/messenger/msg_reader.py +++ b/serverish/messenger/msg_reader.py @@ -641,7 +641,9 @@ def _next_batch_size(self) -> int: 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 + # the policy wins; otherwise honor the legacy mutable `reader.batch` + # attribute (initialized to the static default) + return self.batching.max_batch or self.batch max_batch = self.batching.max_batch or self.batch_defaults.max_batch rate = self._consumption_rate() if rate is None: diff --git a/tests/test_policies.py b/tests/test_policies.py index 5a0f761..7734cff 100644 --- a/tests/test_policies.py +++ b/tests/test_policies.py @@ -204,6 +204,8 @@ def test_reader_static_batch_from_policy(): assert reader._next_batch_size() == 7 plain_static = get_reader('test.policies.unit', batch_policy=BatchPolicy(dynamic=False)) assert plain_static._next_batch_size() == 100 # static default size + plain_static.batch = 42 # legacy mutable attribute stays honored in static mode + assert plain_static._next_batch_size() == 42 def test_reader_dynamic_batch_sizing(): diff --git a/tests/test_policies_nats.py b/tests/test_policies_nats.py index e84acfb..c67fcf7 100644 --- a/tests/test_policies_nats.py +++ b/tests/test_policies_nats.py @@ -84,7 +84,7 @@ async def flaky(*args, **kwargs): @pytest.mark.nats async def test_publish_on_unopened_publisher_warns_once(messenger, unique_subject): pub = get_publisher(unique_subject) - with pytest.warns(DeprecationWarning, match="never opened"): + with pytest.warns(DeprecationWarning, match="never-opened"): await pub.publish(data={'v': 1}) with warnings.catch_warnings(): warnings.simplefilter("error") # second publish must NOT warn again @@ -92,6 +92,18 @@ async def test_publish_on_unopened_publisher_warns_once(messenger, unique_subjec await pub.close() +@pytest.mark.nats +async def test_publish_on_closed_publisher_warns_accurately(messenger, unique_subject): + """A publisher that WAS opened and then closed must not be accused of + being 'never opened' (Copilot review, PR #40).""" + pub = get_publisher(unique_subject) + await pub.open() + await pub.publish(data={'v': 1}) + await pub.close() + with pytest.warns(DeprecationWarning, match="closed publisher"): + await pub.publish(data={'v': 2}) + + @pytest.mark.nats async def test_opened_publisher_does_not_warn(messenger, unique_subject): async with get_publisher(unique_subject) as pub: From 12bfc91f9d08005456cef670668724b5df5fa448 Mon Sep 17 00:00:00 2001 From: majkelx Date: Fri, 7 Aug 2026 18:02:31 +0200 Subject: [PATCH 4/5] tests: split suite for spped up --- pyproject.toml | 1 + 1 file changed, 1 insertion(+) diff --git a/pyproject.toml b/pyproject.toml index 584f831..9809601 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -28,6 +28,7 @@ rich = "^13.5.2" aiodns = ">=3.1.1,<5" testcontainers = {extras = ["nats"], version = "^4.14.2"} pytest-timeout = "^2.4.0" +pytest-xdist = "^3.8.0" [build-system] From 8d1dcecbd7c91c38a32559fcb414c44e8708bdbb Mon Sep 17 00:00:00 2001 From: majkelx Date: Fri, 7 Aug 2026 18:05:14 +0200 Subject: [PATCH 5/5] Fix UNLIMITED check by value, not identity --- serverish/messenger/msg_publisher.py | 2 +- tests/test_policies.py | 11 +++++++++++ 2 files changed, 12 insertions(+), 1 deletion(-) diff --git a/serverish/messenger/msg_publisher.py b/serverish/messenger/msg_publisher.py index 31306d3..77600e1 100644 --- a/serverish/messenger/msg_publisher.py +++ b/serverish/messenger/msg_publisher.py @@ -65,7 +65,7 @@ def __init__(self, error_policy: ErrorPolicy | None = None, **kwargs) -> None: 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): + and self.errors.retry.attempts != UNLIMITED): # equality, not identity: float('inf') is a distinct object kwargs['ack_timeout_retries'] = int(self.errors.retry.attempts) # --- end policy normalization --- # Health monitoring fields diff --git a/tests/test_policies.py b/tests/test_policies.py index 7734cff..48c4a18 100644 --- a/tests/test_policies.py +++ b/tests/test_policies.py @@ -283,6 +283,17 @@ def test_publisher_error_policy(): raise_on_publish_error=False) +def test_publisher_unlimited_attempts_by_value(): + """UNLIMITED must be recognized by value - float('inf') is a distinct + object from math.inf, identity checks would OverflowError on int(inf) + (Copilot review, PR #40).""" + from serverish.messenger import get_publisher + pub = get_publisher('test.policies.unit', + error_policy=ErrorPolicy(retry=RetryPolicy(attempts=float('inf')))) + assert pub._ack_retry.attempts == UNLIMITED + assert pub.ack_timeout_retries == 2 # legacy int param untouched by the unlimited policy + + def test_publisher_legacy_ack_retries_stay_live(): from serverish.messenger import get_publisher pub = get_publisher('test.policies.unit')