Deadlines, cancellation and the reader session, Python 3.11+ (2.5.1)

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

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

pyproject.toml

@@ -1,14 +1,14 @@
 [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"
 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}

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

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):

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:

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):

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:

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}")

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