feat: Check the preconditions on a subject recursively, for the subject together with every subject below it.

README.md

@@ -46,7 +46,7 @@ await client.verify_api_token()
 
 ### Talking to Servers With and Without the v2 API
 
-The client SDK works with servers that offer the v2 API of EventSourcingDB, which it has since version 1.3, as well as with servers that only offer the v1 API. Creating a client does not send any request. Instead, on its first request, the client asks the server whether it offers the v2 API, and remembers the answer. If the server answers a path of the v2 API with 404 or 405, for example after it was rolled back to a version without it, the client asks again on the next request. If the server cannot be asked, for example because it is not reachable, the first request fails with that error, and the next request asks again.
+The client SDK works with servers that offer the v2 API of EventSourcingDB, which it has since version 1.3, as well as with servers that only offer the v1 API. Creating a client does not send any request. Instead, on its first request, the client asks the server whether it offers the v2 API, and remembers the answer. If the server answers a path of the v2 API with 404 or 405, for example after it was rolled back to a version without it, the client asks again on the next request. If the server cannot be asked, for example because it is not reachable, the first request fails with that error, and the next request asks again. A write with a recursive precondition, which needs the v2 API, asks again as well if the client remembers that the server does not offer it, while every other request keeps using that answer, see [Checking Preconditions on Subjects Recursively](#checking-preconditions-on-subjects-recursively).
 
 With the v2 API, every function that reads, such as `read_events`, `observe_events`, `run_eventql_query`, `read_subjects`, `read_event_type`, and `read_event_types`, uses the HTTP method `QUERY` ([RFC 10008](https://www.rfc-editor.org/rfc/rfc10008)). If there is a proxy, a load balancer, or a firewall in front of EventSourcingDB, it has to let requests with this method through. The client does not fall back to `POST`.
 
@@ -106,7 +106,7 @@ Servers without the v2 API report errors without a code. For them, the client as
 
 EventSourcingDB never stores a write halfway: the events of a call to `write_events`, or the schema of a call to `register_event_schema`, are either stored completely or not at all. If one of these functions fails in a way that leaves open whether the server stored the write, it raises an `OutcomeUnknownError`, whose `cause` attribute holds the error that the write failed with. This happens, for example, if the server stalls or the connection breaks once the request has a connection, if the server answers with a status such as `500`, `502`, or `504`, if the answer does not come from EventSourcingDB, or if the server ends its answer before it is complete. Before you repeat such a write, find out whether it was stored, for example by reading the events of the subject, or repeat it with a precondition such as `IsSubjectOnEventId` that only holds if it was not stored.
 
-The function raises the error itself, which tells that the write was certainly not stored, only if the request could not get a connection at all, such as when the name of the server cannot be resolved, connecting or TLS fails, or the request waits for a free connection for longer than the stall timeout, or if the server refused the write: with an error that it reports within its answer, other than `ErrorCode.STORE_FAILED`, with one of the statuses `400`, `401`, `408`, `413`, `429`, and `503`, or, for servers without the v2 API, with any status `4xx`, `503`, or `507`.
+The function raises the error itself, which tells that the write was certainly not stored, only if the client refused the write before sending it, such as with an `IncompatibleServerError`, if the client could not find out which API the server offers, which it asks with a ping before the write where needed (see [Talking to Servers With and Without the v2 API](#talking-to-servers-with-and-without-the-v2-api)), for example because the ping stalled, the server answered it with a status such as `500`, or the answer did not come from EventSourcingDB, if the request could not get a connection at all, such as when the name of the server cannot be resolved, connecting or TLS fails, or the request waits for a free connection for longer than the stall timeout, or if the server refused the write: with an error that it reports within its answer, other than `ErrorCode.STORE_FAILED`, with one of the statuses `400`, `401`, `408`, `413`, `429`, and `503`, or, for servers without the v2 API, with any status `4xx`, `503`, or `507`.
 
 Cancelling a write once it has a connection, for example with `asyncio.timeout` or by cancelling its task, leaves its outcome open as well. It is cancelled as usual nonetheless, so that `asyncio.timeout`, `asyncio.wait_for`, and task groups work as always: the `asyncio.CancelledError` is not wrapped, but marked, and carries a note that says so. To tell whether the outcome of a write is unknown, call `is_outcome_unknown` with the error. It returns `True` for an `OutcomeUnknownError`, for a marked `CancelledError`, and for an error caused by one of them, such as the `TimeoutError` of `asyncio.timeout`:
 
@@ -228,6 +228,33 @@ written_events = await client.write_events(
 
 *Note that according to the CloudEvents standard, event IDs must be of type string.*
 
+#### Checking Preconditions on Subjects Recursively
+
+The `IsSubjectPristine`, `IsSubjectPopulated`, and `IsSubjectOnEventId` preconditions check the given subject alone. If you want them to check the subject together with all nested subjects, import the `PreconditionOptions` class and pass it with the `recursive` option set to `True`:
+
+```python
+from eventsourcingdb import IsSubjectOnEventId, PreconditionOptions
+
+written_events = await client.write_events(
+  event_candidates = [
+    # event candidates
+  ],
+  preconditions = [
+    IsSubjectOnEventId(
+      subject = '/books/42',
+      event_id = '0',
+      options = PreconditionOptions(
+        recursive = True
+      ),
+    )
+  ],
+)
+```
+
+Then `IsSubjectPristine` holds if neither the subject nor any of its nested subjects has an event, `IsSubjectPopulated` holds if any of them has one, and `IsSubjectOnEventId` holds if the latest event among all of them has the given ID. A recursive precondition that does not hold is reported like any other precondition that does not hold.
+
+Checking a precondition recursively needs the v2 API. If the client remembers that the server does not offer it, it asks the server again before the write, since the server may offer it by now, for example after an upgrade. If asking again fails, the write fails with that error, and the client keeps its answer. If the server still does not offer it, `write_events` raises an `IncompatibleServerError` before it sends the events, so none of them is written. `IncompatibleServerError` is a subclass of `CustomError`. If you set the `recursive` option to `False`, the precondition works with every server, just as without options.
+
 #### Using the `isEventQlQueryTrue` precondition
 
 If you want to write events depending on an EventQL query, import the `IsEventQlQueryTrue` class and pass it as a list of preconditions in the second argument:

eventsourcingdb/__init__.py

@@ -8,6 +8,7 @@
     DbApiError,
     ErrorCode,
     HashVerificationFailedError,
+    IncompatibleServerError,
     InternalError,
     OutcomeUnknownError,
     ServerError,
@@ -36,6 +37,7 @@
     IsSubjectPopulated,
     IsSubjectPristine,
     Precondition,
+    PreconditionOptions,
 )
 
 __all__ = [
@@ -54,6 +56,7 @@
     "HashVerificationFailedError",
     "IfEventIsMissingDuringObserve",
     "IfEventIsMissingDuringRead",
+    "IncompatibleServerError",
     "InternalError",
     "IsEventQlQueryTrue",
     "IsSubjectOnEventId",
@@ -64,6 +67,7 @@
     "Order",
     "OutcomeUnknownError",
     "Precondition",
+    "PreconditionOptions",
     "ReadEventsOptions",
     "ReadFromLatestEvent",
     "ReadSubjectsOptions",

eventsourcingdb/client.py

@@ -12,6 +12,7 @@
 from .read_subjects import ReadSubjectsOptions
 from .transport import ChooseTransport
 from .write_events import Precondition
+from .write_events.preconditions import is_recursive
 
 T = TypeVar('T')
 
@@ -78,7 +79,14 @@ async def write_events(
         if preconditions is None:
             preconditions = []
 
-        transport = await self.__transports.get()
+        # Only the v2 API checks a recursive precondition, so the client asks
+        # again whether the server offers it rather than refuse the write on
+        # an answer that may be outdated.
+        if any(is_recursive(precondition) for precondition in preconditions):
+            transport = await self.__transports.get_asking_again_without_v2_api()
+        else:
+            transport = await self.__transports.get()
+
         return await transport.write_events(event_candidates, preconditions)
 
     async def read_events(

eventsourcingdb/errors/__init__.py

@@ -4,6 +4,7 @@
 from .db_api_error import DbApiError
 from .error_code import ErrorCode
 from .hash_verification_failed_error import HashVerificationFailedError
+from .incompatible_server_error import IncompatibleServerError
 from .internal_error import InternalError
 from .is_outcome_unknown import is_outcome_unknown
 from .outcome_unknown_error import OutcomeUnknownError
@@ -19,6 +20,7 @@
     "DbApiError",
     "ErrorCode",
     "HashVerificationFailedError",
+    "IncompatibleServerError",
     "InternalError",
     "OutcomeUnknownError",
     "ServerError",

eventsourcingdb/errors/incompatible_server_error.py

@@ -0,0 +1,15 @@
+from .custom_error import CustomError
+
+
+class IncompatibleServerError(CustomError):
+    """
+    A write has a recursive precondition, which only the v2 API offers, but
+    the server does not offer it, as the client found out right before the
+    write rather than from an earlier answer. The client raises this error
+    before it sends the write, so nothing is written.
+    """
+
+    def __init__(self) -> None:
+        super().__init__(
+            "The server does not offer the v2 API, which a recursive precondition needs."
+        )

eventsourcingdb/transport/choose_transport.py

@@ -24,14 +24,20 @@ class ChooseTransport:
     that need the answer at the same time share a single ping. Once the server
     answers a path of the v2 API with 404 or 405, for example after it was
     rolled back to a version without it, the answer is forgotten, and the
-    next request asks again.
+    next request asks again. A write with a recursive precondition, which
+    needs the v2 API, asks again if the answer is that the server does not
+    offer it, since it may offer it by now, for example after an upgrade.
+    Writes that ask again at the same time share a single ping, while every
+    other request keeps using the answer, which only a definite answer of
+    that ping replaces.
     """
 
     def __init__(self, http_client: HttpClient) -> None:
         self.__http_client = http_client
         self.__v1_transport = V1Transport(http_client)
         self.__transport: Transport | None = None
         self.__probe: asyncio.Task[Transport] | None = None
+        self.__reask: asyncio.Task[Transport] | None = None
 
     async def get(self) -> Transport:
         if self.__transport is not None:
@@ -56,9 +62,34 @@ async def get(self) -> Transport:
         await asyncio.wait({probe})
         return probe.result()
 
+    async def get_asking_again_without_v2_api(self) -> Transport:
+        # Without an answer yet, the ping of get finds a fresh one, and with
+        # the answer that the server offers the v2 API, there is nothing to
+        # ask again about.
+        if self.__transport is not self.__v1_transport:
+            return await self.get()
+
+        # The answer stays remembered while the ping asks again, so that every
+        # other request keeps using it, and nothing else replaces it meanwhile,
+        # since only the answer that the server offers the v2 API is ever
+        # forgotten. Unlike the ping of get, this ping is shared only while it
+        # runs, so that every write that needs it gets an answer that is newer
+        # than the one it asks again about.
+        reask = self.__reask
+        if reask is None or reask.get_loop() is not asyncio.get_running_loop() or reask.done():
+            reask = asyncio.create_task(self.__detect())
+            reask.add_done_callback(self.__remember)
+            self.__reask = reask
+
+        # Waiting rather than shielding, for the same reasons as in get.
+        await asyncio.wait({reask})
+        return reask.result()
+
     def __remember(self, probe: asyncio.Task[Transport]) -> None:
         if self.__probe is probe:
             self.__probe = None
+        if self.__reask is probe:
+            self.__reask = None
 
         if ChooseTransport.__succeeded(probe):
             self.__transport = probe.result()

eventsourcingdb/transport/guard_write.py

@@ -20,7 +20,10 @@ async def guard_write(
     tells a failure that leaves open whether the server stored the write by
     raising an OutcomeUnknownError with the failure as its cause.
 
-    The write was certainly not stored only if the request never got a
+    Apart from a write that fails before it calls this function, such as one
+    with a recursive precondition for a server without the v2 API, or one for
+    which the client could not find out which API the server offers, the
+    write was certainly not stored only if the request never got a
     connection, such as when the name of the server could not be resolved,
     connecting or TLS failed, or the request waited for a free connection in
     vain, or if the server explicitly refused the write, which the given

eventsourcingdb/transport/v1/transport.py

@@ -1,12 +1,14 @@
 import json
 from collections import OrderedDict
 from collections.abc import AsyncGenerator
+from dataclasses import replace
 from http import HTTPStatus
 from typing import Any
 
 from ...errors import (
     CustomError,
     DbApiError,
+    IncompatibleServerError,
     InternalError,
     ServerError,
     ValidationError,
@@ -23,6 +25,7 @@
 from ...read_event_types import EventType
 from ...read_events import ReadEventsOptions
 from ...write_events import Precondition
+from ...write_events.preconditions import PRECONDITIONS_ON_A_SUBJECT, is_recursive
 from ..error_of_answer import error_of_answer
 from ..guard_write import guard_write
 from ..stream_broke_off import stream_broke_off
@@ -73,6 +76,22 @@ def is_refusal(error: DbApiError) -> bool:
     }
 
 
+def precondition_of_v1_api(precondition: Precondition) -> Any:
+    """
+    Turns a precondition into what the v1 API expects. The v1 API knows no
+    options of a precondition and rejects them, so a precondition whose
+    options are not recursive is sent as one without options. It cannot check
+    a recursive precondition, which fails the write before it is sent, rather
+    than have the server check less than was asked.
+    """
+    if is_recursive(precondition):
+        raise IncompatibleServerError()
+    if isinstance(precondition, PRECONDITIONS_ON_A_SUBJECT):
+        return replace(precondition, options=None).to_json()
+
+    return precondition.to_json()
+
+
 def error_of_stream_item(message: dict) -> DbApiError:
     """
     Turns an error item of a stream into the error that the server reports
@@ -140,7 +159,9 @@ async def write_events(
         request_body = json.dumps(
             {
                 'events': [event_candidate.to_json() for event_candidate in event_candidates],
-                'preconditions': [precondition.to_json() for precondition in preconditions]
+                'preconditions': [
+                    precondition_of_v1_api(precondition) for precondition in preconditions
+                ],
             }
         )
 

eventsourcingdb/write_events/__init__.py

@@ -1,3 +1,4 @@
+from .precondition_options import PreconditionOptions
 from .preconditions import (
     IsEventQlQueryTrue,
     IsSubjectOnEventId,
@@ -12,4 +13,5 @@
     "IsSubjectPopulated",
     "IsSubjectPristine",
     "Precondition",
+    "PreconditionOptions",
 ]

eventsourcingdb/write_events/precondition_options.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass
+class PreconditionOptions:
+    recursive: bool
+
+    def to_json(self) -> dict[str, Any]:
+        return {"recursive": self.recursive}

eventsourcingdb/write_events/preconditions.py

@@ -2,40 +2,75 @@
 from dataclasses import dataclass
 from typing import Any
 
+from .precondition_options import PreconditionOptions
+
 
 class Precondition(ABC):
     @abstractmethod
     def to_json(self) -> Any:
         ...
 
 
+def with_options(payload: dict[str, Any], options: PreconditionOptions | None) -> dict[str, Any]:
+    # Options that were not given are left out rather than sent as null, which
+    # the v2 API rejects.
+    if options is None:
+        return payload
+
+    return {**payload, "options": options.to_json()}
+
+
 @dataclass
 class IsSubjectPristine(Precondition):
     subject: str
+    options: PreconditionOptions | None = None
 
     def to_json(self) -> dict[str, Any]:
-        return {"type": "isSubjectPristine", "payload": {"subject": self.subject}}
+        return {
+            "type": "isSubjectPristine",
+            "payload": with_options({"subject": self.subject}, self.options),
+        }
 
 
 @dataclass
 class IsSubjectPopulated(Precondition):
     subject: str
+    options: PreconditionOptions | None = None
 
     def to_json(self) -> dict[str, Any]:
-        return {"type": "isSubjectPopulated", "payload": {"subject": self.subject}}
+        return {
+            "type": "isSubjectPopulated",
+            "payload": with_options({"subject": self.subject}, self.options),
+        }
 
 
 @dataclass
 class IsSubjectOnEventId(Precondition):
     subject: str
     event_id: str
+    options: PreconditionOptions | None = None
 
     def to_json(self) -> dict[str, Any]:
         return {
             "type": "isSubjectOnEventId",
-            "payload": {"subject": self.subject, "eventId": self.event_id},
+            "payload": with_options(
+                {"subject": self.subject, "eventId": self.event_id}, self.options
+            ),
         }
 
+
+# The preconditions on a subject, which take options.
+PRECONDITIONS_ON_A_SUBJECT = (IsSubjectPristine, IsSubjectPopulated, IsSubjectOnEventId)
+
+
+def is_recursive(precondition: Precondition) -> bool:
+    return (
+        isinstance(precondition, PRECONDITIONS_ON_A_SUBJECT)
+        and precondition.options is not None
+        and bool(precondition.options.recursive)
+    )
+
+
 @dataclass
 class IsEventQlQueryTrue(Precondition):
     query: str