Drain pending provider updates on normal context exit

.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

docs/api_reference.md

@@ -13,6 +13,7 @@
       - Channel
       - Decoder
       - Doc
+      - UnsubscribableStream
       - Encoder
       - Map
       - MapEvent

pyproject.toml

@@ -31,7 +31,7 @@ classifiers = [
 ]
 
 dependencies = [
-    "anyio >=4.4.0,<5.0.0",
+    "anyio >=4.14.0,<5.0.0",
 ]
 
 [dependency-groups]

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

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,15 +385,15 @@ def events(
         subdocs: Literal[False] = False,
         max_buffer_size: float = float("inf"),
         async_transactions: bool = False,
-    ) -> MemoryObjectReceiveStream[TransactionEvent]: ...
+    ) -> UnsubscribableStream[TransactionEvent]: ...
 
     @overload
     def events(
         self,
         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):

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

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