diff --git a/CHANGELOG.md b/CHANGELOG.md index 5141b58..a1aec81 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,8 +13,8 @@ there was nothing to have changed from. Everything above it is ordinary. - **litelink's read-cache settings on every reader of the published tables** (#77): `memory_cache`, `disk_cache`, `cache_key` and - `disk_cache_volume_limit` on `Stream.snapshot`, `scan`, `sql`, `live`, and on - `connect` for a catch-up. + `disk_cache_volume_limit` on `Stream.snapshot`, `scan`, `sql` and `live`: + the reads that repeat. Not on `connect`, whose catch-up reads its gap once. - **litelink's defaults:** memory on, disk off. - **The key is the caller's:** relative to litelink's cache root, absolute, or its `default`. Nothing is keyed automatically. diff --git a/README.md b/README.md index 0e498b5..06ec3ef 100644 --- a/README.md +++ b/README.md @@ -126,7 +126,7 @@ streamcast.connect(uri, *, offset=, cursor=None, cursor_uri=None, catch_up=False, ...) -> Subscription await sub.recv() · async for offset, ts, row in sub async for batch in sub.batches(limit=500) # what has arrived, never waiting for more -# Every reader of the published tables also takes litelink's cache settings: +# snapshot, scan, sql and live also take litelink's cache settings: # memory_cache=True, disk_cache=False, cache_key=None, disk_cache_volume_limit=0.8 streamcast.publish(uri, ...) -> Publication # server needs publish=True await producer.send(row) · await producer.send_many(rows) diff --git a/docs/API.md b/docs/API.md index e61bc59..1315a66 100644 --- a/docs/API.md +++ b/docs/API.md @@ -653,8 +653,7 @@ arithmetic. ```python streamcast.connect(uri, *, offset=, cursor=None, cursor_uri=None, s3_options=None, upload_every=30.0, catch_up=False, - catch_up_retries=3, metadata=None, memory_cache=True, - disk_cache=False, cache_key=None, disk_cache_volume_limit=0.8, + catch_up_retries=3, metadata=None, **websockets_kwargs) -> Subscription ``` @@ -1394,7 +1393,7 @@ publishing is behind, rather than running the reader out of memory. **Caching what is read is litelink's, and yours to choose.** The four keywords are `litelink.duckdb_connection`'s, with its defaults, and the same on `snapshot`, `scan`, -`sql`, `live` and `connect` (for a catch-up): +`sql` and `live` — the reads that repeat: | keyword | default | what it does | |---|---|---| @@ -1407,7 +1406,9 @@ Nothing chooses a key for you, because only you know what deserves a cache of it stream id, a team, a job. Readers asking for different settings read through different DuckDB databases. A disk cache earns its keep where reads cross a network to object storage; against a store on the same machine, it measured no faster than reading it -again. +again. `connect` takes none of them: a catch-up reads the gap to its cursor once, never +the same files twice, so it reads with litelink's defaults, as does a `live` view's +catch-up at open or after a dropped connection. **A cached reader still sees every publish.** Each read resolves the table's current metadata with `litelink.current_metadata`, outside DuckDB's caches: through a disk-cached diff --git a/src/streamcast/_catchup.py b/src/streamcast/_catchup.py index b02cd2b..81e9adc 100644 --- a/src/streamcast/_catchup.py +++ b/src/streamcast/_catchup.py @@ -44,7 +44,6 @@ from streamcast import _snapshot from streamcast._errors import NotReplayable, StreamcastError -from streamcast._published import DEFAULT_CACHE, ReadCache if TYPE_CHECKING: from collections.abc import AsyncGenerator, Awaitable, Callable @@ -136,7 +135,6 @@ class Catcher: """ __slots__ = ( - "_cache", "_first", "_handshake", "_name", @@ -158,10 +156,8 @@ def __init__( start: int, retries: int, handshake: Callable[[int], Awaitable[tuple[Any, Any]]], - cache: ReadCache = DEFAULT_CACHE, ) -> None: self._uri = uri - self._cache = cache self._stream_id = stream_id self._name = name self._s3 = s3_options @@ -180,7 +176,6 @@ async def _open(self) -> _snapshot.Snapshot: self._uri, s3_options=self._s3, stream_id=self._stream_id, - cache=self._cache, ) except _snapshot.SnapshotUnavailable as exc: raise CatchUpUnavailable(str(exc)) from exc diff --git a/src/streamcast/_client.py b/src/streamcast/_client.py index 44241ce..68499d2 100644 --- a/src/streamcast/_client.py +++ b/src/streamcast/_client.py @@ -63,7 +63,6 @@ TooSlow, ) from streamcast._protocol import decode, parse_greeting, parse_refusal -from streamcast._published import ReadCache from streamcast._remote import UPLOAD_EVERY, RemoteCursor if TYPE_CHECKING: @@ -719,7 +718,6 @@ class connect: # noqa: N801 — `websockets.connect` is lowercase and this mirr """ __slots__ = ( - "_cache", "_catch_up", "_catch_up_retries", "_cursor", @@ -748,22 +746,10 @@ def __init__( catch_up: bool = False, catch_up_retries: int = CATCH_UP_RETRIES, metadata: str | None = None, - memory_cache: bool = True, - disk_cache: bool = False, - cache_key: str | PathLike[str] | None = None, - disk_cache_volume_limit: float = 0.8, compression: str | None = None, **kwargs: Any, ) -> None: self._uri = uri - # How a catch-up's reads of the published tables are cached: - # litelink's settings, passed through. See `Stream.snapshot`. - self._cache = ReadCache.of( - memory_cache=memory_cache, - disk_cache=disk_cache, - cache_key=cache_key, - disk_cache_volume_limit=disk_cache_volume_limit, - ) # `Any`, so the checker resolves `**self._kwargs` against # `websockets.connect`'s very precise signature rather than against a # value type inferred from `compression`. @@ -865,7 +851,6 @@ async def _recover(self, refused: NotReplayable) -> Subscription: start, self._catch_up_retries, self._handshake, - self._cache, ) # BEFORE handing anything back, so an unreadable table raises here # rather than from whatever line first calls `recv`. Entering the diff --git a/src/streamcast/_live.py b/src/streamcast/_live.py index 3f45210..486e58f 100644 --- a/src/streamcast/_live.py +++ b/src/streamcast/_live.py @@ -92,8 +92,9 @@ def __init__( cache: ReadCache = DEFAULT_CACHE, ) -> None: self._broker = broker - # How every read of the published tables is cached — the base, each - # rebase, and a catch-up — as `Stream.live` was asked. + # How the base and every rebase read the published tables, as + # `Stream.live` was asked. A catch-up, at open or after a dropped + # connection, reads its gap once and is left to litelink's defaults. self._cache = cache # What the view will hold from the broker, unpublished, before it # stops: a publisher that has stalled must not run this process out @@ -148,7 +149,6 @@ async def _connect(self) -> Any: catch_up=True, s3_options=self._s3, where=self._where, - **self._cache.keywords(), # ty: ignore[invalid-argument-type] ) try: subscription = await connecting diff --git a/src/streamcast/_published.py b/src/streamcast/_published.py index d71581b..74d86d3 100644 --- a/src/streamcast/_published.py +++ b/src/streamcast/_published.py @@ -97,15 +97,6 @@ def of( disk_cache_volume_limit, ) - def keywords(self) -> dict[str, object]: - """Back to those keywords, for a call that takes them.""" - return { - "memory_cache": self.memory_cache, - "disk_cache": self.disk_cache, - "cache_key": self.cache_key, - "disk_cache_volume_limit": self.disk_cache_volume_limit, - } - DEFAULT_CACHE: Final = ReadCache() """litelink's defaults: memory on, disk off.""" diff --git a/src/streamcast/_stream.py b/src/streamcast/_stream.py index a38bcb0..b291807 100644 --- a/src/streamcast/_stream.py +++ b/src/streamcast/_stream.py @@ -743,7 +743,8 @@ async def snapshot( caller knows what deserves a cache of its own, so nothing is keyed for it. Readers with different settings read through different databases; see `litelink.duckdb_connection`. The same keywords are on - `scan`, `sql`, `live` and `connect` (for a catch-up). + `scan`, `sql` and `live`. Not on `connect`: a catch-up reads its gap + once, which no cache helps. """ return await _snapshot.snapshot( metadata_uri, diff --git a/tests/test_catchup.py b/tests/test_catchup.py index 84e8fb5..83f72d4 100644 --- a/tests/test_catchup.py +++ b/tests/test_catchup.py @@ -199,41 +199,6 @@ async def test_batches_cross_the_join_whole_and_in_order( # connection. Deterministic: no handshake completes in one loop turn. assert [len(batch) for batch in got[:2]] == [1500, TOTAL - 1500] - async def test_a_catch_up_reads_with_the_callers_cache_settings( - self, serve, published_log, s3, tmp_path, monkeypatch - ): - """`connect`'s cache keywords reach the catch-up's reads (#77).""" - from streamcast import _published # noqa: PLC0415 - - asked: list[_published.ReadCache] = [] - real = _published.connection - - def recording(s3_options, *, remote, cache=_published.DEFAULT_CACHE): # noqa: ANN001, ANN202 - asked.append(cache) - return real(s3_options, remote=remote, cache=cache) - - monkeypatch.setattr(_published, "connection", recording) - stream = streamcast.Stream("trades", log=published_log, max_replay=600) - await fill(stream, published_log) - - async with serve(stream, maintain=False) as uri: - async with streamcast.connect( - uri, - offset=1, - catch_up=True, - s3_options=s3, - disk_cache=True, - cache_key=tmp_path / "cache", - ) as sub: - assert (await sub.recv())[0] == 1 - - assert asked - assert all( - cache - == _published.ReadCache(disk_cache=True, cache_key=str(tmp_path / "cache")) - for cache in asked - ) - @pytest.mark.slow async def test_nothing_published_during_the_catch_up_is_lost( self, serve, published_log, s3