diff --git a/README.md b/README.md index 20951db..96f875f 100644 --- a/README.md +++ b/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: diff --git a/eventsourcingdb/__init__.py b/eventsourcingdb/__init__.py index bdafcb6..23f8960 100644 --- a/eventsourcingdb/__init__.py +++ b/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", diff --git a/eventsourcingdb/client.py b/eventsourcingdb/client.py index dedea55..bc5cc5a 100644 --- a/eventsourcingdb/client.py +++ b/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( diff --git a/eventsourcingdb/errors/__init__.py b/eventsourcingdb/errors/__init__.py index 6402b1e..61d1fab 100644 --- a/eventsourcingdb/errors/__init__.py +++ b/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", diff --git a/eventsourcingdb/errors/incompatible_server_error.py b/eventsourcingdb/errors/incompatible_server_error.py new file mode 100644 index 0000000..40fc96b --- /dev/null +++ b/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." + ) diff --git a/eventsourcingdb/transport/choose_transport.py b/eventsourcingdb/transport/choose_transport.py index afa5b5a..4ebceea 100644 --- a/eventsourcingdb/transport/choose_transport.py +++ b/eventsourcingdb/transport/choose_transport.py @@ -24,7 +24,12 @@ 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: @@ -32,6 +37,7 @@ def __init__(self, http_client: HttpClient) -> None: 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() diff --git a/eventsourcingdb/transport/guard_write.py b/eventsourcingdb/transport/guard_write.py index 5c4a4ea..d51810e 100644 --- a/eventsourcingdb/transport/guard_write.py +++ b/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 diff --git a/eventsourcingdb/transport/v1/transport.py b/eventsourcingdb/transport/v1/transport.py index 281c471..a154e02 100644 --- a/eventsourcingdb/transport/v1/transport.py +++ b/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 + ], } ) diff --git a/eventsourcingdb/write_events/__init__.py b/eventsourcingdb/write_events/__init__.py index 2e5949a..ce2a7d2 100644 --- a/eventsourcingdb/write_events/__init__.py +++ b/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", ] diff --git a/eventsourcingdb/write_events/precondition_options.py b/eventsourcingdb/write_events/precondition_options.py new file mode 100644 index 0000000..237e889 --- /dev/null +++ b/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} diff --git a/eventsourcingdb/write_events/preconditions.py b/eventsourcingdb/write_events/preconditions.py index 18ee390..e967b75 100644 --- a/eventsourcingdb/write_events/preconditions.py +++ b/eventsourcingdb/write_events/preconditions.py @@ -2,6 +2,8 @@ from dataclasses import dataclass from typing import Any +from .precondition_options import PreconditionOptions + class Precondition(ABC): @abstractmethod @@ -9,33 +11,66 @@ 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 diff --git a/tests/test_api_detection.py b/tests/test_api_detection.py index 1c8b52d..c8e9be2 100644 --- a/tests/test_api_detection.py +++ b/tests/test_api_detection.py @@ -13,12 +13,18 @@ CustomError, DbApiError, ErrorCode, + EventCandidate, + IncompatibleServerError, + IsSubjectPristine, OutcomeUnknownError, + Precondition, + PreconditionOptions, ReadEventsOptions, ReadSubjectsOptions, ServerError, ) +from .event.fixtures import WRITTEN_EVENTS_FROM_V2_API, WRITTEN_EVENTS_WITH_HASH_V2 from .shared.database import Database from .shared.requests import ALL_REQUESTS, WRITES, Request, read_event_type, read_events from .shared.stream_server import ( @@ -103,6 +109,30 @@ async def answer_read_events(request: web.Request) -> web.StreamResponse: ) +async def answer_writes_and_read_events(request: web.Request) -> web.StreamResponse: + # Writing events stores two, and reading events finds none, through either + # API. + if request.path == '/api/v2/data/write-events': + return web.Response(text=WRITTEN_EVENTS_FROM_V2_API, headers=SERVER_HEADER) + if request.path == '/api/v1/write-events': + return web.Response(text=WRITTEN_EVENTS_WITH_HASH_V2, headers=SERVER_HEADER) + return await answer_read_events(request) + + +async def write_events_with(client: Client, precondition: Precondition) -> None: + await client.write_events( + [ + EventCandidate( + source='https://www.eventsourcingdb.io', + subject='/books/42', + type='io.eventsourcingdb.library.book-acquired', + data={}, + ) + ], + [precondition], + ) + + def answer_without_v2_api( status: int = 404, headers: Mapping[str, str] = SERVER_HEADER ) -> web.StreamResponse: @@ -727,6 +757,197 @@ async def test_keeps_the_v2_api_that_a_newer_ping_found_if_an_earlier_request_fa assert apis_of(server) == ['v2', 'v2', 'v2', 'v2', 'v2'] +class TestApiDetectionForARecursivePrecondition: + # Only the v2 API checks a recursive precondition, and a server that did + # not offer it may offer it by now, for example after an upgrade. + RECURSIVE = IsSubjectPristine('/books', PreconditionOptions(recursive=True)) + + @staticmethod + @pytest.mark.asyncio + async def test_asks_again_and_writes_through_the_v2_api_once_the_server_offers_it() -> None: + async with StreamServer( + None, + answer_probe=answer_probes_in_turn( + answer_probe_without_v2_api, answer_probe_with_v2_api + ), + answer=answer_writes_and_read_events, + ) as server: + client = server.get_client() + await read_events(client) + + await write_events_with(client, TestApiDetectionForARecursivePrecondition.RECURSIVE) + + assert server.probe_count == 2 + assert [(request.method, request.path) for request in server.requests] == [ + ('POST', '/api/v1/read-events'), + ('POST', '/api/v2/data/write-events'), + ] + assert json.loads(server.requests[-1].body)['preconditions'] == [ + { + 'type': 'isSubjectPristine', + 'payload': {'subject': '/books', 'options': {'recursive': True}}, + } + ] + + await read_events(client) + + assert server.probe_count == 2 + assert apis_of(server) == ['v1', 'v2', 'v2'] + + @staticmethod + @pytest.mark.asyncio + async def test_asks_again_and_refuses_the_write_if_the_server_still_lacks_the_v2_api() -> None: + async with StreamServer(None, answer=answer_writes_and_read_events) as server: + client = server.get_client() + await read_events(client) + + with pytest.raises(IncompatibleServerError): + await write_events_with(client, TestApiDetectionForARecursivePrecondition.RECURSIVE) + + assert server.probe_count == 2 + assert apis_of(server) == ['v1'] + + await read_events(client) + + assert server.probe_count == 2 + assert apis_of(server) == ['v1', 'v1'] + + @staticmethod + @pytest.mark.asyncio + @pytest.mark.parametrize( + 'precondition', + [ + pytest.param(IsSubjectPristine('/books'), id='without options'), + pytest.param( + IsSubjectPristine('/books', PreconditionOptions(recursive=False)), + id='not recursive', + ), + ], + ) + async def test_keeps_the_answer_for_a_write_without_a_recursive_precondition( + precondition: Precondition, + ) -> None: + async with StreamServer( + None, + answer_probe=answer_probes_in_turn( + answer_probe_without_v2_api, answer_probe_with_v2_api + ), + answer=answer_writes_and_read_events, + ) as server: + client = server.get_client() + await read_events(client) + + await write_events_with(client, precondition) + + assert server.probe_count == 1 + assert apis_of(server) == ['v1', 'v1'] + + @staticmethod + @pytest.mark.asyncio + async def test_lets_every_other_request_use_the_answer_while_it_asks_again() -> None: + probe_arrived = asyncio.Event() + release_probe = asyncio.Event() + + async def answer_probe_once_released(request: web.Request) -> web.StreamResponse: + probe_arrived.set() + await release_probe.wait() + return await answer_probe_with_v2_api(request) + + async with StreamServer( + None, + answer_probe=answer_probes_in_turn( + answer_probe_without_v2_api, answer_probe_once_released + ), + answer=answer_writes_and_read_events, + ) as server: + client = server.get_client() + await read_events(client) + + write = asyncio.create_task( + write_events_with(client, TestApiDetectionForARecursivePrecondition.RECURSIVE) + ) + async with asyncio.timeout(GUARD_TIMEOUT_SECONDS): + await probe_arrived.wait() + + read = asyncio.create_task(read_events(client)) + done, _ = await asyncio.wait({read}, timeout=GUARD_TIMEOUT_SECONDS) + did_read_while_asking = read in done + + release_probe.set() + async with asyncio.timeout(GUARD_TIMEOUT_SECONDS): + await asyncio.gather(read, write) + + assert did_read_while_asking, 'The read waited for the ping that asks again.' + assert server.probe_count == 2 + assert apis_of(server) == ['v1', 'v1', 'v2'] + + @staticmethod + @pytest.mark.asyncio + async def test_keeps_the_answer_if_asking_again_fails() -> None: + async with StreamServer( + None, + answer_probe=answer_probes_in_turn( + answer_probe_without_v2_api, + answer_probe_with_status(500, 'internal server error'), + answer_probe_without_v2_api, + ), + answer=answer_writes_and_read_events, + ) as server: + client = server.get_client() + await read_events(client) + + with pytest.raises(DbApiError) as error: + await write_events_with(client, TestApiDetectionForARecursivePrecondition.RECURSIVE) + + assert error.value.status == 500 + assert server.probe_count == 2 + assert apis_of(server) == ['v1'] + + await read_events(client) + + assert server.probe_count == 2 + assert apis_of(server) == ['v1', 'v1'] + + @staticmethod + @pytest.mark.asyncio + async def test_shares_the_ping_that_asks_again_between_concurrent_writes() -> None: + probe_arrived = asyncio.Event() + release_probe = asyncio.Event() + + async def answer_probe_once_released(request: web.Request) -> web.StreamResponse: + probe_arrived.set() + await release_probe.wait() + return await answer_probe_with_v2_api(request) + + async with StreamServer( + None, + answer_probe=answer_probes_in_turn( + answer_probe_without_v2_api, answer_probe_once_released + ), + answer=answer_writes_and_read_events, + ) as server: + client = server.get_client() + await read_events(client) + + writes = [ + asyncio.create_task( + write_events_with(client, TestApiDetectionForARecursivePrecondition.RECURSIVE) + ) + for _ in range(3) + ] + + async with asyncio.timeout(GUARD_TIMEOUT_SECONDS): + await probe_arrived.wait() + release_probe.set() + await asyncio.gather(*writes) + + assert server.probe_count == 2 + assert [request.path for request in server.requests] == [ + '/api/v1/read-events', + *['/api/v2/data/write-events'] * 3, + ] + + class TestApiDetectionAgainstEventSourcingDB: @staticmethod @pytest.mark.asyncio diff --git a/tests/test_v2_api.py b/tests/test_v2_api.py index c9e586a..b7a54d6 100644 --- a/tests/test_v2_api.py +++ b/tests/test_v2_api.py @@ -18,10 +18,13 @@ IfEventIsMissingDuringObserve, IfEventIsMissingDuringRead, IsSubjectOnEventId, + IsSubjectPopulated, + IsSubjectPristine, ObserveEventsOptions, ObserveFromLatestEvent, Order, OutcomeUnknownError, + PreconditionOptions, ReadEventsOptions, ReadFromLatestEvent, ReadSubjectsOptions, @@ -198,6 +201,55 @@ async def test_sends_the_preconditions_of_a_write() -> None: {'type': 'isSubjectOnEventId', 'payload': {'subject': '/books/42', 'eventId': '0'}} ] + @staticmethod + @pytest.mark.asyncio + @pytest.mark.parametrize( + ('options', 'sent_options'), + [ + pytest.param(None, {}, id='without options'), + pytest.param( + PreconditionOptions(recursive=False), + {'options': {'recursive': False}}, + id='not recursive', + ), + pytest.param( + PreconditionOptions(recursive=True), + {'options': {'recursive': True}}, + id='recursive', + ), + ], + ) + async def test_sends_the_options_of_the_preconditions_on_a_subject_only_if_given( + options: PreconditionOptions | None, sent_options: dict[str, Any] + ) -> None: + async with StreamServer( + answer_with_lines(WRITTEN_EVENTS_FROM_V2_API.strip()), api='v2' + ) as server: + await server.get_client().write_events( + [ + EventCandidate( + source='https://www.eventsourcingdb.io', + subject='/books/42', + type='io.eventsourcingdb.library.book-acquired', + data={}, + ), + ], + [ + IsSubjectPristine('/books', options), + IsSubjectPopulated('/books', options), + IsSubjectOnEventId('/books', '0', options), + ], + ) + + assert server.last_request_json()['preconditions'] == [ # type: ignore[index] + {'type': 'isSubjectPristine', 'payload': {'subject': '/books', **sent_options}}, + {'type': 'isSubjectPopulated', 'payload': {'subject': '/books', **sent_options}}, + { + 'type': 'isSubjectOnEventId', + 'payload': {'subject': '/books', 'eventId': '0', **sent_options}, + }, + ] + @staticmethod @pytest.mark.asyncio async def test_registers_an_event_schema_with_post() -> None: diff --git a/tests/test_write_events.py b/tests/test_write_events.py index b3b1326..77f93f2 100644 --- a/tests/test_write_events.py +++ b/tests/test_write_events.py @@ -1,23 +1,91 @@ import asyncio +import json import pytest from aiohttp import ClientConnectorDNSError, web from eventsourcingdb import ( + Client, ConnectionStalledError, + CustomError, + DbApiError, + ErrorCode, EventCandidate, + IncompatibleServerError, IsEventQlQueryTrue, IsSubjectOnEventId, IsSubjectPopulated, IsSubjectPristine, OutcomeUnknownError, + Precondition, + PreconditionOptions, + ReadEventsOptions, ServerError, ) from .conftest import STALL_TIMEOUT_SECONDS, TEST_DEADLINE_SECONDS, TestData +from .event.fixtures import WRITTEN_EVENTS_WITH_HASH_V2 from .shared.database import Database from .shared.stream_server import StreamServer +RECURSIVE = PreconditionOptions(recursive=True) +NOT_RECURSIVE = PreconditionOptions(recursive=False) + +# Events 0 and 1 are written to /books/42 and to /books/42/chapters/1, so that +# /books has no event of its own but events below it, the latest event of +# /books/42 is not the latest one below it, and /authors has no event at all. +SUBJECTS_OF_WRITTEN_EVENTS = ["/books/42", "/books/42/chapters/1"] +IDS_OF_WRITTEN_EVENTS = ["0", "1"] + +RECURSIVE_PRECONDITIONS_THAT_HOLD = [ + pytest.param(IsSubjectPristine("/authors", RECURSIVE), id="isSubjectPristine"), + pytest.param(IsSubjectPopulated("/books", RECURSIVE), id="isSubjectPopulated"), + pytest.param(IsSubjectOnEventId("/books/42", "1", RECURSIVE), id="isSubjectOnEventId"), +] +RECURSIVE_PRECONDITIONS_THAT_ARE_VIOLATED = [ + pytest.param(IsSubjectPristine("/books", RECURSIVE), id="isSubjectPristine"), + pytest.param(IsSubjectPopulated("/authors", RECURSIVE), id="isSubjectPopulated"), + pytest.param(IsSubjectOnEventId("/books/42", "0", RECURSIVE), id="isSubjectOnEventId"), +] +PRECONDITIONS_THAT_ARE_NOT_RECURSIVE_AND_HOLD = [ + pytest.param(IsSubjectPristine("/books", NOT_RECURSIVE), id="isSubjectPristine"), + pytest.param(IsSubjectPopulated("/books/42", NOT_RECURSIVE), id="isSubjectPopulated"), + pytest.param(IsSubjectOnEventId("/books/42", "0", NOT_RECURSIVE), id="isSubjectOnEventId"), +] +PRECONDITIONS_THAT_ARE_NOT_RECURSIVE_AND_VIOLATED = [ + pytest.param(IsSubjectPristine("/books/42", NOT_RECURSIVE), id="isSubjectPristine"), + pytest.param(IsSubjectPopulated("/books", NOT_RECURSIVE), id="isSubjectPopulated"), + pytest.param(IsSubjectOnEventId("/books/42", "1", NOT_RECURSIVE), id="isSubjectOnEventId"), +] + + +def event_candidate(test_data: TestData, subject: str) -> EventCandidate: + return EventCandidate( + source=test_data.TEST_SOURCE_STRING, + subject=subject, + type="com.foo.bar", + data={}, + ) + + +async def write_events_below_books(client: Client, test_data: TestData) -> None: + await client.write_events( + [event_candidate(test_data, subject) for subject in SUBJECTS_OF_WRITTEN_EVENTS] + ) + + +async def write_event_with( + client: Client, test_data: TestData, precondition: Precondition +) -> None: + await client.write_events([event_candidate(test_data, "/books/42")], [precondition]) + + +async def read_ids_of_all_events(client: Client) -> list[str]: + return [ + event.event_id + async for event in client.read_events("/", ReadEventsOptions(recursive=True)) + ] + class TestWriteSubjects: @staticmethod @@ -357,3 +425,136 @@ async def write_stream(_: web.StreamResponse) -> None: ) # Since the server may have stored the events, their outcome is unknown. assert isinstance(error.value.cause, ConnectionStalledError) + + +class TestPreconditionsOnASubject: + @staticmethod + @pytest.mark.asyncio + @pytest.mark.parametrize("precondition", RECURSIVE_PRECONDITIONS_THAT_HOLD) + async def test_writes_if_a_recursive_precondition_holds_for_the_subject_and_below_it( + database: Database, + test_data: TestData, + precondition: Precondition, + ) -> None: + client = database.get_client() + await write_events_below_books(client, test_data) + + if await database.offers_v2_api(): + await write_event_with(client, test_data, precondition) + + assert await read_ids_of_all_events(client) == [*IDS_OF_WRITTEN_EVENTS, "2"] + else: + # Without the v2 API, the client refuses the write instead. + with pytest.raises(IncompatibleServerError): + await write_event_with(client, test_data, precondition) + + assert await read_ids_of_all_events(client) == IDS_OF_WRITTEN_EVENTS + + @staticmethod + @pytest.mark.asyncio + @pytest.mark.parametrize("precondition", RECURSIVE_PRECONDITIONS_THAT_ARE_VIOLATED) + async def test_refuses_to_write_if_a_recursive_precondition_is_violated( + database: Database, + test_data: TestData, + precondition: Precondition, + ) -> None: + client = database.get_client() + await write_events_below_books(client, test_data) + + if await database.offers_v2_api(): + with pytest.raises(DbApiError) as error: + await write_event_with(client, test_data, precondition) + + assert error.value.code == ErrorCode.PRECONDITION_VIOLATED + else: + # Without the v2 API, the client refuses the write instead. + with pytest.raises(IncompatibleServerError): + await write_event_with(client, test_data, precondition) + + assert await read_ids_of_all_events(client) == IDS_OF_WRITTEN_EVENTS + + @staticmethod + @pytest.mark.asyncio + @pytest.mark.parametrize("precondition", PRECONDITIONS_THAT_ARE_NOT_RECURSIVE_AND_HOLD) + async def test_writes_if_a_precondition_that_is_not_recursive_holds_for_the_subject_alone( + database: Database, + test_data: TestData, + precondition: Precondition, + ) -> None: + client = database.get_client() + await write_events_below_books(client, test_data) + + await write_event_with(client, test_data, precondition) + + assert await read_ids_of_all_events(client) == [*IDS_OF_WRITTEN_EVENTS, "2"] + + @staticmethod + @pytest.mark.asyncio + @pytest.mark.parametrize("precondition", PRECONDITIONS_THAT_ARE_NOT_RECURSIVE_AND_VIOLATED) + async def test_refuses_to_write_if_a_precondition_that_is_not_recursive_is_violated( + database: Database, + test_data: TestData, + precondition: Precondition, + ) -> None: + client = database.get_client() + await write_events_below_books(client, test_data) + + with pytest.raises(DbApiError) as error: + await write_event_with(client, test_data, precondition) + + assert error.value.code == ErrorCode.PRECONDITION_VIOLATED + assert await read_ids_of_all_events(client) == IDS_OF_WRITTEN_EVENTS + + +class TestPreconditionsOnASubjectOnTheV1Api: + @staticmethod + @pytest.mark.asyncio + @pytest.mark.parametrize("precondition", RECURSIVE_PRECONDITIONS_THAT_HOLD) + async def test_fails_without_sending_the_write_if_a_precondition_is_recursive( + test_data: TestData, + precondition: Precondition, + ) -> None: + async def write_stream(response: web.StreamResponse) -> None: + await response.write(WRITTEN_EVENTS_WITH_HASH_V2.encode("utf-8")) + + async with StreamServer(write_stream) as server: + with pytest.raises(IncompatibleServerError) as error: + await server.get_client().write_events( + [event_candidate(test_data, "/books/42")], + [IsSubjectPristine("/books/23"), precondition], + ) + + # The answer of the first request is fresh, so it is not asked again. + assert server.probe_count == 1 + assert server.requests == [] + + assert isinstance(error.value, CustomError) + assert str(error.value) == ( + "The server does not offer the v2 API, which a recursive precondition needs." + ) + + @staticmethod + @pytest.mark.asyncio + async def test_sends_preconditions_whose_options_are_not_recursive_as_without_options( + test_data: TestData, + ) -> None: + async def write_stream(response: web.StreamResponse) -> None: + await response.write(WRITTEN_EVENTS_WITH_HASH_V2.encode("utf-8")) + + async with StreamServer(write_stream) as server: + await server.get_client().write_events( + [event_candidate(test_data, "/books/42")], + [ + IsSubjectPristine("/books", NOT_RECURSIVE), + IsSubjectPopulated("/books", NOT_RECURSIVE), + IsSubjectOnEventId("/books", "0", NOT_RECURSIVE), + ], + ) + + [request] = server.requests + assert (request.method, request.path) == ("POST", "/api/v1/write-events") + assert json.loads(request.body)["preconditions"] == [ + {"type": "isSubjectPristine", "payload": {"subject": "/books"}}, + {"type": "isSubjectPopulated", "payload": {"subject": "/books"}}, + {"type": "isSubjectOnEventId", "payload": {"subject": "/books", "eventId": "0"}}, + ]