diff --git a/README.md b/README.md index 91c3870..5462cad 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,19 @@ this will install `nats-py` package. - Custom JetStream API replacing nats-py private internals access ## Changes +* 2.5.1 A reader which is open reads. A recovery cut short by the network could leave a reader open and deaf + for good: its consumer alive on the server, the answers reaching nobody (seen in a long-running service + after a power outage). The reader now replaces its consumer and subscription as one, the new pair first, + and judges them by the server's answers to its own pull requests instead of asking whether the consumer + exists - one JetStream API request per reader every 10 s less. A missing consumer is never a fatal error, + a missing stream still is. `RetryPolicy.total_timeout` of a reader limits an attempt to recover as well, + not only the pauses between attempts. +* 2.5 Deadlines and cancellation. **Python 3.11+.** `publish(timeout=X)` is now the deadline of the whole + call, retries included (it used to apply to each of up to 3 attempts); without it a publish gives up after + `RetryPolicy.total_timeout` (15 s; a journal entry after 5 s). New `RetryPolicy.attempt_timeout`. Every timeout + is a `MessengerTimeout`. A reader's `total_timeout` is a real deadline and its clock stops when the server + answers again. A cancelled `read_next()` always ends (a cancellation lost in the client library kept it + polling for ever). `wait_for_psce()` is deprecated: use `asyncio.timeout()`. * 2.4 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 diff --git a/doc/messenger.md b/doc/messenger.md index eb78e9b..b46cef7 100644 --- a/doc/messenger.md +++ b/doc/messenger.md @@ -61,29 +61,57 @@ from serverish.messenger import single_publish await single_publish('svc.status.my-service', data={'state': 'done'}) ``` -### Errors and retries +### Errors, retries and deadlines -By default a failed publish raises. Missing JetStream acks are retried -transparently: every publish carries a globally unique `Nats-Msg-Id`, so the -server drops a repeated copy. When retries are exhausted you get -`MessengerPublishAckTimeout` — note the message *may* have been delivered. +By default a failed publish raises. A publish waits for the server to confirm +that the message is stored. When the confirmation does not come within +`attempt_timeout` (5 s) the publisher asks again, and it gives up at its +deadline, `total_timeout` (15 s), with `MessengerPublishAckTimeout` — note the +message *may* have been delivered. `publish()` never takes longer than the deadline. -The server remembers message ids only for the stream's duplicate window -(2 minutes unless the stream is configured otherwise). The default retries -are immediate and stay far inside it. If you set a `RetryPolicy` whose -delays add up to more than the window, a late retry of a message that was -in fact stored is stored again — keep `total_timeout` below the window when -duplicates matter. +Both times belong to the publisher's `RetryPolicy`: ```python from serverish.messenger import ErrorPolicy, OnError, RetryPolicy, get_publisher -# non-critical telemetry: log errors instead of raising, retry acks harder +# a command that must not hang: confirmed within 3 s or an error +pub = get_publisher('svc.cmd.dome', error_policy=ErrorPolicy(retry=RetryPolicy(total_timeout=3.0))) + +# non-critical telemetry: log errors instead of raising, ask again every second for 10 s pub = get_publisher('telemetry.conditions.mysensor', error_policy=ErrorPolicy(on_error=OnError.LOG, - retry=RetryPolicy(attempts=5, delay=0.1))) + retry=RetryPolicy(attempt_timeout=1.0, total_timeout=10.0))) ``` +`await pub.publish(data, timeout=3.0)` is a shortcut for a single call: it +replaces `total_timeout` for that one publish. `RetryPolicy.attempts` can stop +the retries before the deadline; by default as many are made as fit in it. + +Asking again is safe: every publish carries a globally unique `Nats-Msg-Id`, +so the server drops a repeated copy. It remembers message ids only for the +stream's duplicate window (2 minutes unless the stream is configured +otherwise). The default deadline stays far inside it; if you set a +`total_timeout` longer than the window, a late retry of a message that was +in fact stored is stored again. + +A journal (`get_journalpublisher`) is best effort: its deadline is 5 s, the +failure is logged, never raised, and `await journal.info('...', timeout=0.5)` +waits even shorter for one entry. + +Every timeout of serverish is a `MessengerTimeout` (and a builtin `TimeoutError`). + +For readers the same `RetryPolicy` paces the recovery from failures: by +default a reader retries for ever (`total_timeout=UNLIMITED`); with a finite +`total_timeout` it raises when a failure lasts longer than that. The clock +starts with a failure and stops when the server answers again. + +A reader recovers by itself from everything that can heal: a lost connection, +a restarted server, an expired consumer, answers which stopped coming. Do not +wrap it in retry loops or watchdogs. Only what cannot heal stops it, with +`MessengerReaderConfigError`: no stream stores the subject, or the server +refuses the reader's configuration. + + ### Journal (human-readable log over NATS) ```python diff --git a/pyproject.toml b/pyproject.toml index 8b472b0..79e11a2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "serverish" -version = "2.4.0" +version = "2.5.1" description = "helpers for server alike projects" authors = ["Mikołaj Kałuszyński", "MMME team"] readme = "README.md" @@ -8,7 +8,7 @@ repository = "https://github.com/AkondLab/serverish" include = ["serverish/schema/*"] [tool.poetry.dependencies] -python = "^3.10" +python = "^3.11" aiodns = {version=">=3.2,<5", optional=true} param = "^2.1.0" nats-py = { version = "^2.7.2", optional= true} diff --git a/serverish/base/asyncio_util_functions.py b/serverish/base/asyncio_util_functions.py index 2f38b21..e88c8ce 100644 --- a/serverish/base/asyncio_util_functions.py +++ b/serverish/base/asyncio_util_functions.py @@ -1,25 +1,22 @@ import asyncio -from contextlib import suppress +import warnings async def wait_for_psce(fut, timeout): - """ - Meaning: psce - prevent silent cancel error \n - This is a changed method 'wait_for()' from library 'asyncio' that prevents 'CancelledError' from being ignored - while the internal task is finished. \n - Description of the problem: the base method 'wait_for()' also works in such a way that when the inner task - is finished or there is no place for it to stop, function 'wait_for()' will return the result even if the inner - task is canceled, so the 'CancelledError' will not be propagated higher. This method prevents this situation. + """Deprecated: use ``asyncio.timeout()``. Kept for API compatibility. - :param fut: future oc coroutine - :param timeout: timeout, czn be float or int number - """ - task = asyncio.ensure_future(fut) - try: - return await asyncio.shield(asyncio.wait_for(task, timeout=timeout)) - except asyncio.CancelledError: - task.cancel() - with suppress(asyncio.CancelledError): - await task - raise + The name meant "prevent silent cancel error": a workaround for an + ``asyncio.wait_for()`` defect of old Python versions, where a cancellation + arriving together with the result was lost. ``asyncio.timeout()`` has no + such defect:: + async with asyncio.timeout(timeout): + result = await fut + + :param fut: awaitable to wait for + :param timeout: seconds, or None for no limit + """ + warnings.warn("wait_for_psce() is deprecated: use 'async with asyncio.timeout(...)'", + DeprecationWarning, stacklevel=2) + async with asyncio.timeout(timeout): + return await fut diff --git a/serverish/base/exceptions.py b/serverish/base/exceptions.py index 8303053..bfd6c49 100644 --- a/serverish/base/exceptions.py +++ b/serverish/base/exceptions.py @@ -62,7 +62,16 @@ def __init__(self, *args, code: int | None = None, err_code: int | None = None, self.description = description -class MessengerTransportTimeout(MessengerTransportError, TimeoutError): +class MessengerTimeout(ServerishError, TimeoutError): + """Something did not happen in time - the common base of every serverish timeout + + Catch it (or the builtin `TimeoutError`) to handle a missing publish + confirmation, an unanswered request and a slow server alike. + """ + pass + + +class MessengerTransportTimeout(MessengerTransportError, MessengerTimeout): """Raised when the NATS server did not answer in time""" pass @@ -96,7 +105,7 @@ def __init__(self, subject:str) -> None: class MessengerRequestCanceled(MessengerRequestNoResponse): pass -class MessengerRequestTimeout(MessengerRequestNoResponse, TimeoutError): +class MessengerRequestTimeout(MessengerRequestNoResponse, MessengerTimeout): pass class MessengerRequestNoResultYet(MessengerRequestNoResponse): diff --git a/serverish/base/task_manager.py b/serverish/base/task_manager.py index 6ed4a2a..0c2cd9f 100644 --- a/serverish/base/task_manager.py +++ b/serverish/base/task_manager.py @@ -4,7 +4,6 @@ import param from serverish.base import StatusReport -from serverish.base.asyncio_util_functions import wait_for_psce from serverish.base.collector import Collector from serverish.base.hasstatuses import HasStatuses from serverish.base.idmanger import IdManager @@ -92,7 +91,8 @@ async def wait_for(self, timeout: float = None, cancel_on_timeout: bool = True): else: t = self.task try: - await asyncio.wait_for(t, timeout=timeout) + async with asyncio.timeout(timeout): + await t except asyncio.TimeoutError: logger.debug(f'wait_for Task {self.name} timeout (task canceling:{cancel_on_timeout})') raise @@ -145,7 +145,8 @@ async def cancel_all(self, timeout: float = 10): for task in tasks: try: - await wait_for_psce(task.task, timeout) + async with asyncio.timeout(timeout): + await task.task except asyncio.TimeoutError: logger.error(f'Timeout during all tasks cancellation.') except asyncio.CancelledError: diff --git a/serverish/connection/nats_errors.py b/serverish/connection/nats_errors.py index 43d7016..cf558a8 100644 --- a/serverish/connection/nats_errors.py +++ b/serverish/connection/nats_errors.py @@ -84,6 +84,35 @@ def nats_errors_translated(context: str) -> Iterator[None]: raise translate_nats_error(e, context) from e +#: JetStream API `err_code` of "stream not found" +_ERR_STREAM_NOT_FOUND = 10059 + + +def is_missing_stream(error: BaseException | None) -> bool: + """Tells whether a raw client library error says that the stream is not there + + A 404 from JetStream names what is missing by its `err_code`. The stream missing is + a deployment error; a consumer missing (10014) is ordinary life of ephemeral consumers. + The client library raises the 404 on its own, with no `err_code`, in one place a reader + can meet: when it finds no stream storing the subject. + """ + return (isinstance(error, nats.js.errors.NotFoundError) + and error.err_code in (None, _ERR_STREAM_NOT_FOUND)) + + +#: JetStream API `err_code` of "maximum consumers limit reached" +_ERR_CONSUMER_LIMIT = 10026 + + +def is_consumer_limit(error: BaseException | None) -> bool: + """Tells whether a raw client library error says that the stream takes no more consumers + + The server reports it as a 400, like a wrong configuration, but it is not one: a slot + is freed when any consumer of the stream goes away. + """ + return isinstance(error, nats.js.errors.BadRequestError) and error.err_code == _ERR_CONSUMER_LIMIT + + def jetstream_error_code(error: Exception) -> int | None: """Returns the HTTP-like status code of a JetStream API error, raw or translated""" if isinstance(error, MessengerJetStreamError): diff --git a/serverish/messenger/msg_core_read.py b/serverish/messenger/msg_core_read.py index 3523315..6cf6a93 100644 --- a/serverish/messenger/msg_core_read.py +++ b/serverish/messenger/msg_core_read.py @@ -164,7 +164,8 @@ async def read_next(self) -> tuple[dict, dict]: if self._stop.is_set(): raise MessengerReaderStopped try: - item = await asyncio.wait_for(self._queue.get(), timeout=1.0) + async with asyncio.timeout(1.0): + item = await self._queue.get() self._queue.task_done() return item except asyncio.TimeoutError: diff --git a/serverish/messenger/msg_document_read.py b/serverish/messenger/msg_document_read.py index f2c7348..2c1f5b5 100644 --- a/serverish/messenger/msg_document_read.py +++ b/serverish/messenger/msg_document_read.py @@ -110,10 +110,8 @@ async def open(self) -> None: # Initial read - wait for first document try: - data, meta = await asyncio.wait_for( - self.read_next(), - timeout=self.initial_wait - ) + async with asyncio.timeout(self.initial_wait): + data, meta = await self.read_next() await self.document._update(data) self._seq = meta['nats']['seq'] log.info(f"Document loaded from {self.subject}: seq={self._seq}") diff --git a/serverish/messenger/msg_journal_pub.py b/serverish/messenger/msg_journal_pub.py index 50c8ae0..16c11fa 100644 --- a/serverish/messenger/msg_journal_pub.py +++ b/serverish/messenger/msg_journal_pub.py @@ -16,6 +16,7 @@ from serverish.messenger import Messenger from serverish.messenger.messenger import MsgDriver from serverish.messenger.msg_publisher import MsgPublisher +from serverish.messenger.policies import UNLIMITED, RetryPolicy @dataclass @@ -55,8 +56,16 @@ class MsgJournalPublisher(MsgPublisher): Use this class for text based, log-like messages to be displayed for a user. Interface mimics `logging` module, but note, that methods are async. Note, that raise_on_publish_error is set to False by default, which differs from the default in MsgPublisher. + + A journal is best effort: an entry which the server does not confirm within + 5 s is given up (logged, not raised), so that a program on an unstable + network is not held up by its own journal. Pass ``timeout=`` to a logging + call to wait shorter (or longer) for a single entry. """ raise_on_publish_error = param.Boolean(default=False) # override default + #: one attempt, 5 s: a journal entry is not worth holding the program up + retry_defaults = RetryPolicy(attempts=UNLIMITED, delay=0.1, backoff=1.0, max_delay=UNLIMITED, + total_timeout=5.0, attempt_timeout=5.0) conversations = param.Dict(default={}, doc="Conversations being tracked") _nameToLevel = None @@ -90,7 +99,8 @@ async def log(self, icon: str | None = None, actions: list[str] | None = None, timestamp: datetime | Sequence | None = None, - meta: dict | None = None + meta: dict | None = None, + timeout: float | None = None ) -> JournalEntry: """Log a message to the journal. @@ -103,6 +113,8 @@ async def log(self, actions (list[str], optional): List of actions to display. Defaults to None. timestamp (datetime | Sequence | None, optional): Timestamp of the message. Defaults to None. meta (dict, optional): Additional metadata to publish. Defaults to None. + timeout (float, optional): how long to wait for the server to confirm the entry [s]; + None = the journal's default (5 s) """ if timestamp is None: timestamp = dt_utcnow_array() @@ -116,7 +128,7 @@ async def log(self, icon=icon, actions=actions ) - return await self.publish_journal_operation(entry, op='publish', meta=meta) + return await self.publish_journal_operation(entry, op='publish', meta=meta, timeout=timeout) async def debug(self, message: str, *args, **kwargs) -> JournalEntry: """Log a message with level DEBUG on the journal.""" @@ -142,20 +154,22 @@ async def critical(self, message: str, *args, **kwargs) -> JournalEntry: """Log a message with level CRITICAL on the journal.""" return await self.log(logging.CRITICAL, message, *args, **kwargs) - async def publish_journal_operation(self, entry: JournalEntry, op: str, meta=None) -> JournalEntry: + async def publish_journal_operation(self, entry: JournalEntry, op: str, meta=None, + timeout: float | None = None) -> JournalEntry: """Publish a journal operation Args: entry (JournalEntry): Journal entry to publish op (str): Operation to perform on the journal entry. One of 'publish', 'update', 'close' meta (dict, optional): Additional metadata to publish. Defaults to None. + timeout (float, optional): deadline for the publish [s]; None = the journal's default """ if meta is None: meta = {} meta.update({'message_type': f'journal.{op}'}) dentry = entry.to_dict() dentry['op'] = op - await self.publish(dentry, meta = meta) + await self.publish(dentry, meta=meta, timeout=timeout) return entry diff --git a/serverish/messenger/msg_publisher.py b/serverish/messenger/msg_publisher.py index f525018..72b76a5 100644 --- a/serverish/messenger/msg_publisher.py +++ b/serverish/messenger/msg_publisher.py @@ -3,6 +3,7 @@ import asyncio import time import warnings +from dataclasses import replace import jsonschema import nats.errors @@ -41,10 +42,15 @@ class MsgPublisher(MsgDriver): persistent (bool): Require the subject to be persistent (default `False`, `True` in serverish 3.0) raise_on_publish_error (bool): Raise on publish error (default `True`). When `False`, errors (a missing connection included) are logged and reported in the returned message instead - ack_timeout_retries (int): How many times to retry a publish for which no JetStream - ack arrived in time (default 2). Retries are deduplicated server-side via - ``Nats-Msg-Id``. When retries are exhausted, `MessengerPublishAckTimeout` is - raised (subject to `raise_on_publish_error`). + ack_timeout_retries (int | None): Legacy limit on the number of retries of a publish for + which no JetStream ack arrived in time; None (default) = as many as fit in the deadline. + Prefer ``error_policy=ErrorPolicy(retry=RetryPolicy(attempts=...))``. + + Timing: a publish asks the server again whenever the confirmation does not + arrive within `RetryPolicy.attempt_timeout` (5 s) and gives up at its + deadline with `MessengerPublishAckTimeout` (subject to `raise_on_publish_error`). + The deadline is ``publish(timeout=...)`` or, without it, `RetryPolicy.total_timeout` + (15 s). `publish()` never takes longer than that. Health monitoring: The publisher tracks publish statistics accessible via `health_status` property: @@ -57,14 +63,16 @@ class MsgPublisher(MsgDriver): persistent = param.Boolean(default=False, doc="Require the subject to be persistent: fail on open() and " "publish() if published messages are not stored") raise_on_publish_error = param.Boolean(default=True, doc="Raise on publish error") - ack_timeout_retries = param.Integer(default=2, bounds=(0, None), - doc="Retries when JetStream ack does not arrive in time; " - "safe thanks to Nats-Msg-Id deduplication") + ack_timeout_retries = param.Integer(default=None, allow_None=True, bounds=(0, None), + doc="Legacy limit on retries when the JetStream ack does not arrive " + "in time; None = as many as fit in the deadline") #: 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=UNLIMITED, total_timeout=UNLIMITED) + #: Ask again after 5 s without a confirmation, give up after 15 s. The short + #: pause keeps a publish which fails at once from spinning until its deadline. + retry_defaults = RetryPolicy(attempts=UNLIMITED, delay=0.1, backoff=1.0, max_delay=UNLIMITED, + total_timeout=15.0, attempt_timeout=5.0) def __init__(self, error_policy: ErrorPolicy | None = None, **kwargs) -> None: # --- policy normalization (docs/design/POLICIES.md) --- @@ -113,12 +121,15 @@ def _ack_retry(self) -> RetryPolicy: 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: + async def publish(self, data: dict | None = None, meta: dict | None = None, + timeout: float | None = None, **kwargs) -> dict: """Publishes a messages to publisher subject Args: data (dict): message data meta (dict): message metadata + timeout (float): deadline for the whole call in seconds, retries included; + None = `RetryPolicy.total_timeout` of the publisher (15 s by default) kwargs: additional arguments to pass to the connection Returns: dict: published message, or message to be published with the `error` tag in meta if failed @@ -157,7 +168,7 @@ async def publish(self, data: dict | None = None, meta: dict | None = None, **kw # silently dropping messages within the stream's duplicate window. headers.setdefault('Nats-Msg-Id', f"{self.messenger.instance_id}:{msg['meta']['id']}") try: - await self._publish_with_ack_retries(bdata, headers, msg['meta']['id'], **kwargs) + await self._publish_with_ack_retries(bdata, headers, msg['meta']['id'], timeout, **kwargs) # Track successful publish self._publish_count += 1 self._last_publish_time = time.monotonic() @@ -200,8 +211,9 @@ async def publish(self, data: dict | None = None, meta: dict | None = None, **kw msg['meta']['status'] = str(error) return msg - async def _publish_with_ack_retries(self, bdata: bytes, headers: dict, msg_id: str, **kwargs) -> None: - """Publishes to JetStream, retrying when the ack does not arrive in time + async def _publish_with_ack_retries(self, bdata: bytes, headers: dict, msg_id: str, + timeout: float | None = None, **kwargs) -> None: + """Publishes to JetStream, asking again when the ack does not arrive in time A missing ack does not mean the message was not delivered - it may well be stored while the ack was processed too late (e.g. starved @@ -210,38 +222,50 @@ async def _publish_with_ack_retries(self, bdata: bytes, headers: dict, msg_id: s (2 minutes unless the stream says otherwise): a retry sequence which fits in the window never produces duplicates, a retry sent later can. + The deadline (`timeout`, else the policy's `total_timeout`) is hard: + an attempt is cut at it and none starts after it. + Raises: MessengerNotConnected: JetStream is not connected - MessengerPublishAckTimeout: no ack after all retries + MessengerPublishAckTimeout: no ack before the deadline, or after all the retries allowed """ js = self.connection.js if js is None: raise MessengerNotConnected(f"Trying to publish to subject '{self.subject}' failed. " f"JetStream not connected") retry = self._ack_retry - start = time.monotonic() + if timeout is not None: + retry = replace(retry, total_timeout=timeout) + loop = asyncio.get_running_loop() + start = loop.time() attempt = 0 while True: + # time for this attempt: its own allowance, cut at the deadline + budget = min(retry.attempt_timeout, retry.total_timeout - (loop.time() - start)) try: - await js.publish(self.subject, bdata, headers=headers, **kwargs) + if budget <= 0: + raise TimeoutError + async with asyncio.timeout(None if budget == UNLIMITED else budget): + # the budget goes to the client library as it is, UNLIMITED included: left + # out, the library would put its own default (5 s) in place of "no limit" + await js.publish(self.subject, bdata, headers=headers, timeout=budget, **kwargs) return - except nats.errors.TimeoutError as e: - elapsed = time.monotonic() - start - if retry.keeps_retrying(attempt, elapsed): - delay = retry.delay_for(attempt, elapsed) - log.warning(f"No JetStream ack for message {msg_id} on '{self.subject}' " - 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 {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 + except TimeoutError as e: + elapsed = loop.time() - start + if not retry.keeps_retrying(attempt, elapsed): + raise MessengerPublishAckTimeout( + f"No JetStream ack for message {msg_id} on '{self.subject}' within {elapsed:.1f}s " + f"({attempt + 1} attempt(s)). The message MAY have been delivered - a missing ack " + f"often means a starved client event loop, not a delivery failure") from e + delay = retry.delay_for(attempt, elapsed) + log.warning(f"No JetStream ack for message {msg_id} on '{self.subject}' " + f"(attempt {attempt + 1}), retrying" + + (f" in {delay:.1f}s" if delay else "") + + " - deduplicated by Nats-Msg-Id") + # also with no delay: an attempt which fails without waiting for anything must + # not keep the event loop (and a cancellation of this call) from running + await asyncio.sleep(delay) + attempt += 1 @property def health_status(self) -> dict: diff --git a/serverish/messenger/msg_reader.py b/serverish/messenger/msg_reader.py index 2a04786..45ad79b 100644 --- a/serverish/messenger/msg_reader.py +++ b/serverish/messenger/msg_reader.py @@ -2,6 +2,7 @@ import logging import asyncio +import contextlib from asyncio import Event from collections import deque from datetime import datetime, timezone @@ -16,12 +17,14 @@ from nats.js import JetStreamContext from nats.js.api import DeliverPolicy, ConsumerConfig -from serverish.base import BurstLog, wait_for_psce +from serverish.base import BurstLog from serverish.base.exceptions import (MessengerReaderStopped, MessengerReaderConfigError, MessengerNotConnected, MessengerJetStreamError, - MessengerReaderAlreadyOpen, MessengerRequestTimeout) + MessengerReaderAlreadyOpen, MessengerRequestTimeout, + MessengerTransportTimeout) from serverish.base.fifoset import FifoSet -from serverish.connection.nats_errors import nats_errors_translated, translate_nats_error +from serverish.connection.nats_errors import (is_consumer_limit, is_missing_stream, nats_errors_translated, + translate_nats_error) from serverish.messenger import Messenger from serverish.messenger.messenger import MsgDriver, subject_family from serverish.messenger.policies import (UNLIMITED, BatchPolicy, DeliveryPolicy, ErrorPolicy, @@ -35,6 +38,32 @@ class _ReconnectNeededError(Exception): pass + +@dataclass(eq=False) +class _ConsumerSession: + """A JetStream consumer together with the local subscription its answers arrive on + + The two are useful only as a pair: the consumer answers pull requests to the inbox + of the subscription. A reader holds one session, and replaces it as a whole. + + The session is judged by the only thing that matters to the reader: do the server's + answers to its pull requests arrive. "The consumer exists on the server" proves less - + it stays true for a session whose subscription is lost. + """ + subscription: JetStreamContext.PullSubscription + stream: str | None = None # None: not known (a subscription assigned from outside) + consumer: str | None = None + unanswered: int = 0 # pull requests in a row the server did not answer + gone: bool = False # the server said the consumer does not exist any more + + def needs_replacement(self, unanswered_limit: int) -> str | None: + """Tells why the session cannot serve the reader any more, None if it can""" + if self.gone: + return "the consumer is gone (expired or deleted)" + if self.unanswered >= unanswered_limit: + return f"{self.unanswered} pull requests in a row got no answer" + return None + class MsgReader(MsgDriver): """A class for reading data from a Messenger subject @@ -158,11 +187,12 @@ def __init__(self, subject, parent = None, 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._server_answered: bool = False # did the server answer the last fetch_available() 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._session: _ConsumerSession | None = None self.push_subscription: JetStreamContext.PushSubscription | None = None self.last_seq: int | None = None self._stop: Event = Event() @@ -187,6 +217,23 @@ def __init__(self, subject, parent = None, log.debug(f"Created {self}") + #: Pull requests in a row without any answer after which the session is replaced. + #: One lost answer may be a hiccup of the link; the second one, asked for after + #: the retry pause, is a pattern. + unanswered_limit = 2 + #: A reader waiting for messages asks the server for an immediate answer at least + #: this often [s]: silence alone does not tell an idle subject from a lost session. + probe_interval = 10.0 + + @property + def pull_subscription(self) -> JetStreamContext.PullSubscription | None: + """The subscription of the current consumer session, None when the reader has none""" + return self._session.subscription if self._session is not None else None + + @pull_subscription.setter + def pull_subscription(self, subscription: JetStreamContext.PullSubscription | None) -> None: + self._session = _ConsumerSession(subscription) if subscription is not None else None + def __aiter__(self): log.debug(f"Entering iteration {self}") return self @@ -240,8 +287,39 @@ class EndIterationException(_LoopException): pass error: Exception = None error_since: float = field(default_factory=time.monotonic) # start of current error streak retries: int = 0 # retries scheduled in the current error streak (not loop passes) - last_consumer_check: float = field(default_factory=time.monotonic) - consumer_check_interval: float = 10.0 # Check consumer health every 10 seconds + deadline: asyncio.Timeout | None = None # limits the current loop pass while failing + # cancel requests seen by the calling task when read_next() started + cancel_requests: int = field(default_factory=lambda: asyncio.current_task().cancelling()) + + def recovered(self) -> None: + """Ends the current error streak: the reader works again, with data or without""" + if self.error is not None: + log.info(self.fmt("recovered")) + self.error = None + self.retries = 0 + if self.deadline is not None: + self.deadline.reschedule(None) # the deadline was the failure's, not the reader's + + @contextlib.asynccontextmanager + async def limited_by(self, deadline: asyncio.Timeout): + """Makes the deadline of a loop pass known to `recovered()`, for as long as the pass lasts""" + self.deadline = deadline + try: + yield + finally: + self.deadline = None + + def raise_if_cancelled(self) -> None: + """Raises CancelledError when the calling task was cancelled during this read_next() + + A cancellation can be lost below us: a timeout and a cancel landing in the + same loop step may surface as a plain TimeoutError (serverish#32), which the + waiting loop takes for "no message yet" and polls on for ever. The task still + knows it was asked to stop - honour that. + """ + if asyncio.current_task().cancelling() > self.cancel_requests: + raise asyncio.CancelledError + last_probe: float = field(default_factory=time.monotonic) # last pull request the server was asked to answer at once def async_shield(func): @functools.wraps(func) @@ -331,51 +409,33 @@ async def ensure_not_stopped(self) -> None: raise self.EndIterationException('stop') @async_shield - async def ensure_consumer(self) -> None: - # Proactive health check: verify consumer exists periodically (every 10s) - # This prevents silent failures when ephemeral consumers expire - now = time.monotonic() - should_check = ( - self.error is not None or # Always check after errors - (now - self.last_consumer_check) > self.consumer_check_interval # Periodic check - ) - - if should_check: - self.last_consumer_check = now - self.reader._last_health_check_time = now - try: - ci = await self.reader.pull_subscription.consumer_info() - # Consumer exists and is accessible - log.debug(self.fmt(f"Consumer health check OK: {ci.name}")) - except nats.js.errors.NotFoundError: - burst_log.warning("Consumer has gone", - self.fmt("Consumer has gone (detected by health check), recreating")) - await self.reader._reopen() - log.info(self.fmt(f"Consumer re-opened after health check detected loss")) - raise self.ContinueException('reopen') - except Exception as e: - burst_log.warning("Consumer health check failed", - self.fmt(f"Error during consumer health check: {e}, will try to recreate")) - await self.reader._reopen() - log.info(self.fmt(f"Consumer re-opened after health check error")) - raise self.ContinueException('reopen') - - @async_shield - async def check_reconnect_needed(self) -> None: - """Check if NATS reconnection occurred and recreate consumer if needed""" - if self.reader._reconnect_needed.is_set(): - self.reader._reconnect_needed.clear() - log.info(self.fmt("NATS reconnection detected, recreating consumer")) - await self.reader._reopen() - log.info(self.fmt("Consumer recreated after NATS reconnection")) - raise self.ContinueException('reconnect') + async def ensure_session(self) -> None: + """Replaces the consumer session when it cannot serve the reader any more + + What is wrong with the session is learnt from the pull requests themselves + (`fetch_available`), at no cost: an answer proves the whole path, from the + consumer on the server to the queue of this reader. + """ + reader = self.reader + if reader._reconnect_needed.is_set(): + reader._reconnect_needed.clear() + reason = "NATS reconnected" + elif reader._session is None: + reason = "the reader has no consumer session" + else: + reason = reader._session.needs_replacement(reader.unanswered_limit) + if reason is None: + return + burst_log.warning("Replacing the consumer session", + self.fmt(f"{reason}, replacing the consumer session")) + await reader._replace_session() + log.info(self.fmt("Consumer session replaced")) + raise self.ContinueException('session') @async_shield async def read_batch(self) -> None: if len(self.reader.messages) == 0: - # Check for reconnection before fetching - await self.check_reconnect_needed() - + session = self.reader._session # 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 @@ -383,8 +443,25 @@ async def read_batch(self) -> None: 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 + self.last_probe = time.monotonic() + if not self.reader._server_answered and not session.gone: + # Silence. Older servers (seen: 2.10) are also silent when the consumer + # is gone, newer ones say so (503) - ask, this once, which silence it is + session.gone = await self.reader._consumer_is_gone(session) + if session.gone: + # an ephemeral consumer expiring is ordinary life (a frozen client, + # a restarted server), not an error: the next pass replaces the session + raise self.ContinueException('gone') + if not self.reader._server_answered: + # No data and no "no data": the answers do not reach us. This is the + # failure, whatever the cause; the retry policy paces what follows + raise MessengerTransportTimeout( + f"The server did not answer a pull request for '{self.reader.subject}' " + f"within {fetch_timeout} s") + self.reader._last_health_check_time = self.last_probe + self.recovered() # the server answers: the failure is over, even with no data + + # If no messages were available (404 from the server), decide what to do if len(new_msgs) == 0: if self.reader.nowait: # Server confirmed no messages - finish iteration @@ -392,33 +469,26 @@ async def read_batch(self) -> None: raise self.EndIterationException('nowait') # Non-nowait mode: wait for NEW messages to arrive. - # Break into shorter intervals so the reconnect-needed - # check at the top of the loop runs often enough that - # low-rate subjects don't sit on a stale subscription - # for tens of seconds after a NATS reconnect. + # Break into shorter intervals so a NATS reconnect is noticed + # soon and low-rate subjects don't sit on a stale session. blocking_interval = 2.0 - max_wait_cycles = 50 # Total wait: 50 * 2s = 100s (unchanged) - - for cycle in range(max_wait_cycles): - # Check for reconnection before each wait cycle - if self.reader._reconnect_needed.is_set(): - self.reader._reconnect_needed.clear() - log.info(self.fmt("NATS reconnection detected during wait, recreating consumer")) - await self.reader._reopen() - raise self.ContinueException('reconnect_during_wait') - - log.debug(self.fmt(f"Waiting for messages (cycle {cycle + 1}/{max_wait_cycles}, {blocking_interval}s)")) + while not self.reader._reconnect_needed.is_set(): + self.raise_if_cancelled() + log.debug(self.fmt(f"Waiting for messages ({blocking_interval}s)")) try: - new_msgs = await self.reader.pull_subscription.fetch(1, timeout=blocking_interval) + new_msgs = await session.subscription.fetch(1, timeout=blocking_interval) if new_msgs: log.debug(self.fmt(f"Received {len(new_msgs)} new message(s)")) break except asyncio.TimeoutError: - # Timeout is normal. The consumer may have expired meanwhile, but asking - # the server after every cycle is a JetStream API request per reader - # every 2 s; the periodic health check asks often enough. - if time.monotonic() - self.last_consumer_check > self.consumer_check_interval: - break # back to the main loop: ensure_consumer() runs the health check + # Normal: no message came. It is also all a broken session would + # show here, so the main loop asks the server now and then - a + # no_wait pull is always answered, with messages or without + if time.monotonic() - self.last_probe > self.reader.probe_interval: + break + except nats.js.errors.ServiceUnavailableError: + session.gone = True # nobody answers the pull requests: no such consumer + break except Exception as e: burst_log.warning("Waiting for messages failed", self.fmt(f"Error waiting for messages: {e}")) @@ -439,36 +509,64 @@ def logput(self, msg: str) -> None: self.log.append(msg) st = _ReadNextState(self) + + def give_up(error: Exception): + log.error(st.fmt(f"retry budget exhausted " + f"(attempts={self._retry.attempts}, " + f"total_timeout={self._retry.total_timeout}s), " + f"raising: {error}")) + translated = translate_nats_error(error, f"Reading from '{self.subject}' failed") + if translated is error: + raise error + raise translated from error + + def recovery_deadline() -> float | None: + """Seconds left for the reader to recover, None when nothing limits this pass + + `total_timeout` is the deadline of a failure: it limits the attempts to recover + as well as the pauses between them. A healthy reader waits for messages for ever. + """ + if st.error is None or self.error_behavior != 'WAIT' or self._retry.total_timeout == UNLIMITED: + return None + return max(self._retry.total_timeout - (time.monotonic() - st.error_since), 0.0) + # The loop which tries hard to get some data while True: + st.raise_if_cancelled() + failing = st.error try: - # If the step method: - # - retuns data - it will raiseReturnException - # - corrected something - it will raise ContinueException - # - wants stop iteration - it will raise EndIterationException - # - encounters an unrecoverable error - it will raise ErrorException - # - did not do anything - it will return and next step will be called - # Those mthod should not raise any other exceptions, if they do, it is a bug, - # this is enforced by @async_shield (conversion of nay exception to ErrorException) - - # 0. Do we have some data to return already? - await st.pop_msg() - # 1. Check if iteration have been stopped externally - await st.ensure_not_stopped() - # 2. Is it at least open? - await st.ensure_open() - # 3. Check consumer, maybe ephemeral consumer is gone - await st.ensure_consumer() - # 4. Pull batch of messages - await st.read_batch() + async with asyncio.timeout(recovery_deadline()) as deadline, st.limited_by(deadline): + # If the step method: + # - retuns data - it will raiseReturnException + # - corrected something - it will raise ContinueException + # - wants stop iteration - it will raise EndIterationException + # - encounters an unrecoverable error - it will raise ErrorException + # - did not do anything - it will return and next step will be called + # Those mthod should not raise any other exceptions, if they do, it is a bug, + # this is enforced by @async_shield (conversion of nay exception to ErrorException) + + # 0. Do we have some data to return already? + await st.pop_msg() + # 1. Check if iteration have been stopped externally + await st.ensure_not_stopped() + # 2. Is it at least open? + await st.ensure_open() + # 3. Replace the consumer session if it stopped serving us + await st.ensure_session() + # 4. Pull batch of messages + await st.read_batch() + except TimeoutError: + if not deadline.expired(): + raise # not the deadline of this pass: nothing here may let one out + if st.error is not None: # else: recovered in the very moment the deadline fired + st.logput('deadline') + give_up(failing) # the failure outlasted total_timeout in the middle of an attempt except st.ContinueException as e: # one of the method did something st.logput(f'{e.task}-ok') log.debug(st.fmt(f"continue after: {e.task}")) except st.ReturnException as e: # we have data to return st.logput(f'{e.task}-ret') - if st.error is not None: - log.info(st.fmt(f"recovered")) - st.error = None + st.recovered() log.debug(st.fmt(f"data returned")) return e.data, e.meta except st.EndIterationException as e: @@ -490,9 +588,12 @@ def logput(self, msg: str) -> None: log.error(st.fmt(f"fatal reader configuration error (not retrying): {e.error}")) self._stop.set() raise e.error + # A missing consumer is not among them: ephemeral consumers come and go, + # the session is replaced. A missing stream is a deployment error. + # Neither is a stream which takes no more consumers: a slot may be freed. cause = e.error.__cause__ if isinstance(e.error, MessengerJetStreamError) else e.error - if isinstance(cause, (nats.js.errors.BadRequestError, - nats.js.errors.NotFoundError)): + refused = isinstance(cause, nats.js.errors.BadRequestError) and not is_consumer_limit(cause) + if refused or is_missing_stream(cause): log.error(st.fmt(f"fatal NATS error (not retrying): {e.error}")) self._stop.set() raise MessengerReaderConfigError( @@ -511,19 +612,14 @@ def logput(self, msg: str) -> None: case 'WAIT': elapsed = time.monotonic() - st.error_since if not self._retry.keeps_retrying(st.retries, elapsed): - 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}")) - error = translate_nats_error(e.error, f"Reading from '{self.subject}' failed") - if error is e.error: - raise e.error - raise error from e.error + give_up(e.error) wait_time = self._retry.delay_for(st.retries, elapsed) st.retries += 1 burst_log.warning("Reading failed, retrying", st.fmt(f"read_next error, (retry in {wait_time:.1f}s): {e.error}")) await asyncio.sleep(wait_time) + if time.monotonic() - st.error_since >= self._retry.total_timeout: + give_up(e.error) # the deadline passed during the pause: no retry after it case _: # should not be reached log.error(st.fmt(f"Invalid on_connection_close value {self.error_behavior}. " f"Report this as a bug to the serverish maintainers!")) @@ -557,10 +653,12 @@ async def fetch_available(self, batch=10, timeout=2.0): """ import json - pull_subscription = self.pull_subscription + session = self._session + pull_subscription = session.subscription msgs = [] needed = batch queue = pull_subscription._sub._pending_queue + self._server_answered = False # First, drain messages already in local queue while not queue.empty() and needed > 0: @@ -571,6 +669,9 @@ async def fetch_available(self, batch=10, timeout=2.0): if not status: # Skip status messages msgs.append(msg) needed -= 1 + # delivered by the server to this very queue: the session works + self._server_answered = True + session.unanswered = 0 queue.task_done() except Exception: pass @@ -616,7 +717,8 @@ def log_backstop(): break try: - msg = await asyncio.wait_for(queue.get(), timeout=remaining) + async with asyncio.timeout(remaining): + msg = await queue.get() except asyncio.TimeoutError: log_backstop() break @@ -625,6 +727,8 @@ def log_backstop(): break pull_subscription._sub._pending_size -= len(msg.data) queue.task_done() + self._server_answered = True # a message or a status: the server is there + session.unanswered = 0 status = JetStreamContext.is_status_msg(msg) if status == "404": @@ -635,10 +739,21 @@ def log_backstop(): # Request Timeout - server closed the request after a partial batch log.debug(f"{self} Server closed partial batch (408 status)") break + elif status == "503" or (status == "409" and "Consumer Deleted" in self._status_description(msg)): + # No responders for the pull request / the consumer deleted under it: + # the server is there, the consumer is not + log.debug(f"{self} Server reports the consumer is gone ({status} status)") + session.gone = True + break + elif status == "100": + continue # a heartbeat, the answer is still to come elif status: - # Other status message (shouldn't happen with pull, but handle it) - log.debug(f"{self} Received status: {status}") - continue + # The server refused the request (409: a limit of the consumer, a change of the + # cluster). Not "no data": the reader would never get any and call itself healthy + description = self._status_description(msg) + raise MessengerJetStreamError( + f"The server refused a pull request for '{self.subject}': {status} {description}", + code=int(status), description=description) # Real message msgs.append(msg) @@ -662,8 +777,15 @@ def log_backstop(): # num_pending unknown (no JS metadata): keep reading until a # closing status arrives or the backstop hits (degraded mode) + if not self._server_answered: + session.unanswered += 1 return msgs + @staticmethod + def _status_description(msg: Msg) -> str: + """Returns the description of a JetStream status message, '' if it has none""" + return (msg.headers or {}).get('Description') or '' + def _next_batch_size(self) -> int: """Pull size for the next request. @@ -728,37 +850,38 @@ async def open(self) -> None: MessengerRequestTimeout: the server did not create the consumer in time MessengerTransportError: any other failure of the underlying NATS client """ - if self.pull_subscription is not None: + if self._session is not None: raise MessengerReaderAlreadyOpen("Reader already open, do not reuse MsgReader instances") log.debug(f"Opening {self}") - js = self.connection.js - self.connection.add_reconnect_cb(self.on_nats_reconnect) + self._session = await self._create_session() + await super().open() + async def _create_session(self, resume: bool = False) -> _ConsumerSession: + """Creates a consumer on the server and the subscription its answers arrive on - consumer_conf = await self._create_consumer_cfg() - - log.debug(f"Creating pull subscription for {self}") - self.pull_subscription = await self._create_pull_subscribtion(consumer_conf) - - # if consumer_conf.deliver_policy != DeliverPolicy.NEW: - # log.debug(f"Creating pull subscription for {self}") - # self.pull_subscription = await self._create_pull_subscribtion(consumer_conf) - # else: - # log.debug(f"Creating push subscription for {self}") - # self.push_subscription = await js.subscribe(self.subject, - # config=consumer_conf) + One attempt, all or nothing: on any failure, a cancellation included, nothing + is left behind for the reader to hold. - # self._emptied.set() - await super().open() + Args: + resume: continue after the last message read, instead of the reader's delivery policy - async def _create_pull_subscribtion(self, consumer_conf: ConsumerConfig): - # Durable consumer is probably not needed (at least problematic) - # consumer_conf.durable_name = self.name if consumer_conf.durable_name is None else consumer_conf.durable_name + Raises: + MessengerReaderConfigError: no stream stores the subject, or the server refuses the reader + MessengerNotConnected: no connection to subscribe through + MessengerRequestTimeout: the server did not create the consumer in time + MessengerTransportError: any other failure of the underlying NATS client + """ if self.connection is None or self.connection.js is None: raise MessengerNotConnected('Cannot create pull subscription, not connected to NATS') js = self.connection.js + consumer_conf = await self._create_consumer_cfg() + if resume and self.last_seq is not None: + log.info(f"New consumer session of {self} continues from seq={self.last_seq + 1}") + consumer_conf.deliver_policy = DeliverPolicy.BY_START_SEQUENCE + consumer_conf.opt_start_time = None + consumer_conf.opt_start_seq = self.last_seq + 1 with nats_errors_translated(f"Creating consumer for '{self.subject}' failed"): try: subject, stream, filters = self.subject, None, None @@ -770,17 +893,146 @@ async def _create_pull_subscribtion(self, consumer_conf: ConsumerConfig): subject = filters[0] # only this part of the family is persistent elif filters: consumer_conf.filter_subjects = filters - ret = await js.pull_subscribe(subject, - durable=consumer_conf.durable_name, - stream=stream, - config=consumer_conf) + subscription = await js.pull_subscribe(subject, + durable=consumer_conf.durable_name, + stream=stream, + config=consumer_conf) except nats.errors.TimeoutError as e: raise MessengerRequestTimeout( f"Creating consumer for '{self.subject}' timed out: {e}") from e - ci = await ret.consumer_info() - if filters: - await self._verify_family_filters(ret, ci, filters) - return ret + except nats.js.errors.NotFoundError as e: + if is_missing_stream(e): + raise MessengerReaderConfigError( + f"No stream stores the subject '{self.subject}': {e}") from e + raise + try: + ci = await subscription.consumer_info() + if filters: + await self._verify_family_filters(subscription, ci, filters) + except BaseException: + # Subscribed, but not a session yet: nobody would ever hold this subscription. + # The consumer, if it was created, expires by itself - nobody pulls from it. + try: + await subscription.unsubscribe() + except Exception as e: + log.debug(f"{self} could not drop the subscription of a failed session: {e}") + raise + return _ConsumerSession(subscription, stream=ci.stream_name, consumer=ci.name) + + async def _consumer_is_gone(self, session: _ConsumerSession) -> bool: + """Tells whether the server knows the consumer of the session no more + + Asked only when a pull request got no answer, never as a routine: a consumer + which exists says nothing about the session being usable. + + Raises: + nats.js.errors.NotFoundError: the stream itself is gone + """ + try: + await session.subscription.consumer_info() + except nats.js.errors.NotFoundError as e: + if is_missing_stream(e): + raise + return True + except Exception as e: + log.debug(f"{self} no answer about the consumer either: {e}") + return False + + async def _replace_session(self) -> None: + """Gives the reader a new consumer session, continuing after the last message read + + The new session is complete before the old one is touched, so a failure leaves + the reader exactly as it was: with a session it can use, or try to replace again. + One attempt - whoever calls paces the retries (the reader loop: its RetryPolicy). + + Two cases cannot have "the new one first", and give the old consumer up before: + + - a consumer which carries a name fixed by the configuration (a durable one) is that + name. It is created anew, right after the last message read; binding the existing + one would leave what it has delivered and the reader has not read to be redelivered + later, out of order; + - a stream which takes no more consumers has no slot for a second one. + """ + old = self._session + if old is not None and not old.gone and await self._has_fixed_name(old): + await self._give_up_consumer(old) + try: + new = await self._create_session(resume=True) + except MessengerJetStreamError as e: + if old is None or old.gone or not is_consumer_limit(e.__cause__): + raise + await self._give_up_consumer(old) + new = await self._create_session(resume=True) + self._session = new + self._reconnect_count += 1 + if old is not None: + await self._discard_session(old, keep_consumer=old.gone) + + async def _give_up_consumer(self, session: _ConsumerSession) -> None: + """Deletes the consumer of a session the reader still holds, to make room for the next one + + The session stays the reader's, known to have nothing to read from: should the deletion + or the creation which follows fail, the next attempt starts from where this one stopped. + + Raises: + MessengerTransportError: the consumer could not be deleted - it may still be there + """ + await self._delete_consumer(session) + session.gone = True + + async def _has_fixed_name(self, session: _ConsumerSession) -> bool: + """Tells whether the next session would get the very consumer of this one + + Judged by the name the consumer really carries, not by what was asked for: the client + library names a consumer after `durable_name`, and today ignores `name` of the consumer + configuration (every such session gets a consumer of its own). + """ + fixed = {self.consumer_cfg.get('durable_name'), self.consumer_cfg.get('name')} - {None} + if not fixed: + return False + with nats_errors_translated(f"Identifying the consumer of '{self.subject}' failed"): + _, consumer = await self._identify(session) + return consumer in fixed + + async def _identify(self, session: _ConsumerSession) -> tuple[str, str]: + """Returns the stream and the name of the session's consumer, asking the server if not known""" + if session.consumer is None: # a subscription assigned from outside + ci = await session.subscription.consumer_info() + session.stream, session.consumer = ci.stream_name, ci.name + return session.stream, session.consumer + + async def _delete_consumer(self, session: _ConsumerSession) -> None: + """Deletes the consumer of the session; one which is not there any more is as good as deleted + + Raises: + MessengerTransportError: the consumer could not be deleted + """ + with nats_errors_translated(f"Deleting the consumer of '{self.subject}' failed"): + try: + stream, consumer = await self._identify(session) + await self.connection.js.delete_consumer(stream=stream, consumer=consumer) + except nats.js.errors.NotFoundError: + pass + + async def _discard_session(self, session: _ConsumerSession, keep_consumer: bool = False) -> None: + """Removes a session the reader does not hold any more; best effort, never fails + + Whatever cannot be removed now removes itself: a consumer nobody pulls from + expires on the server (`inactive_threshold`). + + Args: + keep_consumer: drop the subscription only - there is no consumer of this session to delete + """ + try: + await session.subscription.unsubscribe() + except Exception as e: + log.debug(f"{self} could not unsubscribe a discarded session: {e}") + if keep_consumer: + return + try: + await self._delete_consumer(session) + except Exception as e: + log.debug(f"{self} could not delete the consumer of a discarded session (it will expire): {e}") async def _resolve_family(self) -> tuple[str | None, list[str]]: """Finds where the subject family is stored and which of its patterns are @@ -834,62 +1086,6 @@ async def _verify_family_filters(self, subscription, consumer_info, filters: lis f"The server does not filter the subject family of '{self.subject}' (asked for {filters}, " f"got {enforced or 'no filter'}). Subject families on persistent subjects need NATS server 2.10 or newer") - async def _reopen(self) -> None: - """Reopens the pull subscription with retry logic and better error handling""" - self._reconnect_count += 1 - max_retries = 3 - for attempt in range(max_retries): - try: - # Clean up existing subscription. `unsubscribe()` only detaches the - # local handle; the JetStream consumer must be deleted explicitly - # or it leaks server-side across every reconnect (one orphan per - # reopen, accumulating until the stream's consumer cap or the - # process restarts). - if self.pull_subscription is not None: - try: - ci = await self.pull_subscription.consumer_info() - except Exception as e: - log.debug(f"Error fetching consumer_info during reopen attempt {attempt + 1}: {e}") - ci = None - try: - await self.pull_subscription.unsubscribe() - except Exception as e: - log.debug(f"Error unsubscribing during reopen attempt {attempt + 1}: {e}") - if ci is not None: - try: - await self.connection.js.delete_consumer(stream=ci.stream_name, consumer=ci.name) - except nats.js.errors.NotFoundError: - pass # already gone, nothing to delete - except Exception as e: - log.debug(f"Error deleting old consumer during reopen attempt {attempt + 1}: {e}") - - # Create new consumer configuration - consumer_conf = await self._create_consumer_cfg() - - # Set policy for 'new messages' if any message was received - if self.last_seq is not None: - log.info(f"Reopening {self}, from seq={self.last_seq + 1}") - consumer_conf.deliver_policy = DeliverPolicy.BY_START_SEQUENCE - consumer_conf.opt_start_time = None - consumer_conf.opt_start_seq = self.last_seq + 1 - - # Create new subscription - self.pull_subscription = await self._create_pull_subscribtion(consumer_conf=consumer_conf) - log.info(f"Successfully reopened {self} after {attempt + 1} attempts") - return # Success! - - except Exception as e: - if attempt == max_retries - 1: - log.error(f"Failed to reopen {self} after {max_retries} attempts: {e}") - raise - - wait_time = (attempt + 1) * 2 # Exponential backoff: 2s, 4s, 6s - burst_log.warning("Reopening the reader failed, retrying", - f"{self} Reopen attempt {attempt + 1} failed, retrying in {wait_time}s: {e}") - await asyncio.sleep(wait_time) - - - async def _create_consumer_cfg(self) -> ConsumerConfig: cfg = self.consumer_cfg.copy() @@ -934,21 +1130,11 @@ async def _create_consumer_cfg(self) -> ConsumerConfig: async def close(self) -> None: self.connection.remove_reconnect_cb(self.on_nats_reconnect) await super().close() - await self._close_pull_subscription() + session, self._session = self._session, None + if session is not None: + await self._discard_session(session) await self._close_push_subscription() - async def _close_pull_subscription(self) -> None: - if self.pull_subscription is not None: - try: - ci = await self.pull_subscription.consumer_info() - await self.pull_subscription.unsubscribe() - await self.connection.js.delete_consumer(stream=ci.stream_name, consumer=ci.name) - except nats.js.errors.NotFoundError: - pass # no consumer is ok - except Exception as e: - log.warning(f'Exception while closing PULL subscription: {e}') - self.pull_subscription = None - async def _close_push_subscription(self) -> None: if self.push_subscription is not None: await self.push_subscription.unsubscribe() @@ -982,7 +1168,8 @@ async def wait_for_empty(self, timeout: float | None = None) -> None: self._msg_processed.clear() while self.push_subscription is not None and self.push_subscription.pending_msgs > 0: to = timeout - (datetime.now() - now).total_seconds() if timeout is not None else None - await wait_for_psce(self._msg_processed.wait(), timeout=to) + async with asyncio.timeout(to): + await self._msg_processed.wait() self._msg_processed.clear() @@ -1043,9 +1230,9 @@ def health_status(self) -> dict: - messages_received: Total count of messages received - last_message_time: Timestamp of last message (monotonic time) or None - last_message_ago: Seconds since last message or None if no messages yet - - reconnect_count: Number of times the consumer was recreated + - reconnect_count: Number of times the consumer session was replaced - last_error: String representation of last error or None - - last_health_check_time: Timestamp of last health check or None + - last_health_check_time: Timestamp (monotonic time) of the last pull request the server answered, or None - pending_messages: Number of messages in local pending queue - pending_bytes: Bytes in local pending queue - connection_slow_consumers: Slow consumer count from connection (for diagnostics) diff --git a/serverish/messenger/msg_rpc_req.py b/serverish/messenger/msg_rpc_req.py index 69700b2..3994206 100644 --- a/serverish/messenger/msg_rpc_req.py +++ b/serverish/messenger/msg_rpc_req.py @@ -107,7 +107,8 @@ async def result(self, t:Task, wait: float | None = None) -> (dict, dict): if wait is None: raise MessengerRequestNoResultYet() try: - r = await asyncio.wait_for(asyncio.shield(t.task), wait) + async with asyncio.timeout(wait): + r = await asyncio.shield(t.task) except asyncio.TimeoutError: raise MessengerRequestNoResultYet() except asyncio.CancelledError: diff --git a/serverish/messenger/msg_single_read.py b/serverish/messenger/msg_single_read.py index 185ff38..824f72c 100644 --- a/serverish/messenger/msg_single_read.py +++ b/serverish/messenger/msg_single_read.py @@ -61,7 +61,8 @@ async def read(self, wait: float | None = None) -> tuple[dict, dict]: interval *= 2 try: - data, meta = await asyncio.wait_for(self.read_next(), timeout=wait) + async with asyncio.timeout(wait): + data, meta = await self.read_next() finally: await self.close() return data, meta diff --git a/serverish/messenger/policies.py b/serverish/messenger/policies.py index 31cd8c1..16d03a6 100644 --- a/serverish/messenger/policies.py +++ b/serverish/messenger/policies.py @@ -25,7 +25,7 @@ from __future__ import annotations import math -from dataclasses import dataclass, field, replace +from dataclasses import dataclass, field, fields, replace from datetime import datetime from enum import Enum @@ -205,13 +205,18 @@ 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`. + explicit no-limit marker for `attempts` and the time limits. + + `total_timeout` is a deadline, not a hint: no attempt and no pause reaches + past it. `attempt_timeout` is the time one attempt may take (publishers: + how long to wait for the server's confirmation before asking again). """ 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 + total_timeout: float | None = None # deadline for the whole operation [s]; UNLIMITED = none + attempt_timeout: float | None = None # time allowed for one attempt [s]; UNLIMITED = none def __post_init__(self): if self.attempts is not None and self.attempts != UNLIMITED and ( @@ -219,19 +224,17 @@ def __post_init__(self): 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'): + for name in ('delay', 'max_delay', 'total_timeout', 'attempt_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") + if self.attempt_timeout == 0: + raise ValueError("RetryPolicy.attempt_timeout must be positive: no attempt succeeds in no time") 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)))) + return RetryPolicy(**{f.name: own if (own := getattr(self, f.name)) is not None else getattr(defaults, f.name) + for f in fields(self)}) def delay_for(self, attempt: int, elapsed: float = 0.0) -> float: """Delay before retry number `attempt` (0-based), on a fully resolved policy. diff --git a/tests/test_cancellation.py b/tests/test_cancellation.py new file mode 100644 index 0000000..e4c1239 --- /dev/null +++ b/tests/test_cancellation.py @@ -0,0 +1,105 @@ +"""Cancellation must always win: a reader waiting for data ends when its task is cancelled.""" +import asyncio +import warnings + +import nats.errors +import pytest + +from serverish.base import wait_for_psce +from serverish.messenger import get_reader + + +@pytest.mark.nats +async def test_read_next_honours_a_cancellation_lost_below_it(messenger, unique_subject): + """serverish#32: a cancel swallowed in the fetch path must not leave read_next() polling for ever. + + The fetch below is the worst case seen in production: the cancellation + delivered to it comes out as a plain timeout ("no message yet"). + """ + reader = get_reader(unique_subject, deliver_policy='new') + await reader.open() + swallowed = [] + + async def fetch_losing_cancellation(*args, **kwargs): + try: + await asyncio.sleep(0.2) + except asyncio.CancelledError: + swallowed.append(1) + raise nats.errors.TimeoutError + + reader.pull_subscription.fetch = fetch_losing_cancellation + try: + task = asyncio.create_task(reader.read_next()) + await asyncio.sleep(1.0) # the reader is in its waiting loop + task.cancel() + done, pending = await asyncio.wait([task], timeout=5) + assert done, 'read_next() kept running after its task was cancelled' + assert task.cancelled() + assert swallowed, 'the test did not exercise the lost-cancellation path' + finally: + del reader.pull_subscription.fetch + await reader.close() + + +@pytest.mark.nats +async def test_read_next_under_timeout_returns_on_time(messenger, unique_subject): + """The caller's pattern of serverish#32: read with a timeout on a silent subject.""" + reader = get_reader(unique_subject, deliver_policy='new') + await reader.open() + try: + loop = asyncio.get_running_loop() + started = loop.time() + with pytest.raises(TimeoutError): + async with asyncio.timeout(2): + await reader.read_next() + assert loop.time() - started < 4 + finally: + await reader.close() + + +async def test_wait_for_psce_shim_delivers_result_and_warns(): + async def work(): + return 42 + + with pytest.warns(DeprecationWarning, match='asyncio.timeout'): + assert await wait_for_psce(work(), timeout=1) == 42 + + +async def test_wait_for_psce_shim_times_out_and_cancels_the_work(): + cancelled = asyncio.Event() + + async def work(): + try: + await asyncio.sleep(10) + except asyncio.CancelledError: + cancelled.set() + raise + + with warnings.catch_warnings(): + warnings.simplefilter('ignore', DeprecationWarning) + with pytest.raises(TimeoutError): + await wait_for_psce(work(), timeout=0.1) + assert cancelled.is_set() + + +async def test_wait_for_psce_shim_propagates_the_callers_cancellation(): + cancelled = asyncio.Event() + + async def work(): + try: + await asyncio.sleep(10) + except asyncio.CancelledError: + cancelled.set() + raise + + async def caller(): + with warnings.catch_warnings(): + warnings.simplefilter('ignore', DeprecationWarning) + await wait_for_psce(work(), timeout=10) + + task = asyncio.create_task(caller()) + await asyncio.sleep(0.1) + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + assert cancelled.is_set() diff --git a/tests/test_delivery_policies.py b/tests/test_delivery_policies.py index ea5c774..7e7cc53 100644 --- a/tests/test_delivery_policies.py +++ b/tests/test_delivery_policies.py @@ -242,7 +242,7 @@ async def test_reader_sequence_tracking(messenger, unique_subject): break # Force a reconnection with the reader - await sub._reopen() + await sub._replace_session() # Publish 3 more messages for i in range(3): diff --git a/tests/test_policies.py b/tests/test_policies.py index f0e912c..3f9a209 100644 --- a/tests/test_policies.py +++ b/tests/test_policies.py @@ -129,7 +129,9 @@ def test_publisher_retry_delay_is_not_capped_to_zero(): pub = get_publisher('test.policies.retry_delay', error_policy=ErrorPolicy(retry=RetryPolicy(attempts=5, delay=0.1))) assert pub._ack_retry.delay_for(0) == 0.1 - assert MsgPublisher.retry_defaults.delay_for(0) == 0.0 # no policy: immediate retries, as before + slow = get_publisher('test.policies.retry_delay', + error_policy=ErrorPolicy(retry=RetryPolicy(attempts=5, delay=3.0))) + assert slow._ack_retry.delay_for(0) == 3.0 def test_retry_keeps_retrying(): @@ -278,7 +280,10 @@ def test_driver_error_defaults(): 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 MsgPublisher.retry_defaults.attempts == UNLIMITED # as many as fit in the deadline... + assert MsgPublisher.retry_defaults.total_timeout == 15.0 # ...which is finite + assert MsgPublisher.retry_defaults.attempt_timeout == 5.0 + assert get_journalpublisher('test.policies.unit')._ack_retry.total_timeout == 5.0 # best effort assert MsgReader.retry_defaults.attempts == UNLIMITED assert MsgReader.retry_defaults.max_delay == 15.0 @@ -321,7 +326,7 @@ def test_publisher_unlimited_attempts_by_value(): 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 + assert pub.ack_timeout_retries is None # legacy int param untouched by the unlimited policy def test_publisher_legacy_ack_retries_stay_live(): diff --git a/tests/test_reader_session.py b/tests/test_reader_session.py new file mode 100644 index 0000000..54f4216 --- /dev/null +++ b/tests/test_reader_session.py @@ -0,0 +1,405 @@ +"""The consumer session of a reader: whatever fails around it, an open reader reads again by itself. + +The cases come from the observatory outage of 2026-11-14, where readers of a long-running +service stayed open and deaf for hours after the network was back. +""" +import asyncio +import json +import uuid + +import nats.errors +import nats.js.errors +import pytest + +from serverish.base import MessengerJetStreamError, MessengerReaderConfigError +from serverish.connection.nats_errors import is_consumer_limit, is_missing_stream +from serverish.messenger import BatchPolicy, ErrorPolicy, OnError, RetryPolicy, get_publisher, get_reader + +pytestmark = [pytest.mark.nats, pytest.mark.timeout(40)] + +FAST = ErrorPolicy(on_error=OnError.WAIT, retry=RetryPolicy(delay=0.05, max_delay=0.2)) + + +class ApiDown: + """JetStream refuses to create consumers, as on a link which lets nothing through in time""" + + def __init__(self, reader, delay: float = 0.0): + self.js = reader.connection.js + self.original = self.js.pull_subscribe + self.delay = delay + self.attempts = 0 + + def __enter__(self): + async def times_out(*args, **kwargs): + self.attempts += 1 + await asyncio.sleep(self.delay) + raise nats.errors.TimeoutError + + self.js.pull_subscribe = times_out + return self + + def __exit__(self, *exc): + self.js.pull_subscribe = self.original + + +class Reading: + """A reader being read in the background, a publisher to feed it""" + + def __init__(self, subject, **reader_kwargs): + self.reader = get_reader(subject, deliver_policy='new', **reader_kwargs) + self.reader.probe_interval = 0.5 + self.publisher = get_publisher(subject) + self.got: list[int] = [] + self.task: asyncio.Task | None = None + + async def __aenter__(self): + await self.publisher.open() + await self.reader.open() + self.task = asyncio.create_task(self._read()) + await self.publish_and_expect(0) # the reader works before anything is broken + return self + + async def __aexit__(self, *exc): + self.task.cancel() + await asyncio.gather(self.task, return_exceptions=True) + await self.reader.close() + await self.publisher.close() + + async def _read(self): + async for data, meta in self.reader: + self.got.append(data['n']) + + async def publish_and_expect(self, n: int, within: float = 15.0): + await self.publisher.publish(data={'n': n}) + async with asyncio.timeout(within): + while n not in self.got: + assert not self.task.done(), f'the reader ended: {self.task}' + await asyncio.sleep(0.02) + + +def test_only_a_missing_stream_is_a_deployment_error(): + assert is_missing_stream(nats.js.errors.NotFoundError()) # no stream stores the subject + assert is_missing_stream(nats.js.errors.NotFoundError(code=404, err_code=10059)) # stream not found + assert not is_missing_stream(nats.js.errors.NotFoundError(code=404, err_code=10014)) # consumer not found + assert not is_missing_stream(nats.js.errors.ServiceUnavailableError(code=503)) + assert not is_missing_stream(nats.errors.TimeoutError()) + + +def test_a_full_stream_is_not_a_wrong_configuration(): + assert is_consumer_limit(nats.js.errors.BadRequestError(code=400, err_code=10026)) # maximum consumers limit reached + assert not is_consumer_limit(nats.js.errors.BadRequestError(code=400, err_code=10012)) + assert not is_consumer_limit(nats.js.errors.NotFoundError(code=404, err_code=10014)) + + +async def test_subject_without_a_stream_is_refused_at_open(messenger): + reader = get_reader('nostream.for.this.subject', deliver_policy='new') + with pytest.raises(MessengerReaderConfigError, match='No stream stores'): + await reader.open() + assert reader.pull_subscription is None + + +async def test_reader_reads_again_after_its_subscription_is_lost(messenger, unique_subject): + """The state of the outage itself: the consumer lives, its answers reach nobody.""" + async with Reading(unique_subject, error_policy=FAST) as r: + lost = r.reader._session + await lost.subscription.unsubscribe() + await r.publish_and_expect(1) + assert r.reader._session is not lost + assert r.reader.health_status['reconnect_count'] == 1 + + +async def test_failed_replacement_leaves_a_working_reader_working(messenger, unique_subject): + """A replacement which cannot be made costs nothing: the session in use stays in use.""" + async with Reading(unique_subject, error_policy=FAST) as r: + session = r.reader._session + with ApiDown(r.reader) as api: + r.reader.reconnect() # what a NATS reconnect asks for + await r.publish_and_expect(1) # read while no consumer can be created + assert api.attempts == 1 # the failed attempt is not repeated: nothing is wrong + assert r.reader._session is session + await r.publish_and_expect(2) + + +async def test_lost_subscription_and_no_way_to_replace_it_for_a_while(messenger, unique_subject): + """Lost answers while consumers cannot be created: the reader keeps trying, and reads when it can.""" + async with Reading(unique_subject, error_policy=FAST) as r: + lost = r.reader._session + with ApiDown(r.reader, delay=0.2) as api: + await lost.subscription.unsubscribe() + async with asyncio.timeout(20): + while api.attempts < 4: + await asyncio.sleep(0.05) + assert r.reader._session is lost # nothing better was made, nothing was lost on the way + assert not r.task.done() + await r.publish_and_expect(1) + assert r.reader._session is not lost + + +async def test_expired_consumer_is_replaced_without_an_error(messenger, unique_subject): + """A consumer gone from the server is ordinary life, also for a reader told to raise on errors.""" + async with Reading(unique_subject, error_policy=ErrorPolicy(on_error=OnError.RAISE)) as r: + gone = r.reader._session + await r.reader.connection.js.delete_consumer(stream=gone.stream, consumer=gone.consumer) + await r.publish_and_expect(1) + assert r.reader._session is not gone + assert r.reader._last_error is None + + +async def test_missing_consumer_reported_by_the_api_does_not_stop_the_reader(messenger, unique_subject): + async with Reading(unique_subject, error_policy=FAST) as r: + real = r.reader.fetch_available + calls = [] + + async def consumer_not_found_once(*args, **kwargs): + calls.append(1) + if len(calls) == 1: + raise nats.js.errors.NotFoundError(code=404, err_code=10014, description='consumer not found') + return await real(*args, **kwargs) + + r.reader.fetch_available = consumer_not_found_once + try: + async with asyncio.timeout(10): + while not calls: + await asyncio.sleep(0.05) + await r.publish_and_expect(1) + finally: + del r.reader.fetch_available + + +async def test_cancelled_replacement_leaves_the_session_in_use(messenger, unique_subject): + async with Reading(unique_subject, error_policy=FAST) as r: + session = r.reader._session + with ApiDown(r.reader, delay=30) as api: + r.reader.reconnect() + async with asyncio.timeout(10): + while not api.attempts: + await asyncio.sleep(0.02) + r.task.cancel() # in the middle of creating the new session + await asyncio.gather(r.task, return_exceptions=True) + assert r.reader._session is session + r.task = asyncio.create_task(r._read()) + await r.publish_and_expect(1) + + +async def test_half_made_session_leaves_no_subscription_behind(messenger, unique_subject): + """The consumer is created, asking about it fails: the subscription made for it is dropped.""" + async with Reading(unique_subject, error_policy=FAST) as r: + js, nc = r.reader.connection.js, r.reader.connection.nc + original = js.pull_subscribe + made = [] + + async def subscribe_then_fail_info(*args, **kwargs): + subscription = await original(*args, **kwargs) + made.append(subscription) + + async def no_info(*a, **k): + raise nats.errors.TimeoutError + + subscription.consumer_info = no_info + return subscription + + before = len(nc._subs) + js.pull_subscribe = subscribe_then_fail_info + try: + with pytest.raises(Exception): + await r.reader._replace_session() + finally: + js.pull_subscribe = original + assert len(made) == 1 + assert len(nc._subs) == before + await r.publish_and_expect(1) + + +@pytest.mark.parametrize('named_by', ['durable_name', 'name']) +async def test_reader_with_a_named_consumer_gets_a_new_session(messenger, unique_subject, named_by): + """A durable consumer is its name: it is created anew, continuing after the last message read. + + `name` of the consumer configuration is ignored by the client library today - such a reader gets a + consumer of its own for every session. Either way the reader ends with one working consumer. + """ + name = f'{named_by}_' + unique_subject.rsplit('.', 1)[-1] + async with Reading(unique_subject, error_policy=FAST, **{named_by: name}) as r: + js, old = r.reader.connection.js, r.reader._session + await r.reader._replace_session() + new = r.reader._session + assert new is not old + info = await js.consumer_info(new.stream, new.consumer) # raises if it is not there + if new.consumer == old.consumer: # the same name: created anew, not bound + assert new.consumer == name and named_by == 'durable_name' + assert info.config.opt_start_seq == r.reader.last_seq + 1 + else: # another consumer: the old one is gone + with pytest.raises(nats.js.errors.NotFoundError): + await js.consumer_info(old.stream, old.consumer) + await r.publish_and_expect(1) + assert r.reader.health_status['reconnect_count'] == 1 # no second replacement was needed + + +async def test_durable_reader_loses_nothing_delivered_to_the_session_it_gave_up(messenger, unique_subject): + """Messages the server delivered to the old session and nobody read come again, in order, before newer ones.""" + durable = 'durable_' + unique_subject.rsplit('.', 1)[-1] + publisher = get_publisher(unique_subject) + reader = get_reader(unique_subject, deliver_policy='all', error_policy=FAST, durable_name=durable) + await publisher.open() + await reader.open() + try: + for n in range(5): + await publisher.publish(data={'n': n}) + assert (await reader.read_next())[0] == {'n': 0} + old = reader.pull_subscription + # the server delivers further messages to the session which is about to be given up + await reader.connection.nc.publish(old._nms, b'{"batch": 2, "no_wait": true}', old._deliver) + async with asyncio.timeout(5): + while old._sub._pending_queue.qsize() < 2: + await asyncio.sleep(0.02) + await reader._replace_session() + await publisher.publish(data={'n': 5}) + got = [] + async with asyncio.timeout(10): + while len(got) < 5: + got.append((await reader.read_next())[0]['n']) + assert got == [1, 2, 3, 4, 5] + finally: + await reader.close() + await publisher.close() + + +async def test_session_is_replaced_on_a_stream_with_no_spare_consumer(messenger): + """No slot for "the new one first": the old consumer is given up, the reader reads on.""" + token = uuid.uuid4().hex[:8] + js = messenger.connection.js + await js.add_stream(name=f'full_{token}', subjects=[f'full.{token}.>'], storage='memory', max_consumers=1) + try: + async with Reading(f'full.{token}.x', error_policy=FAST) as r: + old = r.reader._session + await r.reader._replace_session() + assert r.reader._session is not old + await r.publish_and_expect(1) + # ... also when the loop finds out by itself, and when it has to wait for the slot + await r.reader._session.subscription.unsubscribe() + await r.publish_and_expect(2) + finally: + await js.delete_stream(f'full_{token}') + + +async def test_consumer_which_could_not_be_given_up_stays_with_the_reader(messenger): + """No spare consumer and the old one cannot be deleted: the reader keeps it, to try again - and to read on.""" + token = uuid.uuid4().hex[:8] + js = messenger.connection.js + await js.add_stream(name=f'full_{token}', subjects=[f'full.{token}.>'], storage='memory', max_consumers=1) + original = js.delete_consumer + try: + async with Reading(f'full.{token}.x', error_policy=FAST) as r: + old = r.reader._session + + async def times_out(*args, **kwargs): + raise nats.errors.TimeoutError + + js.delete_consumer = times_out + with pytest.raises(Exception, match='Deleting the consumer'): + await r.reader._replace_session() + assert r.reader._session is old and not old.gone + await r.publish_and_expect(1) # the session it kept is the one which works + js.delete_consumer = original + await r.reader._replace_session() + assert r.reader._session is not old + await r.publish_and_expect(2) + finally: + js.delete_consumer = original + await js.delete_stream(f'full_{token}') + + +async def test_refused_pull_request_is_an_error_not_an_empty_subject(messenger, unique_subject): + """A consumer limited to smaller requests than the reader makes: every pull is refused, and the reader says so.""" + reader = get_reader(unique_subject, deliver_policy='new', error_policy=ErrorPolicy(on_error=OnError.RAISE), + batch_policy=BatchPolicy(dynamic=False, max_batch=5)) + await reader.open() + try: + js = reader.connection.js + info = await reader.pull_subscription.consumer_info() + config = {k: v for k, v in info.config.as_dict().items() if v is not None} | {'max_batch': 1} + await js._api_request(f'{js._prefix}.CONSUMER.CREATE.{info.stream_name}.{info.name}', + json.dumps({'stream_name': info.stream_name, 'config': config, + 'action': 'update'}).encode(), timeout=2) + with pytest.raises(MessengerJetStreamError, match='Exceeded MaxRequestBatch') as refused: + await reader.read_next() + assert refused.value.code == 409 + finally: + await reader.close() + + +async def test_messages_waiting_in_the_local_queue_are_delivered(messenger, unique_subject): + """Messages delivered by an earlier pull are an answer: no new pull is sent, and no error may be made of that.""" + publisher = get_publisher(unique_subject) + reader = get_reader(unique_subject, deliver_policy='all', error_policy=ErrorPolicy(on_error=OnError.RAISE)) + await publisher.open() + await reader.open() + try: + for n in range(2): + await publisher.publish(data={'n': n}) + subscription = reader.pull_subscription + # a pull whose answer nobody collected (a wait given up, a cancelled read): both messages are queued locally + await reader.connection.nc.publish(subscription._nms, b'{"batch": 2, "no_wait": true}', subscription._deliver) + async with asyncio.timeout(5): + while subscription._sub._pending_queue.qsize() < 2: + await asyncio.sleep(0.02) + got = [(await reader.read_next())[0]['n'] for _ in range(2)] + assert got == [0, 1] + finally: + await reader.close() + await publisher.close() + + +async def test_error_streak_ended_by_a_buffered_message(messenger, unique_subject): + """A failure followed by a message already in the buffer: the message is returned, the failure is over.""" + publisher = get_publisher(unique_subject) + reader = get_reader(unique_subject, deliver_policy='all', + error_policy=ErrorPolicy(on_error=OnError.WAIT, + retry=RetryPolicy(delay=0.05, max_delay=0.05, total_timeout=5.0))) + await publisher.open() + await reader.open() + real_unpack = reader.messenger.unpack_nats_msg + calls = [] + + def malformed_once(msg): + calls.append(1) + if len(calls) == 1: + raise ValueError('malformed message') + return real_unpack(msg) + + try: + for n in range(2): + await publisher.publish(data={'n': n}) + reader.messenger.unpack_nats_msg = malformed_once + data, meta = await asyncio.wait_for(reader.read_next(), timeout=5) + assert data == {'n': 1} + assert len(calls) == 2 + finally: + del reader.messenger.unpack_nats_msg + await reader.close() + await publisher.close() + + +async def test_recovery_attempt_does_not_outlast_the_deadline(messenger, unique_subject): + """total_timeout limits the attempt to recover, not only the pauses between attempts.""" + reader = get_reader(unique_subject, deliver_policy='new', + error_policy=ErrorPolicy(on_error=OnError.WAIT, + retry=RetryPolicy(delay=0.05, max_delay=0.05, total_timeout=1.0))) + await reader.open() + calls = [] + + async def fails_then_hangs(*args, **kwargs): + calls.append(1) + if len(calls) == 1: + raise RuntimeError('fetch failed') + await asyncio.sleep(30) + + reader.fetch_available = fails_then_hangs + loop = asyncio.get_running_loop() + try: + started = loop.time() + with pytest.raises(RuntimeError, match='fetch failed'): + await reader.read_next() + assert 0.9 <= loop.time() - started < 1.5 + finally: + del reader.fetch_available + await reader.close() diff --git a/tests/test_resl_consumer_expiry.py b/tests/test_resl_consumer_expiry.py index 3dcf62f..bb57190 100644 --- a/tests/test_resl_consumer_expiry.py +++ b/tests/test_resl_consumer_expiry.py @@ -125,7 +125,7 @@ async def test_reader_recreates_expired_consumer(resilience_messenger, nats_disr # --- 5. VERIFY RECOVERY --- # The reader's read_next loop should detect the missing consumer via - # ensure_consumer() and call _reopen() to recreate it automatically. + # pull request finds no consumer and replace the session automatically. # Publish a new message and verify the reader receives it. await pub.publish(data={'phase': 'after_expiry', 'n': 2}) @@ -189,7 +189,7 @@ async def test_consumer_expiry_health_status_transitions(resilience_messenger, n await _poll_until_connected(m, timeout=20) # Start a read_next in background so the reader's loop can detect expiry - # and recreate the consumer via _reopen() + # and replace the consumer session read_task = asyncio.create_task(reader.read_next()) # Give the reader loop time to detect the expired consumer and recreate await asyncio.sleep(3) diff --git a/tests/test_timeouts.py b/tests/test_timeouts.py new file mode 100644 index 0000000..fbf8a6e --- /dev/null +++ b/tests/test_timeouts.py @@ -0,0 +1,279 @@ +"""Deadlines: a publish never takes longer than its deadline, a reader never retries past its own.""" +import asyncio + +import nats.errors +import pytest + +from serverish.base import (MessengerPublishAckTimeout, MessengerRequestTimeout, MessengerTimeout, + MessengerTransportTimeout, ServerishError) +from serverish.messenger import (UNLIMITED, ErrorPolicy, OnError, RetryPolicy, get_journalpublisher, + get_publisher, get_reader) + + +class SilentServer: + """Replaces JetStream publish of a publisher with one which never gets a confirmation""" + + def __init__(self, publisher): + self.js = publisher.connection.js + self.original = self.js.publish + self.attempts: list[float] = [] # the time each attempt was given + + def __enter__(self): + async def never_acked(subject, payload, timeout=None, **kwargs): + self.attempts.append(timeout) + await asyncio.sleep(timeout) + raise nats.errors.TimeoutError + + self.js.publish = never_acked + return self + + def __exit__(self, *exc): + self.js.publish = self.original + + +def test_every_timeout_has_one_base(): + for cls in (MessengerPublishAckTimeout, MessengerTransportTimeout, MessengerRequestTimeout): + assert issubclass(cls, MessengerTimeout) + assert issubclass(MessengerTimeout, ServerishError) + assert issubclass(MessengerTimeout, TimeoutError) + + +@pytest.mark.nats +async def test_publish_timeout_is_the_deadline_of_the_whole_call(messenger, unique_subject): + """publish(timeout=X) returns or raises within X, retries included.""" + pub = get_publisher(unique_subject, + error_policy=ErrorPolicy(retry=RetryPolicy(attempt_timeout=0.4, delay=0.0))) + await pub.open() + loop = asyncio.get_running_loop() + try: + with SilentServer(pub) as server: + started = loop.time() + with pytest.raises(MessengerPublishAckTimeout, match='MAY have been delivered'): + await pub.publish(data={'v': 1}, timeout=1.0) + took = loop.time() - started + assert 0.9 <= took < 1.3 + assert len(server.attempts) == 3 # 0.4 + 0.4 + what was left + assert server.attempts[-1] == pytest.approx(0.2, abs=0.1) + finally: + await pub.close() + + +@pytest.mark.nats +async def test_publish_deadline_comes_from_the_policy(messenger, unique_subject): + """Without timeout= the deadline is RetryPolicy.total_timeout.""" + pub = get_publisher(unique_subject, + error_policy=ErrorPolicy(retry=RetryPolicy(total_timeout=0.6, attempt_timeout=0.25, + delay=0.0))) + await pub.open() + loop = asyncio.get_running_loop() + try: + with SilentServer(pub): + started = loop.time() + with pytest.raises(MessengerTimeout): + await pub.publish(data={'v': 1}) + assert 0.5 <= loop.time() - started < 0.9 + finally: + await pub.close() + + +@pytest.mark.nats +async def test_publish_attempts_still_limit_the_retries(messenger, unique_subject): + """A policy may stop earlier than the deadline.""" + pub = get_publisher(unique_subject, + error_policy=ErrorPolicy(retry=RetryPolicy(attempts=1, attempt_timeout=0.2, + delay=0.0, total_timeout=30.0))) + await pub.open() + try: + with SilentServer(pub) as server: + with pytest.raises(MessengerPublishAckTimeout): + await pub.publish(data={'v': 1}) + assert len(server.attempts) == 2 # the first try and one retry + finally: + await pub.close() + + +@pytest.mark.nats +async def test_unlimited_attempt_is_not_cut_by_the_client_library(messenger, unique_subject): + """UNLIMITED means no limit: the client library is not left to put its own default in.""" + pub = get_publisher(unique_subject, error_policy=ErrorPolicy( + retry=RetryPolicy(attempts=0, attempt_timeout=UNLIMITED, total_timeout=UNLIMITED))) + await pub.open() + js = pub.connection.js + original, default = js.publish, js._timeout + given = [] + + async def records(subject, payload, timeout=None, **kwargs): + given.append(timeout) + return await original(subject, payload, timeout=timeout, **kwargs) + + js.publish = records + js._timeout = 0.0001 # what the library would use if it were left to choose + try: + await pub.publish(data={'v': 1}) + assert given == [UNLIMITED] + finally: + js.publish, js._timeout = original, default + await pub.close() + + +def test_attempt_which_gets_no_time_is_refused(): + with pytest.raises(ValueError, match='attempt_timeout must be positive'): + RetryPolicy(attempt_timeout=0) + + +@pytest.mark.nats +async def test_publish_retried_without_a_pause_lets_the_loop_run(messenger, unique_subject): + """delay=0 and attempts which fail without waiting: other tasks still run, and the call can be cancelled.""" + pub = get_publisher(unique_subject, error_policy=ErrorPolicy(retry=RetryPolicy(delay=0.0, total_timeout=30.0))) + await pub.open() + js = pub.connection.js + original = js.publish + + async def fails_at_once(*args, **kwargs): + raise nats.errors.TimeoutError + + js.publish = fails_at_once + try: + publishing = asyncio.create_task(pub.publish(data={'v': 1})) + await asyncio.sleep(0.2) # runs only if the retry loop lets go of the event loop + publishing.cancel() + with pytest.raises(asyncio.CancelledError): + await asyncio.wait_for(publishing, timeout=2) + finally: + js.publish = original + await pub.close() + + +@pytest.mark.nats +async def test_publish_which_fails_at_once_does_not_spin(messenger, unique_subject): + """Default pacing: instant failures are spaced out, not repeated thousands of times.""" + pub = get_publisher(unique_subject) + await pub.open() + js = pub.connection.js + original = js.publish + calls = [] + + async def fails_at_once(*args, **kwargs): + calls.append(1) + raise nats.errors.TimeoutError + + js.publish = fails_at_once + try: + with pytest.raises(MessengerPublishAckTimeout): + await pub.publish(data={'v': 1}, timeout=1.0) + assert 5 <= len(calls) <= 12 + finally: + js.publish = original + await pub.close() + + +@pytest.mark.nats +async def test_publish_is_stored_once_when_the_ack_is_late(messenger, unique_subject): + """The happy path of a retry: the first attempt is stored, its ack lost, the retry deduplicated.""" + pub = get_publisher(unique_subject, error_policy=ErrorPolicy(retry=RetryPolicy(attempt_timeout=2.0))) + await pub.open() + await messenger.purge(unique_subject) + js = pub.connection.js + original = js.publish + calls = [] + + async def first_ack_lost(*args, **kwargs): + calls.append(1) + ack = await original(*args, **kwargs) + if len(calls) == 1: + raise nats.errors.TimeoutError + return ack + + js.publish = first_ack_lost + try: + await pub.publish(data={'v': 1}, timeout=5.0) + finally: + js.publish = original + await pub.close() + assert len(calls) == 2 + reader = get_reader(unique_subject, deliver_policy='all', nowait=True) + stored = [data async for data, meta in reader] + assert stored == [{'v': 1}] + + +@pytest.mark.nats +async def test_journal_does_not_hold_the_program_up(messenger, unique_subject): + """serverish#19: an unconfirmed journal entry is given up quickly and never raises.""" + journal = get_journalpublisher(unique_subject) + await journal.open() + loop = asyncio.get_running_loop() + try: + with SilentServer(journal): + started = loop.time() + await journal.info('on an unstable network', timeout=0.3) + assert loop.time() - started < 0.6 + assert journal._ack_retry.total_timeout == 5.0 # the default without timeout= + finally: + await journal.close() + + +@pytest.mark.nats +async def test_reader_does_not_retry_after_its_deadline(messenger, unique_subject): + """The pause before a retry ends at the deadline, and no retry starts there.""" + reader = get_reader(unique_subject, deliver_policy='new', + error_policy=ErrorPolicy(on_error=OnError.WAIT, + retry=RetryPolicy(delay=10.0, max_delay=10.0, total_timeout=0.5))) + calls = [] + + async def failing(*args, **kwargs): + calls.append(1) + raise RuntimeError('fetch failed') + + reader.fetch_available = failing + loop = asyncio.get_running_loop() + try: + started = loop.time() + with pytest.raises(RuntimeError, match='fetch failed'): + await reader.read_next() + assert 0.4 <= loop.time() - started < 1.0 + assert len(calls) == 1 # no attempt at or after the deadline + finally: + del reader.fetch_available + await reader.close() + + +@pytest.mark.nats +async def test_reader_error_streak_ends_when_the_server_answers(messenger, unique_subject): + """Two failures separated by a recovery are two streaks, also when no message came between them.""" + reader = get_reader(unique_subject, deliver_policy='new', + error_policy=ErrorPolicy(on_error=OnError.WAIT, + retry=RetryPolicy(delay=0.05, max_delay=0.05, total_timeout=1.0))) + await reader.open() + real_fetch_available = reader.fetch_available + real_wait = reader.pull_subscription.fetch + pulls, waits = [], [] + + async def flaky_pull(*args, **kwargs): + pulls.append(1) + if len(pulls) in (1, 3): # fails, recovers, fails again, recovers + raise RuntimeError('fetch failed') + return await real_fetch_available(*args, **kwargs) + + async def wait_interrupted_once(*args, **kwargs): + waits.append(1) + if len(waits) == 1: + await asyncio.sleep(1.3) # silence, longer than the budget of one streak + raise RuntimeError('waiting interrupted') # sends the reader back to pull again + return await real_wait(*args, **kwargs) + + reader.fetch_available = flaky_pull + reader.pull_subscription.fetch = wait_interrupted_once + pub = get_publisher(unique_subject) + await pub.open() + try: + task = asyncio.create_task(reader.read_next()) + await asyncio.sleep(2.5) + assert not task.done(), 'the second failure was charged to the budget of the first one' + assert len(pulls) >= 4 + await pub.publish(data={'v': 1}) + data, meta = await asyncio.wait_for(task, timeout=5) + assert data == {'v': 1} + finally: + del reader.fetch_available + await reader.close() + await pub.close()