From 76ab31fd7754138509adeb9c5ca725321709d570 Mon Sep 17 00:00:00 2001 From: majkelx Date: Sat, 14 Nov 2026 17:19:59 +0100 Subject: [PATCH 1/4] reader session (2.5.1) --- README.md | 7 + doc/messenger.md | 6 + pyproject.toml | 2 +- serverish/connection/nats_errors.py | 16 + serverish/messenger/msg_reader.py | 493 ++++++++++++++++------------ tests/test_delivery_policies.py | 2 +- tests/test_reader_session.py | 228 +++++++++++++ tests/test_resl_consumer_expiry.py | 4 +- 8 files changed, 548 insertions(+), 210 deletions(-) create mode 100644 tests/test_reader_session.py diff --git a/README.md b/README.md index f024896..5462cad 100644 --- a/README.md +++ b/README.md @@ -28,6 +28,13 @@ 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 diff --git a/doc/messenger.md b/doc/messenger.md index ecfc67e..b46cef7 100644 --- a/doc/messenger.md +++ b/doc/messenger.md @@ -105,6 +105,12 @@ 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) diff --git a/pyproject.toml b/pyproject.toml index 839851a..79e11a2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [tool.poetry] name = "serverish" -version = "2.5.0" +version = "2.5.1" description = "helpers for server alike projects" authors = ["Mikołaj Kałuszyński", "MMME team"] readme = "README.md" diff --git a/serverish/connection/nats_errors.py b/serverish/connection/nats_errors.py index 43d7016..a07a80f 100644 --- a/serverish/connection/nats_errors.py +++ b/serverish/connection/nats_errors.py @@ -84,6 +84,22 @@ 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)) + + 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_reader.py b/serverish/messenger/msg_reader.py index a9e2526..2030c00 100644 --- a/serverish/messenger/msg_reader.py +++ b/serverish/messenger/msg_reader.py @@ -19,9 +19,10 @@ 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_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 +36,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 @@ -163,7 +190,7 @@ def __init__(self, subject, parent = None, # --- 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() @@ -188,6 +215,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 @@ -241,6 +285,7 @@ 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) + 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()) @@ -250,6 +295,8 @@ def recovered(self) -> 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 def raise_if_cancelled(self) -> None: """Raises CancelledError when the calling task was cancelled during this read_next() @@ -261,8 +308,7 @@ def raise_if_cancelled(self) -> None: """ if asyncio.current_task().cancelling() > self.cancel_requests: raise asyncio.CancelledError - last_consumer_check: float = field(default_factory=time.monotonic) - consumer_check_interval: float = 10.0 # Check consumer health every 10 seconds + 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) @@ -352,51 +398,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 def ensure_session(self) -> None: + """Replaces the consumer session when it cannot serve the reader any more - @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') + 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 @@ -404,10 +432,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 self.reader._server_answered: - self.recovered() # the server answers: the failure is over, even with no data - - # 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 @@ -415,34 +458,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): + while not self.reader._reconnect_needed.is_set(): self.raise_if_cancelled() - # 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)")) + 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}")) @@ -463,29 +498,59 @@ 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.deadline = 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}")) @@ -513,9 +578,10 @@ 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. cause = e.error.__cause__ if isinstance(e.error, MessengerJetStreamError) else e.error - if isinstance(cause, (nats.js.errors.BadRequestError, - nats.js.errors.NotFoundError)): + if isinstance(cause, nats.js.errors.BadRequestError) or is_missing_stream(cause): log.error(st.fmt(f"fatal NATS error (not retrying): {e.error}")) self._stop.set() raise MessengerReaderConfigError( @@ -532,26 +598,16 @@ def logput(self, msg: str) -> None: log.error(st.fmt(f"finishing iteration on error: {e.error}")) raise MessengerReaderStopped case 'WAIT': - def give_up(): - 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 - elapsed = time.monotonic() - st.error_since if not self._retry.keeps_retrying(st.retries, elapsed): - give_up() + 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() # the deadline passed during the pause: no retry after it + 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!")) @@ -585,7 +641,8 @@ 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 @@ -656,6 +713,7 @@ def log_backstop(): 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": @@ -666,10 +724,19 @@ 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 closed the request for another reason (e.g. 409 on a limit + # of the consumer): there is nothing more to wait for in this request + log.debug(f"{self} Server closed the request: {status} {self._status_description(msg)}") + break # Real message msgs.append(msg) @@ -693,8 +760,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. @@ -759,37 +833,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 @@ -801,17 +876,89 @@ 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). + """ + new = await self._create_session(resume=True) + old, self._session = self._session, new + self._reconnect_count += 1 + if old is not None: + await self._discard_session(old) + + async def _discard_session(self, session: _ConsumerSession) -> 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`). + """ + stream, consumer = session.stream, session.consumer + try: + if consumer is None: # a subscription assigned from outside: ask who it is + ci = await session.subscription.consumer_info() + stream, consumer = ci.stream_name, ci.name + except Exception as e: + log.debug(f"{self} could not identify the consumer of a discarded session: {e}") + try: + await session.subscription.unsubscribe() + except Exception as e: + log.debug(f"{self} could not unsubscribe a discarded session: {e}") + if consumer is not None: + try: + await self.connection.js.delete_consumer(stream=stream, consumer=consumer) + except nats.js.errors.NotFoundError: + pass # gone already + except Exception as e: + log.debug(f"{self} could not delete consumer {consumer} of a discarded session " + f"(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 @@ -865,62 +1012,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() @@ -965,21 +1056,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() @@ -1075,9 +1156,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/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_reader_session.py b/tests/test_reader_session.py new file mode 100644 index 0000000..ba61cca --- /dev/null +++ b/tests/test_reader_session.py @@ -0,0 +1,228 @@ +"""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 nats.errors +import nats.js.errors +import pytest + +from serverish.base import MessengerReaderConfigError +from serverish.connection.nats_errors import is_missing_stream +from serverish.messenger import 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()) + + +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) + + +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) From 5a8383d7eb384900c24f115a82316ab32883d3f5 Mon Sep 17 00:00:00 2001 From: majkelx Date: Sat, 14 Nov 2026 18:10:09 +0100 Subject: [PATCH 2/4] review fixes --- serverish/messenger/msg_reader.py | 32 ++++++++++++--- tests/test_reader_session.py | 65 +++++++++++++++++++++++++++++++ 2 files changed, 91 insertions(+), 6 deletions(-) diff --git a/serverish/messenger/msg_reader.py b/serverish/messenger/msg_reader.py index 2030c00..520cdd6 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 @@ -54,6 +55,10 @@ class _ConsumerSession: 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 same_consumer(self, other: _ConsumerSession) -> bool: + """Tells whether both sessions use one consumer on the server (a durable one, bound twice)""" + return self.consumer is not None and (self.stream, self.consumer) == (other.stream, other.consumer) + 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: @@ -298,6 +303,15 @@ def recovered(self) -> None: 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() @@ -524,8 +538,7 @@ def recovery_deadline() -> float | None: st.raise_if_cancelled() failing = st.error try: - async with asyncio.timeout(recovery_deadline()) as deadline: - st.deadline = deadline + 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 @@ -657,6 +670,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 @@ -932,17 +948,21 @@ async def _replace_session(self) -> None: old, self._session = self._session, new self._reconnect_count += 1 if old is not None: - await self._discard_session(old) + # a durable consumer is bound again, not created: it is the new session's as well + await self._discard_session(old, keep_consumer=old.same_consumer(new)) - async def _discard_session(self, session: _ConsumerSession) -> None: + 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 - the consumer serves another session """ stream, consumer = session.stream, session.consumer try: - if consumer is None: # a subscription assigned from outside: ask who it is + if consumer is None and not keep_consumer: # a subscription assigned from outside: ask who it is ci = await session.subscription.consumer_info() stream, consumer = ci.stream_name, ci.name except Exception as e: @@ -951,7 +971,7 @@ async def _discard_session(self, session: _ConsumerSession) -> None: await session.subscription.unsubscribe() except Exception as e: log.debug(f"{self} could not unsubscribe a discarded session: {e}") - if consumer is not None: + if consumer is not None and not keep_consumer: try: await self.connection.js.delete_consumer(stream=stream, consumer=consumer) except nats.js.errors.NotFoundError: diff --git a/tests/test_reader_session.py b/tests/test_reader_session.py index ba61cca..8ff59c6 100644 --- a/tests/test_reader_session.py +++ b/tests/test_reader_session.py @@ -202,6 +202,71 @@ async def no_info(*a, **k): await r.publish_and_expect(1) +async def test_durable_consumer_survives_the_replacement_of_its_session(messenger, unique_subject): + """A durable consumer is bound by the new session, not created: removing the old session must not remove it.""" + durable = 'durable_' + unique_subject.rsplit('.', 1)[-1] + async with Reading(unique_subject, error_policy=FAST, durable_name=durable) as r: + old = r.reader._session + assert old.consumer == durable + await r.reader._replace_session() + assert r.reader._session is not old and r.reader._session.consumer == durable + await r.reader.connection.js.consumer_info(old.stream, durable) # raises if it was deleted + await r.publish_and_expect(1) + assert r.reader.health_status['reconnect_count'] == 1 # no second replacement was needed + + +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', From 5397c6be18670006919cbc29335f684d36dd39dd Mon Sep 17 00:00:00 2001 From: majkelx Date: Sat, 14 Nov 2026 18:59:15 +0100 Subject: [PATCH 3/4] review fixes 2 --- serverish/connection/nats_errors.py | 13 +++++ serverish/messenger/msg_reader.py | 89 +++++++++++++++++++---------- tests/test_reader_session.py | 87 ++++++++++++++++++++++++++-- 3 files changed, 153 insertions(+), 36 deletions(-) diff --git a/serverish/connection/nats_errors.py b/serverish/connection/nats_errors.py index a07a80f..cf558a8 100644 --- a/serverish/connection/nats_errors.py +++ b/serverish/connection/nats_errors.py @@ -100,6 +100,19 @@ def is_missing_stream(error: BaseException | None) -> bool: 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_reader.py b/serverish/messenger/msg_reader.py index 520cdd6..69d6671 100644 --- a/serverish/messenger/msg_reader.py +++ b/serverish/messenger/msg_reader.py @@ -23,7 +23,8 @@ MessengerReaderAlreadyOpen, MessengerRequestTimeout, MessengerTransportTimeout) from serverish.base.fifoset import FifoSet -from serverish.connection.nats_errors import is_missing_stream, 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, @@ -55,10 +56,6 @@ class _ConsumerSession: 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 same_consumer(self, other: _ConsumerSession) -> bool: - """Tells whether both sessions use one consumer on the server (a durable one, bound twice)""" - return self.consumer is not None and (self.stream, self.consumer) == (other.stream, other.consumer) - 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: @@ -593,8 +590,10 @@ def recovery_deadline() -> float | None: 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) or is_missing_stream(cause): + 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( @@ -749,10 +748,12 @@ def log_backstop(): elif status == "100": continue # a heartbeat, the answer is still to come elif status: - # The server closed the request for another reason (e.g. 409 on a limit - # of the consumer): there is nothing more to wait for in this request - log.debug(f"{self} Server closed the request: {status} {self._status_description(msg)}") - break + # 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) @@ -943,13 +944,50 @@ async def _replace_session(self) -> None: 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 durable consumer is its 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. """ - new = await self._create_session(resume=True) - old, self._session = self._session, new + old = self._session + if old is not None and self.consumer_cfg.get('durable_name'): + await self._delete_consumer(old) + old.gone = True # should the creation fail: this session has nothing to read from + 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._discard_session(old) + old, self._session = None, None + new = await self._create_session(resume=True) + self._session = new self._reconnect_count += 1 if old is not None: - # a durable consumer is bound again, not created: it is the new session's as well - await self._discard_session(old, keep_consumer=old.same_consumer(new)) + await self._discard_session(old, keep_consumer=old.gone) + + 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 @@ -958,27 +996,18 @@ async def _discard_session(self, session: _ConsumerSession, keep_consumer: bool expires on the server (`inactive_threshold`). Args: - keep_consumer: drop the subscription only - the consumer serves another session + keep_consumer: drop the subscription only - there is no consumer of this session to delete """ - stream, consumer = session.stream, session.consumer - try: - if consumer is None and not keep_consumer: # a subscription assigned from outside: ask who it is - ci = await session.subscription.consumer_info() - stream, consumer = ci.stream_name, ci.name - except Exception as e: - log.debug(f"{self} could not identify the consumer of a discarded session: {e}") try: await session.subscription.unsubscribe() except Exception as e: log.debug(f"{self} could not unsubscribe a discarded session: {e}") - if consumer is not None and not keep_consumer: - try: - await self.connection.js.delete_consumer(stream=stream, consumer=consumer) - except nats.js.errors.NotFoundError: - pass # gone already - except Exception as e: - log.debug(f"{self} could not delete consumer {consumer} of a discarded session " - f"(it will expire): {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 diff --git a/tests/test_reader_session.py b/tests/test_reader_session.py index 8ff59c6..68e637c 100644 --- a/tests/test_reader_session.py +++ b/tests/test_reader_session.py @@ -4,14 +4,16 @@ 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 MessengerReaderConfigError -from serverish.connection.nats_errors import is_missing_stream -from serverish.messenger import ErrorPolicy, OnError, RetryPolicy, get_publisher, get_reader +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)] @@ -83,6 +85,12 @@ def test_only_a_missing_stream_is_a_deployment_error(): 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'): @@ -202,19 +210,86 @@ async def no_info(*a, **k): await r.publish_and_expect(1) -async def test_durable_consumer_survives_the_replacement_of_its_session(messenger, unique_subject): - """A durable consumer is bound by the new session, not created: removing the old session must not remove it.""" +async def test_durable_consumer_is_created_anew_under_its_name(messenger, unique_subject): + """A durable consumer is its name: the new session has it, continuing after the last message read.""" durable = 'durable_' + unique_subject.rsplit('.', 1)[-1] async with Reading(unique_subject, error_policy=FAST, durable_name=durable) as r: old = r.reader._session assert old.consumer == durable await r.reader._replace_session() assert r.reader._session is not old and r.reader._session.consumer == durable - await r.reader.connection.js.consumer_info(old.stream, durable) # raises if it was deleted + info = await r.reader.connection.js.consumer_info(old.stream, durable) + assert info.config.opt_start_seq == r.reader.last_seq + 1 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_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) From f7097f7f0fe46c18689107a1493b4a4123ab3431 Mon Sep 17 00:00:00 2001 From: majkelx Date: Sat, 14 Nov 2026 19:29:13 +0100 Subject: [PATCH 4/4] review fixes 3 --- serverish/messenger/msg_reader.py | 23 +++++++++++++++++++---- tests/test_reader_session.py | 30 ++++++++++++++++++++---------- 2 files changed, 39 insertions(+), 14 deletions(-) diff --git a/serverish/messenger/msg_reader.py b/serverish/messenger/msg_reader.py index 69d6671..d6c4bc5 100644 --- a/serverish/messenger/msg_reader.py +++ b/serverish/messenger/msg_reader.py @@ -947,13 +947,14 @@ async def _replace_session(self) -> None: Two cases cannot have "the new one first", and give the old consumer up before: - - a durable consumer is its 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 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 self.consumer_cfg.get('durable_name'): + if old is not None and await self._has_fixed_name(old): await self._delete_consumer(old) old.gone = True # should the creation fail: this session has nothing to read from try: @@ -969,6 +970,20 @@ async def _replace_session(self) -> None: if old is not None: await self._discard_session(old, keep_consumer=old.gone) + 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 diff --git a/tests/test_reader_session.py b/tests/test_reader_session.py index 68e637c..8f2c9cd 100644 --- a/tests/test_reader_session.py +++ b/tests/test_reader_session.py @@ -210,18 +210,28 @@ async def no_info(*a, **k): await r.publish_and_expect(1) -async def test_durable_consumer_is_created_anew_under_its_name(messenger, unique_subject): - """A durable consumer is its name: the new session has it, continuing after the last message read.""" - durable = 'durable_' + unique_subject.rsplit('.', 1)[-1] - async with Reading(unique_subject, error_policy=FAST, durable_name=durable) as r: - old = r.reader._session - assert old.consumer == durable +@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() - assert r.reader._session is not old and r.reader._session.consumer == durable - info = await r.reader.connection.js.consumer_info(old.stream, durable) - assert info.config.opt_start_seq == r.reader.last_seq + 1 + 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 + 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):