diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml index 5aeb9810..d074556f 100644 --- a/.github/workflows/publish.yml +++ b/.github/workflows/publish.yml @@ -35,7 +35,7 @@ jobs: args: --release --target universal2-apple-darwin --out dist -i pypy3.11 - name: Test built wheel - universal2 run: | - pip install pytest "pydantic>=2.5.2,<3" "anyio>=4.4.0,<5" "trio>=0.25.1,<0.32" + pip install pytest "pydantic>=2.5.2,<3" "anyio>=4.14.0,<5" "trio>=0.25.1,<0.32" pip install pycrdt --no-deps --no-index --find-links dist --force-reinstall pytest --ignore tests/test_types.py - name: Upload wheels @@ -67,7 +67,7 @@ jobs: args: --release --features pyo3/abi3-py311 --out dist -i 3.11 - name: Test built wheel run: | - pip install pytest "pydantic>=2.5.2,<3" "anyio>=4.4.0,<5" "trio>=0.25.1,<0.32" + pip install pytest "pydantic>=2.5.2,<3" "anyio>=4.14.0,<5" "trio>=0.25.1,<0.32" pip install pycrdt --no-deps --no-index --find-links dist --force-reinstall pytest --ignore tests/test_types.py - name: Upload wheels @@ -110,7 +110,7 @@ jobs: - name: Test built wheel if: matrix.target == 'x86_64' run: | - pip install pytest "pydantic>=2.5.2,<3" "anyio>=4.4.0,<5" "trio>=0.25.1,<0.32" + pip install pytest "pydantic>=2.5.2,<3" "anyio>=4.14.0,<5" "trio>=0.25.1,<0.32" pip install pycrdt --no-deps --no-index --find-links dist --force-reinstall pytest --ignore tests/test_types.py - name: Upload wheels @@ -153,7 +153,7 @@ jobs: install: | apt-get update apt-get install -y --no-install-recommends python3 python3-pip - pip3 install --break-system-packages -U pytest "pydantic>=2.5.2,<3" "anyio>=4.4.0,<5" "trio>=0.25.1,<0.32" + pip3 install --break-system-packages -U pytest "pydantic>=2.5.2,<3" "anyio>=4.14.0,<5" "trio>=0.25.1,<0.32" run: | pip3 install --break-system-packages pycrdt --no-deps --no-index --find-links dist/ --force-reinstall pytest --ignore tests/test_types.py @@ -189,7 +189,7 @@ jobs: run: | docker run --rm -v "$(pwd):/workspace" -w /workspace alpine:latest sh -c " apk add --no-cache python3 py3-pip && - pip3 install --break-system-packages -U pip pytest 'pydantic>=2.5.2,<3' 'anyio>=4.4.0,<5' 'trio>=0.25.1,<0.32' && + pip3 install --break-system-packages -U pip pytest 'pydantic>=2.5.2,<3' 'anyio>=4.14.0,<5' 'trio>=0.25.1,<0.32' && pip3 install --break-system-packages pycrdt --no-deps --no-index --find-links dist/ --force-reinstall && pytest --ignore tests/test_types.py " @@ -229,7 +229,7 @@ jobs: githubToken: ${{ github.token }} install: | apk add --no-cache python3 py3-pip - pip3 install --break-system-packages -U pip pytest "pydantic>=2.5.2,<3" "anyio>=4.4.0,<5" "trio>=0.25.1,<0.32" + pip3 install --break-system-packages -U pip pytest "pydantic>=2.5.2,<3" "anyio>=4.14.0,<5" "trio>=0.25.1,<0.32" run: | pip3 install --break-system-packages pycrdt --no-deps --no-index --find-links dist/ --force-reinstall pytest --ignore tests/test_types.py diff --git a/docs/api_reference.md b/docs/api_reference.md index ee64c804..68c3ace1 100644 --- a/docs/api_reference.md +++ b/docs/api_reference.md @@ -13,6 +13,7 @@ - Channel - Decoder - Doc + - UnsubscribableStream - Encoder - Map - MapEvent diff --git a/pyproject.toml b/pyproject.toml index 77654294..ed300fa6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -31,7 +31,7 @@ classifiers = [ ] dependencies = [ - "anyio >=4.4.0,<5.0.0", + "anyio >=4.14.0,<5.0.0", ] [dependency-groups] diff --git a/python/pycrdt/__init__.py b/python/pycrdt/__init__.py index 5b373c6e..38c77248 100644 --- a/python/pycrdt/__init__.py +++ b/python/pycrdt/__init__.py @@ -25,6 +25,7 @@ from ._snapshot import Snapshot as Snapshot from ._sticky_index import Assoc as Assoc from ._sticky_index import StickyIndex as StickyIndex +from ._stream import UnsubscribableStream as UnsubscribableStream from ._sync import Decoder as Decoder from ._sync import Encoder as Encoder from ._sync import YMessageType as YMessageType diff --git a/python/pycrdt/_doc.py b/python/pycrdt/_doc.py index 334d9ee6..7aab958c 100644 --- a/python/pycrdt/_doc.py +++ b/python/pycrdt/_doc.py @@ -18,14 +18,15 @@ overload, ) -from anyio import BrokenResourceError, create_memory_object_stream -from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream +from anyio import BrokenResourceError, ClosedResourceError, create_memory_object_stream +from anyio.streams.memory import MemoryObjectSendStream from ._base import BaseDoc, BaseType, Typed, base_types, forbid_read_transaction from ._pycrdt import Doc as _Doc from ._pycrdt import SubdocsEvent, Subscription, TransactionEvent from ._pycrdt import Transaction as _Transaction from ._snapshot import Snapshot +from ._stream import UnsubscribableStream, _with_unsubscribe from ._transaction import NewTransaction, ReadTransaction, Transaction T = TypeVar("T", bound=BaseType) @@ -384,7 +385,7 @@ def events( subdocs: Literal[False] = False, max_buffer_size: float = float("inf"), async_transactions: bool = False, - ) -> MemoryObjectReceiveStream[TransactionEvent]: ... + ) -> UnsubscribableStream[TransactionEvent]: ... @overload def events( @@ -392,7 +393,7 @@ def events( subdocs: Literal[True] = True, max_buffer_size: float = float("inf"), async_transactions: bool = False, - ) -> MemoryObjectReceiveStream[list[SubdocsEvent]]: ... + ) -> UnsubscribableStream[list[SubdocsEvent]]: ... def events( self, @@ -415,6 +416,9 @@ async def main(): ... ``` + Call `events.unsubscribe()` to stop collecting events and drain the buffer + before iteration ends. Closing the receiving stream discards pending events. + Args: subdocs: Whether to iterate over the [SubdocsEvent][pycrdt.SubdocsEvent] events (default is [TransactionEvent][pycrdt.TransactionEvent]). @@ -438,7 +442,17 @@ async def main(): Union[TransactionEvent, SubdocsEvent] ](max_buffer_size=max_buffer_size) self._send_streams[subdocs].add(send_stream) - return receive_stream + return _with_unsubscribe( + receive_stream, partial(self._unsubscribe_events, subdocs, send_stream) + ) + + def _unsubscribe_events(self, subdocs, send_stream): + send_streams = self._send_streams[subdocs] + if send_stream in send_streams: + send_streams.remove(send_stream) + send_stream.close() + if not send_streams: + self.unobserve(self._event_subscription.pop(subdocs)) def _send_event(self, subdocs: bool, event: TransactionEvent | SubdocsEvent): to_remove: list[MemoryObjectSendStream[TransactionEvent | SubdocsEvent]] = [] @@ -449,24 +463,18 @@ def _send_event(self, subdocs: bool, event: TransactionEvent | SubdocsEvent): except BrokenResourceError: to_remove.append(send_stream) for send_stream in to_remove: - send_stream.close() - send_streams.remove(send_stream) - if not send_streams: - self.unobserve(self._event_subscription[subdocs]) + self._unsubscribe_events(subdocs, send_stream) async def _async_send_event(self, subdocs: bool, event: TransactionEvent | SubdocsEvent): to_remove: list[MemoryObjectSendStream[TransactionEvent | SubdocsEvent]] = [] send_streams = self._send_streams[subdocs] - for send_stream in send_streams: + for send_stream in tuple(send_streams): try: await send_stream.send(event) - except BrokenResourceError: + except (BrokenResourceError, ClosedResourceError): to_remove.append(send_stream) for send_stream in to_remove: - send_stream.close() - send_streams.remove(send_stream) - if not send_streams: - self.unobserve(self._event_subscription[subdocs]) + self._unsubscribe_events(subdocs, send_stream) class TypedDoc(Typed): diff --git a/python/pycrdt/_provider.py b/python/pycrdt/_provider.py index eb6a6897..d578cd1f 100644 --- a/python/pycrdt/_provider.py +++ b/python/pycrdt/_provider.py @@ -2,17 +2,21 @@ from contextlib import AsyncExitStack, asynccontextmanager from logging import Logger, getLogger +from sys import exc_info from typing import AsyncIterator, Protocol from anyio import ( TASK_STATUS_IGNORED, Event, Lock, + TaskHandle, create_task_group, ) from anyio.abc import TaskGroup, TaskStatus from ._doc import Doc +from ._pycrdt import TransactionEvent +from ._stream import UnsubscribableStream from ._sync import ( YMessageType, YSyncMessageType, @@ -83,6 +87,10 @@ def __init__(self, doc: Doc, channel: Channel, log: Logger | None = None) -> Non The `synced` event is set when initial synchronization with the remote peer is complete. Later changes continue to synchronize in the background. + On normal context exit, queued updates finish sending through the channel + before the provider stops. Exceptions and cancellation stop it immediately. + Channel sends do not confirm receipt or persistence by the remote peer. + However, a lower-level API can also be used: ```py task = asyncio.create_task(provider.start()) @@ -103,17 +111,10 @@ def __init__(self, doc: Doc, channel: Channel, log: Logger | None = None) -> Non self.synced = Event() self._start_lock = Lock() self._task_group: TaskGroup | None = None + self._events: UnsubscribableStream[TransactionEvent] | None = None + self._updates_task: TaskHandle[None] | None = None async def _run(self): - sync_message = create_sync_message(self._doc) - self.log.debug( - "Sending %s message to endpoint: %s", - YSyncMessageType.SYNC_STEP1.name, - self._channel.path, - ) - await self._channel.send(sync_message) - assert self._task_group is not None - self._task_group.start_soon(self._send_updates) async for message in self._channel: if message[0] == YMessageType.SYNC: self.log.debug( @@ -132,11 +133,22 @@ async def _run(self): ) await self._channel.send(reply) - async def _send_updates(self): - async with self._doc.events() as events: - async for event in events: - message = create_update_message(event.update) - await self._channel.send(message) + async def _send_updates(self) -> None: + assert self._events is not None + async with self._events as events: + try: + sync_message = create_sync_message(self._doc) + self.log.debug( + "Sending %s message to endpoint: %s", + YSyncMessageType.SYNC_STEP1.name, + self._channel.path, + ) + await self._channel.send(sync_message) + async for event in events: + message = create_update_message(event.update) + await self._channel.send(message) + finally: + events.unsubscribe() async def __aenter__(self) -> Provider: async with AsyncExitStack() as exit_stack: @@ -148,6 +160,16 @@ async def __aenter__(self) -> Provider: return self async def __aexit__(self, exc_type, exc_value, exc_tb): + if exc_type is None: + try: + assert self._events is not None + self._events.unsubscribe() + assert self._updates_task is not None + await self._updates_task + except BaseException: + await self.stop() + await self._exit_stack.__aexit__(*exc_info()) + raise await self.stop() return await self._exit_stack.__aexit__(exc_type, exc_value, exc_tb) @@ -172,12 +194,17 @@ async def start( """ async with self._start_lock: async with self._get_or_create_task_group() as self._task_group: + self._events = self._doc.events() + self._updates_task = self._task_group.start_soon(self._send_updates) task_status.started() self.started.set() self._task_group.start_soon(self._run) async def stop(self) -> None: - """Stop the provider.""" + """Stop the provider immediately, discarding pending updates.""" assert self._task_group is not None + assert self._events is not None + self._events.unsubscribe() + self._events.close() self._task_group.cancel_scope.cancel() self._task_group = None diff --git a/python/pycrdt/_stream.py b/python/pycrdt/_stream.py new file mode 100644 index 00000000..6900f013 --- /dev/null +++ b/python/pycrdt/_stream.py @@ -0,0 +1,32 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, Callable, TypeVar, cast + +from anyio.streams.memory import MemoryObjectReceiveStream + +E = TypeVar("E") + + +class UnsubscribableStream(MemoryObjectReceiveStream[E]): + """A receive stream with an unsubscribe method. + + Unsubscribing stops collecting items while leaving buffered items available + to receive. Iteration ends once the buffer is drained. + """ + + unsubscribe: Callable[[], None] + """Stop collecting items. Repeated calls have no effect.""" + + if TYPE_CHECKING: + + def clone(self) -> UnsubscribableStream[E]: ... + + +def _with_unsubscribe( + stream: MemoryObjectReceiveStream[E], unsubscribe: Callable[[], None] +) -> UnsubscribableStream[E]: + stream = cast(UnsubscribableStream[E], stream) + stream.unsubscribe = unsubscribe + original_clone = stream.clone + stream.clone = lambda: _with_unsubscribe(original_clone(), unsubscribe) # type: ignore[method-assign] + return stream diff --git a/tests/test_event_stream.py b/tests/test_event_stream.py new file mode 100644 index 00000000..e1c160e7 --- /dev/null +++ b/tests/test_event_stream.py @@ -0,0 +1,80 @@ +import pytest +from anyio import EndOfStream, Event, create_task_group, fail_after, wait_all_tasks_blocked +from anyio.streams.memory import MemoryObjectReceiveStream +from pycrdt import Doc, Text + +pytestmark = pytest.mark.anyio + + +@pytest.mark.parametrize("async_transactions", [False, True]) +async def test_unsubscribe_drains_buffer(async_transactions): + doc = Doc({"text": Text()}) + text = doc["text"] + async with doc.events(async_transactions=async_transactions) as events: + for value in ("A", "B"): + if async_transactions: + async with doc.transaction(): + text += value + else: + text += value + events.unsubscribe() + events.unsubscribe() + text += "C" + updates = [event.update async for event in events] + replica = Doc({"text": Text()}) + for update in updates: + replica.apply_update(update) + assert str(replica["text"]) == "AB" + async with doc.events() as events: + text += "D" + events.unsubscribe() + assert len([event async for event in events]) == 1 + + +async def test_unsubscribe_keeps_other_subscribers(): + doc = Doc({"text": Text()}) + async with doc.events() as first, doc.events() as second: + doc["text"] += "A" + first.unsubscribe() + doc["text"] += "B" + second.unsubscribe() + assert len([event async for event in first]) == 1 + assert len([event async for event in second]) == 2 + + +async def test_unsubscribe_with_clones(): + doc = Doc({"text": Text()}) + with doc.events() as events, events.clone() as clone: + assert type(events) is MemoryObjectReceiveStream + assert type(clone) is MemoryObjectReceiveStream + assert events.unsubscribe is clone.unsubscribe + assert events.statistics().open_receive_streams == 2 + doc["text"] += "A" + clone.receive_nowait() + clone.unsubscribe() + with pytest.raises(EndOfStream): + await events.receive() + with pytest.raises(EndOfStream): + await clone.receive() + + +async def test_unsubscribe_drains_waiting_async_producer(): + doc = Doc({"text": Text()}) + done = Event() + async with doc.events(max_buffer_size=0, async_transactions=True) as events: + + async def edit(): + async with doc.transaction(): + doc["text"] += "A" + done.set() + + with fail_after(3): + async with create_task_group() as tasks: + tasks.start_soon(edit) + await wait_all_tasks_blocked() + assert events.statistics().tasks_waiting_send == 1 + events.unsubscribe() + assert len([event async for event in events]) == 1 + await done.wait() + with pytest.raises(EndOfStream): + await events.receive() diff --git a/tests/test_provider_shutdown.py b/tests/test_provider_shutdown.py new file mode 100644 index 00000000..7abc12d8 --- /dev/null +++ b/tests/test_provider_shutdown.py @@ -0,0 +1,127 @@ +import pytest +from anyio import Event, create_task_group, fail_after, sleep_forever +from pycrdt import Channel, Doc, Provider, Text, YSyncMessageType, handle_sync_message + +pytestmark = pytest.mark.anyio + + +class SlowChannel(Channel): + def __init__(self, fail=False, block_initial=False): + self.send_started = Event() + self.allow_send = Event() + self.messages = [] + self.fail = fail + self.block_initial = block_initial + + @property + def path(self): + return "test" + + async def send(self, message): + if self.block_initial or message[1] == YSyncMessageType.SYNC_UPDATE: + self.send_started.set() + await self.allow_send.wait() + if self.fail: + raise OSError("send failed") + self.messages.append(message) + + async def recv(self): + await sleep_forever() + + +async def test_exit_drains_queued_and_inflight_updates(): + doc = Doc({"text": Text()}) + channel = SlowChannel() + exited = Event() + + async def edit(): + async with Provider(doc, channel): + for value in ("A", "B", "C"): + doc["text"] += value + exited.set() + + with fail_after(3): + async with create_task_group() as tasks: + tasks.start_soon(edit) + await channel.send_started.wait() + assert not exited.is_set() + channel.allow_send.set() + await exited.wait() + replica = Doc({"text": Text()}) + updates = [m for m in channel.messages if m[1] == YSyncMessageType.SYNC_UPDATE] + assert len(updates) == 3 + for message in updates: + handle_sync_message(message[1:], replica) + assert str(replica["text"]) == "ABC" + + +async def test_exit_without_updates(): + channel = SlowChannel() + with fail_after(3): + async with Provider(Doc(), channel): + pass + assert len(channel.messages) == 1 + + +async def test_exception_exit_does_not_drain(): + doc = Doc({"text": Text()}) + channel = SlowChannel() + with pytest.raises(ExceptionGroup) as error: + with fail_after(3): + async with Provider(doc, channel): + doc["text"] += "A" + raise ValueError("body failed") + assert error.value.subgroup(ValueError) is not None + assert not channel.allow_send.is_set() + + +async def test_cancellation_during_drain(): + doc = Doc({"text": Text()}) + channel = SlowChannel() + stopped = Event() + + async def edit(): + try: + async with Provider(doc, channel): + doc["text"] += "A" + finally: + stopped.set() + + with fail_after(3): + async with create_task_group() as tasks: + tasks.start_soon(edit) + await channel.send_started.wait() + tasks.cancel_scope.cancel() + assert stopped.is_set() + assert not channel.allow_send.is_set() + + +@pytest.mark.parametrize("block_initial", [False, True]) +async def test_send_failure_propagates(block_initial): + doc = Doc({"text": Text()}) + channel = SlowChannel(fail=True, block_initial=block_initial) + channel.allow_send.set() + with pytest.raises(ExceptionGroup) as error: + with fail_after(3): + async with Provider(doc, channel): + doc["text"] += "A" + assert error.value.subgroup(OSError) is not None + + +async def test_exit_waits_for_initial_send(): + channel = SlowChannel(block_initial=True) + exited = Event() + + async def run(): + async with Provider(Doc(), channel): + pass + exited.set() + + with fail_after(3): + async with create_task_group() as tasks: + tasks.start_soon(run) + await channel.send_started.wait() + assert not exited.is_set() + channel.allow_send.set() + await exited.wait() + assert len(channel.messages) == 1