Reader session: an open reader reads (2.5.1)

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

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)
 

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"

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

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