diff --git a/CHANGELOG.md b/CHANGELOG.md index a1aec81..06f8b59 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,31 @@ there was nothing to have changed from. Everything above it is ordinary. ### Added +- **`Stream.retire`** (#96): finish a stream for good. + - **What it does:** publishes every row (including the trailing run a plain + `publish` holds back), retires the log in litelink, and records the + retirement in `metadata.json`, locally and beside the tables. + - **Read-only from then on:** a retired stream is still served. Subscribers + replay and catch up from the published table. A local `send` raises + `StreamRetired`, a publisher is refused with the new close code 4410, and + it gets no maintainer or sidecar. + - **Undoing it:** `Stream.restore(..., revive=True)`, on any box, continues + it on a new log at exactly the retired end, losing nothing. + - **Older builds refuse it:** a retired stream's `metadata.json` is version + 3, so an older streamcast refuses it instead of quietly starting a new log. +- **`Stream.restore(replica_reserve=, published_reserve=)`**, litelink's offset + fences, passed through. +- **`Stream.restore(schema=, sort_by=, config=)`.** The restored log's shape + comes from the stream's metadata by default: its exact schema (binary + encodings and system columns included, which an Iceberg schema doesn't keep) + and its `sort_by`. litelink 0.10.1 checks them against the replica or table, + and needs them for a table no 0.10 publish stamped. `schema=` and `sort_by=` + override, or cover a stream whose metadata predates recording them. + `config=` sets the restored log's policy, and the new log's on `revive=True`. +- **`metadata.json` records each log's `sort_by`.** `serve` fills it in for an + existing stream from its open live log; a log not on the machine stays + unknown (`null`). + - **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` and `live`: @@ -51,6 +76,17 @@ there was nothing to have changed from. Everything above it is ordinary. ### Changed +- **BREAKING: `serve` and `asgi` no longer take `publish=`**, and every served + stream takes publishers. Whether a stream can be written is the stream's to + say: a retired one refuses publishers with 4410. A call passing `publish=` + raises `TypeError`. There's no authentication yet, for readers or writers, + so a reachable port is a readable and writable one, as it was for readers + already. + +- **`Stream.restore` works on a stream that never had WAL replication.** + litelink rebuilds the log from its published table when there's no replica, + fenced 2^40 above what the old log last recorded issuing (litelink#144). + - **litelink 0.9** (`>=0.9.0,<0.10`), for `litelink.current_metadata`. - **A published table's current metadata is resolved outside DuckDB**, with `litelink.current_metadata`. Through a disk-cached connection, diff --git a/README.md b/README.md index 06ec3ef..73d625a 100644 --- a/README.md +++ b/README.md @@ -119,7 +119,7 @@ await streamcast.Stream.live(broker, *, s3_options=None) -> Live # kept curren await live.scan(...) · await live.sql(query) · await live.wait_for(offset | ts=) streamcast.serve(streams, host, port, *, maintain=True, replicate=False, - publish=False, max_backlog=8192, # frames per subscriber + max_backlog=8192, # frames per subscriber max_inbound=65_536, # rows queued to commit max_in_flight=64, ...) -> Server # an int, or {stream: int} streamcast.connect(uri, *, offset=, cursor=None, cursor_uri=None, @@ -128,7 +128,7 @@ streamcast.connect(uri, *, offset=, cursor=None, cursor_uri=None, async for batch in sub.batches(limit=500) # what has arrived, never waiting for more # 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 +streamcast.publish(uri, ...) -> Publication # any served stream await producer.send(row) · await producer.send_many(rows) await producer.submit(row) -> Future # pipelined: up to max_in_flight=64 unanswered streamcast.to_arrow · streamcast.from_arrow · streamcast.Cursor @@ -412,8 +412,8 @@ other; `just bench-replay` prints the arithmetic for your hardware. ### Recovering a server A client moves boxes with a cursor. A **server** moves with `Stream.restore`, which -rebuilds the log itself from its published table and the replicated WAL on a machine -that never held it: +rebuilds the log itself on a machine that never held it: from its replicated WAL when +there is one, and otherwise from its published table alone: ```python stream = streamcast.Stream.restore( @@ -421,8 +421,11 @@ stream = streamcast.Stream.restore( ) ``` -Offsets are **fenced, not reissued** — litelink burns 2²⁰ — so no offset a consumer -holds is ever handed out again carrying different data. The consumer resumes from the +Offsets are **fenced, not reissued**, so no offset a consumer holds is ever handed out +again carrying different data. From a replica, litelink skips 2²⁰ past what the replica +recorded (`replica_reserve=`). From the published table alone, it skips 2⁴⁰ past what the +old log last said it had issued (`published_reserve=`), since rows written after the last +publish are lost with the machine. The consumer resumes from the cursor it already had and sees a gap, which `recv` allows. Existing consumers resume with no intervention, because `max_replay` counts **rows** @@ -434,6 +437,11 @@ The local staging table comes back empty — its Parquet was on the dead machine restored stream that should replay history serves it from the published table, with `replay_published=True`. +The log comes back with its exact shape, read from the stream's metadata: its schema, +binary encodings and system columns included, and its `sort_by`. Pass `schema=` and +`sort_by=` for a stream whose metadata predates recording them, and `config=` for the +restored log's policy, which is otherwise litelink's default when there was no replica. + A **planned** cutover loses nothing — stop the writer, let the sidecar ship its last frames, then restore. Unplanned failover loses whatever never shipped. @@ -513,7 +521,7 @@ from contextlib import asynccontextmanager from fastapi import FastAPI from streamcast.asgi import asgi -streams = asgi([trades, quotes], publish=True) +streams = asgi([trades, quotes]) @asynccontextmanager async def lifespan(app): @@ -565,8 +573,7 @@ async with streamcast.publish("ws://localhost:8765/trades") as producer: offset = await producer.send({"event_ts": 1790038800123456, "price": 85565.0}) ``` -The server must allow it — `serve(..., publish=True)`, off by default so an upgrade never -makes a server writable on its own. `send` returns once the row is durable, exactly as the +Every served stream takes publishers; a retired one refuses them with 4410. `send` returns once the row is durable, exactly as the local call does; `send_many` commits a group in one transaction and is the same throughput lever it is locally. A row the schema refuses raises `Rejected`, naming the column, and the connection stays open so the next row works. diff --git a/docs/API.md b/docs/API.md index 1315a66..7285c9a 100644 --- a/docs/API.md +++ b/docs/API.md @@ -223,7 +223,7 @@ Drops every subscriber with a 1001, concurrently. **Does not close the log.** ```python streamcast.serve(streams, host=None, port=None, *, maintain=True, - replicate=False, publish=False, max_backlog=8192, + replicate=False, max_backlog=8192, max_inbound=65_536, max_in_flight=64, **websockets_kwargs) -> Server ``` @@ -497,7 +497,7 @@ A mounted ASGI app gets no equivalent and needs none — it has routes already, ```python from streamcast.asgi import asgi -asgi(streams, *, maintain=True, replicate=False, publish=False, +asgi(streams, *, maintain=True, replicate=False, max_backlog=8192, max_inbound=65_536, max_in_flight=64) ``` @@ -535,8 +535,8 @@ lifespan events alone starts no maintainer once mounted, and a log with nothing buffers every row it ever receives. The app handles `lifespan` too, for the case where it is run directly rather than mounted; both routes are idempotent. -Routing, refusals and close codes are identical to `serve`'s, including the 4400/4404/4416/ -4429 reasons, with one difference ASGI forces: the handshake is **accepted** before a +Routing, refusals and close codes are identical to `serve`'s, including the 4400/4404/4410/ +4416/4429 reasons, with one difference ASGI forces: the handshake is **accepted** before a refusal can be sent, because a close code only exists on an accepted connection. `websockets` refuses after its handshake too, so what a client sees is the same; rejecting the upgrade instead would turn every refusal into an HTTP 403 with the reason discarded. @@ -602,9 +602,10 @@ async with streamcast.publish("ws://localhost:8765/trades") as producer: offset = await producer.send({"event_ts": 1790038800123456, "price": 85565.0}) ``` -**The server must allow it**: `serve(..., publish=True)`. Off by default, so an upgrade -never makes a server writable on its own. A publisher meeting a server that does not -allow it gets a `ProtocolError` naming the setting rather than a silent failure. +**Every served stream takes publishers.** Whether a stream can be written is the +stream's to say, not the server's: a retired one refuses a publisher with 4410 +(`StreamRetired`). There is no authentication yet, so a port any client can reach is +one any client can read and write. The URI is the stream's, the same string `connect` takes — `?publish` is added by the client, so one address in a config file serves both ends. @@ -1021,8 +1022,8 @@ StreamcastError `.fields` carries whatever numbers survived the close frame — `offset`, `earliest`, `behind`, `max_replay`, `end_offset` — and `str(exc)` is a sentence built from them. -`Close` is the code enum: `BAD_REQUEST` 4400, `NO_SUCH_STREAM` 4404, `NOT_REPLAYABLE` -4416, `TOO_SLOW` 4429. +`Close` is the code enum: `BAD_REQUEST` 4400, `NO_SUCH_STREAM` 4404, `RETIRED` 4410, +`NOT_REPLAYABLE` 4416, `TOO_SLOW` 4429. ## The schema is yours @@ -1209,6 +1210,39 @@ it had. A stream's logs are read together with `UNION ALL BY NAME`, where a chan coerces silently (`int64` beside `string` becomes a string column) rather than failing. Widening is refused too. +### Finishing a stream: `Stream.retire` + +A stream that just stops being written isn't finished: its last rows stay on the box, +because `publish` holds back the trailing run for compaction, and nothing refuses a late +write. With the server stopped: + +```python +streamcast.Stream.retire("trades", root="data") # finished +stream = streamcast.Stream.restore( + "trades", root="data", published="s3://market-data/prod", revive=True, +) # and undone +``` + +Retiring publishes every row (the trailing run included), retires the log in litelink so +it refuses writers for good, and records it in `metadata.json`: when (`at`, UTC +microseconds), where the log ended, its `sort_by` and its `streamcast_ts` span, locally +and beside the tables. A retire that dies partway finishes on the rerun. + +**A retired stream is read-only, and still served.** `new` and `migrate` open its log for +reading and set `stream.retirement`. Subscribers replay and catch up, always from the +published table, its only copy; a `live` view sees nothing new. A local `send` raises +`StreamRetired`, a publisher is refused with 4410, and it gets no maintainer or sidecar. + +**`restore(..., revive=True)` undoes it**, on the same box or another one: the stream +continues on its next log, `trades-v2` or later, starting at exactly the retired end, with +the same columns and sort. Retiring published everything first, so reviving loses nothing, +and nothing is needed from the old box. Without `revive=True`, `restore` refuses a +retired stream. + +A retired stream's `metadata.json` is version 3, so a streamcast build from before +retirement refuses it, rather than taking the retired log for a migration that died and +quietly starting the next one. Reviving writes version 2 again. + **There is no rename.** Changing `price` to `px` removes one column and adds another. Nothing is backfilled or merged, so a read across the seam returns both, each null in the logs that did not have it. Combining them (`coalesce(px, price)`) is up to your diff --git a/docs/SPEC.md b/docs/SPEC.md index 1529abc..4d45eca 100644 --- a/docs/SPEC.md +++ b/docs/SPEC.md @@ -289,6 +289,7 @@ would turn a rejected subscribe into an empty stream. |---|---|---| | 4400 | the path or query will not parse | `ProtocolError` | | 4404 | nothing served there | `StreamNotFound` | +| 4410 | the stream is retired and takes no rows (to a publisher) | `StreamRetired` | | 4416 | that offset cannot be served | `NotReplayable` | | 4429 | you fell too far behind | `TooSlow` | | 1000/1001 | the server finished on purpose | ends the iteration | @@ -1204,9 +1205,9 @@ it. A pipeline that needs end-to-end correlation puts its own id in the payload. ## 6b. Remote publishing `streamcast.publish(uri)` hands rows to a server, which appends them with the -same `Stream.send` / `send_many` a local publisher calls. Opt-in on the server -(`serve(..., publish=True)`), because a server that became writable on an -upgrade would be a change nobody asked for. +same `Stream.send` / `send_many` a local publisher calls. Every served stream +takes publishers; one that cannot be written, a retired stream, refuses them +(4410). **It adds no authority, and that is the whole argument for it.** litelink allows one writer per log, refuses neither a second nor detects one, and has diff --git a/examples/README.md b/examples/README.md index cc4d085..2c2f721 100644 --- a/examples/README.md +++ b/examples/README.md @@ -170,7 +170,7 @@ and the transport only carries frames. Which is why `/health` in that file reads `trades.end_offset` directly without asking the websocket layer anything. Three lines in the file are the whole of it: build the stream with `Stream.new`, wrap it -with `asgi(..., publish=True)`, and `app.mount("/streams", streams)`. The fourth thing to know is that +with `asgi(...)`, and `app.mount("/streams", streams)`. The fourth thing to know is that `async with streams` in the lifespan is **not** optional — Starlette does not run a mounted sub-app's lifespan, so that block is what starts the maintainer which seals the log, and what closes the log on the way out. diff --git a/examples/broker.py b/examples/broker.py index 2203cd4..288fec0 100644 --- a/examples/broker.py +++ b/examples/broker.py @@ -85,8 +85,8 @@ async def main( ) ) - # `publish=True`: producers are clients, and publish over the socket. - async with streamcast.serve(streams, host, port, publish=True): + # Producers are clients, and publish over the socket. + async with streamcast.serve(streams, host, port): for stream in streams: where = "live-only" if live else f"durable, next offset {stream.end_offset}" print(f"ws://{host}:{port}/{stream.name} ({where})", flush=True) diff --git a/examples/fastapi_app.py b/examples/fastapi_app.py index f8a866c..ec1dca2 100644 --- a/examples/fastapi_app.py +++ b/examples/fastapi_app.py @@ -48,9 +48,9 @@ ) # 2. The transport. No host, no port: the ASGI server this runs under owns the -# socket. `publish=True` because producers are clients: they publish over -# the socket, to the same mount subscribers read from. -streams = asgi(trades, publish=True) +# socket. Producers are clients: they publish over the socket, to the +# same mount subscribers read from. +streams = asgi(trades) @contextlib.asynccontextmanager diff --git a/examples/keyed_table/branches.py b/examples/keyed_table/branches.py index 5abf82b..b7fbc69 100644 --- a/examples/keyed_table/branches.py +++ b/examples/keyed_table/branches.py @@ -94,7 +94,7 @@ def book(view: View) -> list[int]: async def main(root: Path) -> dict[str, Any]: stream = streamcast.Stream.new("orders", root=root, schema=SCHEMA) - server = await streamcast.serve(stream, "127.0.0.1", 0, publish=True) + server = await streamcast.serve(stream, "127.0.0.1", 0) uri = f"ws://127.0.0.1:{server.sockets[0].getsockname()[1]}/orders" following: list[asyncio.Task[None]] = [] diff --git a/examples/keyed_table/orders.py b/examples/keyed_table/orders.py index f69d5a9..7d39882 100644 --- a/examples/keyed_table/orders.py +++ b/examples/keyed_table/orders.py @@ -89,7 +89,7 @@ async def stopped_after(view: View, uri: str, offset: int) -> None: async def main(root: Path) -> dict[str, Any]: stream = streamcast.Stream.new("orders", root=root, schema=SCHEMA) - server = await streamcast.serve(stream, "127.0.0.1", 0, publish=True) + server = await streamcast.serve(stream, "127.0.0.1", 0) uri = f"ws://127.0.0.1:{server.sockets[0].getsockname()[1]}/orders" path = root / "orders.db" try: diff --git a/examples/migration/demo.py b/examples/migration/demo.py index 55f688e..0982470 100644 --- a/examples/migration/demo.py +++ b/examples/migration/demo.py @@ -117,9 +117,7 @@ async def main(root: Path, *, bug: bool = False, count: int = 200) -> dict[str, } # No maintainer: a run this short never fills a log enough to seal it. A # pipeline that runs for real keeps `serve`'s default, which starts one. - server = await streamcast.serve( - streams.values(), "127.0.0.1", 0, publish=True, maintain=False - ) + server = await streamcast.serve(streams.values(), "127.0.0.1", 0, maintain=False) base = f"ws://127.0.0.1:{server.sockets[0].getsockname()[1]}" running: list[asyncio.Task[None]] = [] try: diff --git a/examples/otel/demo.py b/examples/otel/demo.py index 6f7dfdb..6ad3d2e 100644 --- a/examples/otel/demo.py +++ b/examples/otel/demo.py @@ -13,7 +13,7 @@ URI. `just demo otel` runs them as separate processes. 1. **A broker** serves three streams, `logs`, `spans` and `metrics`, each - with a log, accepting remote publishers (`publish=True`). + with a log, accepting remote publishers. 2. **The producer**: two services, checkout and payments, handle traced requests, log through the standard `logging` module, and count orders and time requests. OTel's SDK turns each into records, spans and metric data @@ -90,7 +90,6 @@ async def main(root: Path) -> dict[str, Any]: [log_stream, span_stream, metric_stream], "127.0.0.1", 0, - publish=True, maintain=False, ) base = f"ws://127.0.0.1:{server.sockets[0].getsockname()[1]}" diff --git a/pyproject.toml b/pyproject.toml index b34f84f..29a2a61 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -44,12 +44,14 @@ dependencies = [ # for finite-only floats, which is what makes a float column prunable; # 0.5.1 for seals sized by what the Arrow table actually weighs (#84); # 0.8.1 for Python 3.14; 0.9.0 for `current_metadata`, a published - # table's current metadata read outside DuckDB's caches (litelink#141). + # table's current metadata read outside DuckDB's caches (litelink#141); + # 0.10.0 for `restore` without a WAL replica, and its reserve keywords; + # 0.10.1 for `restore`'s schema, sort_by and config. # # Capped below the next minor: under 0.x a minor is litelink's breaking # release, and a published streamcast must not resolve one it was never # tested against. Each release moves the cap deliberately. - "litelink>=0.9.0,<0.10", + "litelink>=0.10.1,<0.11", # Serialisation is on the hot path in BOTH directions — every publish # encodes a row, every replayed row re-encodes one — which is what earns # a compiled dependency here. Measured against stdlib `json` on a diff --git a/src/streamcast/__init__.py b/src/streamcast/__init__.py index 6ada380..1e69ab7 100644 --- a/src/streamcast/__init__.py +++ b/src/streamcast/__init__.py @@ -115,6 +115,7 @@ Rejected, StreamcastError, StreamNotFound, + StreamRetired, TooSlow, ) from streamcast._limits import MAX_BACKLOG @@ -155,6 +156,7 @@ "Stats", "Stream", "StreamNotFound", + "StreamRetired", "StreamcastError", "Subscription", "TooSlow", diff --git a/src/streamcast/_client.py b/src/streamcast/_client.py index 68499d2..d2c9177 100644 --- a/src/streamcast/_client.py +++ b/src/streamcast/_client.py @@ -60,6 +60,7 @@ ProtocolError, StreamcastError, StreamNotFound, + StreamRetired, TooSlow, ) from streamcast._protocol import decode, parse_greeting, parse_refusal @@ -146,6 +147,14 @@ def _refusal( return NotReplayable(str(why), **fields) + if close.code == Close.RETIRED: + at, end = fields.get("at"), fields.get("end_offset") + return StreamRetired( + stream, + at if isinstance(at, int) else None, + end if isinstance(end, int) else None, + ) + if close.code == Close.TOO_SLOW: backlog = fields.get("backlog") @@ -156,10 +165,12 @@ def _refusal( # The sentence is built here for the same reason every other one # is: 123 bytes is not room for it, and the wire carries the name # so the English can be reworded without a protocol change. + # Sent only by a server from before every stream took + # publishers, started without `publish=True`. msg = ( - f"{stream!r} does not accept publishers. The server decides " - f"this, not the stream — start it with " - f"serve(..., publish=True) to allow remote publishing." + f"{stream!r} does not accept publishers: the server is an older " + f"streamcast started without serve(..., publish=True). Upgrade " + f"it, or start it with that." ) return ProtocolError(msg) diff --git a/src/streamcast/_errors.py b/src/streamcast/_errors.py index 1df13d4..9eafc5e 100644 --- a/src/streamcast/_errors.py +++ b/src/streamcast/_errors.py @@ -38,6 +38,10 @@ class Close(IntEnum): NO_SUCH_STREAM = 4404 """Nothing is served at that path.""" + RETIRED = 4410 + """The stream is retired: finished, and taking no rows. Sent to a + publisher; a subscriber still reads its history. See `StreamRetired`.""" + NOT_REPLAYABLE = 4416 """The requested offset cannot be served. `why` says which of the five reasons it is — see `NotReplayable`.""" @@ -85,6 +89,38 @@ def __init__(self, requested: str, serves: tuple[str, ...] = ()) -> None: super().__init__(f"no stream at {requested!r}{known}") +class StreamRetired(StreamcastError): + """The stream is retired (`Stream.retire`): finished, and taking no rows. + + Its history stays readable — served read-only, snapshotted, caught up + from — but nothing appends to it. `Stream.restore(..., revive=True)` + brings it back on a new log that starts where it ended. + """ + + def __init__( + self, stream: str, at: int | None = None, end_offset: int | None = None + ) -> None: + self.stream = stream + self.at = at + """When it was retired, in UTC microseconds; None if a refusal's + reason was trimmed to fit its close frame.""" + self.end_offset = end_offset + """One past its last row; None as `at`.""" + detail = ", ".join( + part + for part in ( + None if at is None else f"at {at} us UTC", + None if end_offset is None else f"ending before offset {end_offset}", + ) + if part is not None + ) + super().__init__( + f"stream {stream!r} is retired{f' ({detail})' if detail else ''} and " + f"takes no rows; Stream.restore(..., revive=True) continues it on a " + f"new log" + ) + + class _Missing(dict): """A format mapping that answers `?` for a key the wire did not carry. @@ -212,6 +248,7 @@ def __init__( "ProtocolError", "Rejected", "StreamNotFound", + "StreamRetired", "StreamcastError", "TooSlow", ] diff --git a/src/streamcast/_log.py b/src/streamcast/_log.py index 6f3f8aa..90464e7 100644 --- a/src/streamcast/_log.py +++ b/src/streamcast/_log.py @@ -59,9 +59,9 @@ from streamcast._protocol import encode_projected as _encode_projected if TYPE_CHECKING: - from collections.abc import AsyncGenerator, Callable, Mapping, Sequence + from collections.abc import AsyncGenerator, Callable, Iterable, Mapping, Sequence - from litelink import LogHandle, WriteHandle + from litelink import LogHandle from streamcast._filter import Predicate @@ -194,15 +194,20 @@ def system(schema: pa.Schema) -> pa.Schema: return pa.schema(owned) -def with_system(schema: pa.Schema) -> pa.Schema: +def with_system(schema: pa.Schema, names: Iterable[str] | None = None) -> pa.Schema: """`schema` with every `SYSTEM` column appended, for creating a log. Last rather than first so a `SELECT *` reads the application's columns in the order they were declared. Not nullable: every row this library writes carries them, and a log that says so is one a reader can rely on. + + `names` appends only those, for rebuilding a log that has fewer — one + created before a system column existed — exactly as it is. """ + wanted = None if names is None else set(names) for field in _SYSTEM_ARROW: - schema = schema.append(field) + if wanted is None or field.name in wanted: + schema = schema.append(field) return schema @@ -387,7 +392,7 @@ def lowest(log: LogHandle) -> int | None: async def replay( - log: WriteHandle, + log: Readable, start: int, stop: int, where: Predicate | None = None, diff --git a/src/streamcast/_metadata.py b/src/streamcast/_metadata.py index 6fbe3fb..f1b0f99 100644 --- a/src/streamcast/_metadata.py +++ b/src/streamcast/_metadata.py @@ -77,11 +77,18 @@ from streamcast import _log, _remote, _schema if TYPE_CHECKING: + from collections.abc import Sequence + import pyarrow as pa from litelink import LogHandle, S3Options, TierStatistics VERSION: Final = 2 -READS: Final = frozenset({1, VERSION}) +RETIRED_VERSION: Final = 3 +"""A retired stream's file. Its own version so that a build that predates +retirement refuses it rather than misreading it: such a build would find the +current log retired, take it for a migration that died, and quietly start the +next log — undoing the retirement without anyone asking.""" +READS: Final = frozenset({1, VERSION, RETIRED_VERSION}) @dataclass(frozen=True, slots=True) @@ -109,6 +116,29 @@ class Entry: """The lowest `streamcast_ts` the log holds, or None if not known yet.""" end_ts: int | None = None """The highest, for a sealed log; None for the live one, whose end moves.""" + sort_by: tuple[str, ...] | None = None + """The columns the log sorts each file by — `()` for none — or None when + not known: a log described before this was recorded, and not on this + disk to ask. With `schema` and `system_schema`, the log's whole shape, + which a restore from a published table needs (litelink 0.10.1).""" + + +@dataclass(frozen=True, slots=True) +class Retirement: + """That the stream is finished (`Stream.retire`), and what reviving it needs. + + Everything a successor log has to agree with, kept here because the + retired log may be on a machine that no longer exists: where it ended, + how it was sorted, and the `streamcast_ts` span its rows cover. + """ + + at: int + """When, in UTC microseconds.""" + end_offset: int + """One past the last row: where a revived stream's next log starts.""" + sort_by: tuple[str, ...] | None = None + start_ts: int | None = None + end_ts: int | None = None @dataclass(frozen=True, slots=True) @@ -122,6 +152,10 @@ class Metadata: """The log being written. Its `end_offset` is None: its end is moving.""" manifest: str | None = None """Where the sealed logs' statistics are (#27), or None until there are any.""" + retirement: Retirement | None = None + """Set by `Stream.retire`: the live log is finished and takes no rows. + + Not `retired`, which names the logs a migration sealed.""" @property def logs(self) -> tuple[Entry, ...]: @@ -160,6 +194,21 @@ def advance( self, sealed_logs=(*self.sealed_logs, sealed), live_log=successor ) + def revive(self, successor: Entry) -> Metadata: + """The retired live log sealed where it ended, `successor` live after + it, and the stream no longer retired.""" + retirement = self.retirement + if retirement is None: + msg = f"stream {self.stream!r} is not retired" + raise ValueError(msg) + + revived = self.advance( + retirement.end_offset, + successor, + span=(retirement.start_ts, retirement.end_ts), + ) + return replace(revived, retirement=None) + def next_name(self) -> str: """`trades-v2`, `trades-v3`, … — named for its place in the sequence. @@ -172,7 +221,9 @@ def to_json(self) -> str: live = self.live_log return json.dumps( { - "streamcast_metadata": VERSION, + "streamcast_metadata": ( + VERSION if self.retirement is None else RETIRED_VERSION + ), "stream": self.stream, "stream_id": self.stream_id, "sealed_logs": [ @@ -185,6 +236,7 @@ def to_json(self) -> str: "end_ts": entry.end_ts, "schema": entry.schema, "system_schema": entry.system_schema, + "sort_by": _listed(entry.sort_by), } for entry in self.sealed_logs ], @@ -195,8 +247,26 @@ def to_json(self) -> str: "start_ts": live.start_ts, "schema": live.schema, "system_schema": live.system_schema, + "sort_by": _listed(live.sort_by), }, "manifest": self.manifest, + **( + {} + if self.retirement is None + else { + "retirement": { + "at": self.retirement.at, + "end_offset": self.retirement.end_offset, + "sort_by": ( + None + if self.retirement.sort_by is None + else list(self.retirement.sort_by) + ), + "start_ts": self.retirement.start_ts, + "end_ts": self.retirement.end_ts, + } + } + ), }, indent=2, ) @@ -226,6 +296,7 @@ def from_json(cls, text: str) -> Metadata: published=entry.get("published"), start_ts=entry.get("start_ts"), end_ts=entry.get("end_ts"), + sort_by=_tupled(entry.get("sort_by")), ) for entry in fields["sealed_logs"] ), @@ -237,13 +308,43 @@ def from_json(cls, text: str) -> Metadata: system_schema=live["system_schema"], published=live.get("published"), start_ts=live.get("start_ts"), + sort_by=_tupled(live.get("sort_by")), ), manifest=fields.get("manifest"), + retirement=( + None + if (retired := fields.get("retirement")) is None + else Retirement( + at=int(retired["at"]), + end_offset=int(retired["end_offset"]), + sort_by=( + None + if retired.get("sort_by") is None + else tuple(retired["sort_by"]) + ), + start_ts=retired.get("start_ts"), + end_ts=retired.get("end_ts"), + ) + ), ) +def _listed(sort_by: tuple[str, ...] | None) -> list[str] | None: + return None if sort_by is None else list(sort_by) + + +def _tupled(sort_by: list[str] | None) -> tuple[str, ...] | None: + return None if sort_by is None else tuple(sort_by) + + def describe( - name: str, start: int, end: int | None, schema: pa.Schema, *, published: str + name: str, + start: int, + end: int | None, + schema: pa.Schema, + *, + published: str, + sort_by: Sequence[str] | None = None, ) -> Entry: """An entry for a log whose table schema is `schema`: both halves of it.""" return Entry( @@ -253,6 +354,22 @@ def describe( _schema.from_arrow(_log.declared(schema)), _schema.from_arrow(_log.system(schema)), published=published, + sort_by=None if sort_by is None else tuple(sort_by), + ) + + +def _system_names(system_schema: dict[str, object]) -> list[str]: + properties = system_schema.get("properties") + return list(properties) if isinstance(properties, dict) else [] + + +def shape(entry: Entry) -> pa.Schema: + """The log's table schema, exactly, rebuilt from its entry: the declared + columns — binary encodings included, which no Iceberg schema keeps — and + the system columns that log has.""" + return _log.with_system( + _schema.to_arrow(entry.schema), + names=_system_names(entry.system_schema), ) @@ -287,6 +404,7 @@ def single(stream: str, log: LogHandle) -> Metadata: None, log.schema, published=log.published, + sort_by=log.sort_by, ) return Metadata( stream=stream, @@ -307,17 +425,27 @@ def complete(metadata: Metadata, log: LogHandle) -> Metadata: root = Path(log.root) def located(entry: Entry) -> Entry: - if entry.published is not None: + if entry.published is not None and entry.sort_by is not None: return entry if entry.name == log.name: - return replace(entry, published=log.published) + return replace( + entry, + published=entry.published or log.published, + sort_by=tuple(log.sort_by), + ) if (root / entry.name).is_dir(): with litelink.open(root, entry.name, read_only=True) as sealed: - return replace(entry, published=sealed.published) + return replace( + entry, + published=entry.published or sealed.published, + sort_by=tuple(sealed.sort_by), + ) - return replace(entry, published=log.published) + # Not on this disk: the prefix is the live log's, and the sort is + # left unknown rather than guessed. + return replace(entry, published=entry.published or log.published) live = located(metadata.live_log) if live.start_ts is None: diff --git a/src/streamcast/_publish.py b/src/streamcast/_publish.py index f8cca47..4750e4b 100644 --- a/src/streamcast/_publish.py +++ b/src/streamcast/_publish.py @@ -366,10 +366,8 @@ class publish: # noqa: N801 — a sibling of `connect`, which mirrors `websocke added here rather than asked of the caller, so one address serves both ends and neither has to know the query string. - **The server must allow it**: `serve(..., publish=True)`. A server that - silently became writable on an upgrade would be a security change nobody - asked for, so the default refuses with `publish_disabled` and says which - setting to change. + Every served stream takes publishers. One that cannot be written — a + retired stream — refuses with 4410 (`StreamRetired`). `max_in_flight` (64) is how many sends `submit` keeps unacknowledged at once. A server that owes one connection fewer replies than that diff --git a/src/streamcast/_server.py b/src/streamcast/_server.py index a03eb7b..eee590f 100644 --- a/src/streamcast/_server.py +++ b/src/streamcast/_server.py @@ -190,7 +190,13 @@ def _supervisors( return [] plan = Maintain() if maintain is True else maintain - logs = [stream.log for stream in routes.values() if stream.log is not None] + # A retired stream's log takes no more passes: it is finished, and + # opened only for reading. + logs = [ + stream.log + for stream in routes.values() + if stream.log is not None and stream.retirement is None + ] # **A name that matches nothing is a raise, not a shrug.** Silently # ignoring it puts the log back in the shared loop — the one thing the @@ -265,7 +271,9 @@ def _sidecars(routes: dict[str, Stream], replicate: bool) -> list[Sidecar]: shipping = [ stream.log for stream in routes.values() - if stream.log is not None and stream.log.config.wal_replication + if stream.log is not None + and stream.retirement is None + and stream.log.config.wal_replication ] if not replicate: if shipping: @@ -328,7 +336,6 @@ def serve( *, maintain: bool | Maintain = True, replicate: bool = False, - publish: bool = False, max_backlog: int | Mapping[str, int] = MAX_BACKLOG, max_inbound: int | Mapping[str, int] = MAX_INBOUND, max_in_flight: int | Mapping[str, int] = MAX_IN_FLIGHT, @@ -398,9 +405,8 @@ def serve( freshness threshold cannot live in a library. `stats="/_internal/streams"` moves it; `stats=False` turns it off. - **On by default, unlike `publish=`**, and the asymmetry is the point. - `publish` grants writes, which nothing else on this port grants. This - discloses strictly LESS than the socket beside it already does: a + **On by default.** It discloses strictly LESS than the socket beside it + already does: a wrong-path connect is answered with `serves=` naming every stream, the greeting carries `end_offset`, and anyone who can reach the port can subscribe and read every row in full. Everything here except the @@ -481,18 +487,8 @@ async def handler(connection: ServerConnection) -> None: return if isinstance(requested, Publish): - # **Opt-in, because a server that silently became writable on an - # upgrade would be a security change nobody asked for.** Every - # other refusal here is about what a caller asked for; this one is - # about what the operator allowed, so it says so rather than - # pretending the stream does not exist. - if not publish: - await connection.close( - Close.BAD_REQUEST, - refusal("publish_disabled"), - ) - return - + # Every stream takes publishers; whether it can be written is the + # stream's to say — a retired one refuses them itself (4410). try: await stream.serve_publisher(connection) except ConnectionClosed: diff --git a/src/streamcast/_stream.py b/src/streamcast/_stream.py index b291807..c756188 100644 --- a/src/streamcast/_stream.py +++ b/src/streamcast/_stream.py @@ -41,7 +41,7 @@ import time from functools import partial from pathlib import Path -from typing import TYPE_CHECKING, Final +from typing import TYPE_CHECKING, Any, Final import litelink from websockets.exceptions import ConnectionClosed @@ -58,7 +58,7 @@ _snapshot, ) from streamcast._codec import compile_codec -from streamcast._errors import NotReplayable, ProtocolError +from streamcast._errors import Close, NotReplayable, ProtocolError, StreamRetired from streamcast._limits import MAX_BACKLOG, MAX_IN_FLIGHT, MAX_INBOUND, MAX_TAIL from streamcast._protocol import ( EARLIEST, @@ -68,6 +68,7 @@ greeting, publish_ack, publish_error, + refusal, ) from streamcast._published import ReadCache from streamcast._stats import Stats @@ -79,7 +80,14 @@ from os import PathLike import pyarrow as pa - from litelink import LogHandle, Row, S3Options, TierStatistics, WriteHandle + from litelink import ( + LocalReadHandle, + LogHandle, + Row, + S3Options, + TierStatistics, + WriteHandle, + ) from streamcast._filter import Predicate, Where from streamcast._transport import Peer @@ -161,6 +169,7 @@ class Stream: "_owned", "_replay_published", "_retired", + "_retirement", "_stream_id", "_s3", "_shape", @@ -180,7 +189,7 @@ def __init__( self, name: str = "", *, - log: WriteHandle | None = None, + log: WriteHandle | LocalReadHandle | None = None, owns_log: bool = False, max_replay: int | None = MAX_REPLAY, floor: int | None = None, @@ -189,6 +198,7 @@ def __init__( schema: Mapping[str, object] | None = None, replay_published: bool = False, group_commit: bool = True, + retirement: _metadata.Retirement | None = None, ) -> None: """Takes an already-open log and builds nothing. See `Stream.new`. @@ -281,10 +291,13 @@ def __init__( # waiting for that to fall below `max_inbound`. See `_room`. self._queued = 0 self._waiting: list[asyncio.Future[None]] = [] + # A retired stream is read-only: its log is opened for reading, and + # `_submit` refuses before anything could reach a writer. + self._retirement = retirement self._writer = ( None - if log is None - else Writer(log, self._deliver, self._fail, group=group_commit) + if log is None or retirement is not None + else Writer(log, self._deliver, self._fail, group=group_commit) # ty: ignore[invalid-argument-type] ) # The shape a subscriber is told at subscribe, built once. None for a # stream with no log: there are no declared columns to publish. Without @@ -414,6 +427,19 @@ def new( replay holds a worker from the `to_thread` pool for its whole scan, and that pool is `min(32, cpu + 4)`. """ + retired = _open_retired(root, name, schema=schema) + if retired is not None: + log, metadata = retired + return cls._read_only( + name, + root, + log, + metadata, + s3_options=s3_options, + max_replay=max_replay, + replay_published=replay_published, + ) + log, metadata = _open_or_create( root, name, @@ -502,6 +528,21 @@ def migrate( which reads every log. A consumer that was caught up when the server stopped resumes exactly at the seam and loses nothing. """ + opened = _open_retired(root, name, schema=schema) + if opened is not None: + # Retired, so served as it is: never migrated onto a new log, which + # is what reviving does (`restore(..., revive=True)`). + log, found = opened + return cls._read_only( + name, + root, + log, + found, + s3_options=s3_options, + max_replay=max_replay, + replay_published=replay_published, + ) + declared = _declaration(schema) metadata = _metadata.load(root, name) current = name if metadata is None else metadata.current.name @@ -575,7 +616,12 @@ def migrate( metadata = metadata.advance( start, _metadata.describe( - new_log.name, start, None, new_log.schema, published=new_log.published + new_log.name, + start, + None, + new_log.schema, + published=new_log.published, + sort_by=new_log.sort_by, ), span=_metadata.span(sealed), ) @@ -631,6 +677,12 @@ def restore( replay_published: bool = False, group_commit: bool = True, max_replay: int | None = MAX_REPLAY, + schema: Mapping[str, object] | None = None, + sort_by: Sequence[str] | None = None, + config: object | None = None, + replica_reserve: int | None = None, + published_reserve: int | None = None, + revive: bool = False, ) -> Stream: """Stand a stream up on a box that never held its log. @@ -640,13 +692,16 @@ def restore( Producer-side failover, and the counterpart to `connect(cursor=)` on the consumer side. `Stream.new` needs the log to be here already; - this rebuilds it from the published table and the replicated WAL, then hands + this rebuilds it from its replicated WAL when there is one, and + otherwise from its published table alone (litelink 0.10), then hands back a stream ready to `serve` and `send` to. **Offsets are fenced, not reissued, and that is what makes the move - invisible to consumers.** litelink burns 2**20 offsets, so the - restored stream resumes above anything the dead machine may have - served. A consumer reconnects with the cursor it already had, sees a + invisible to consumers.** litelink skips `replica_reserve` (2**20) + past what a replica recorded, or `published_reserve` (2**40) past what + the published table says the log issued — rows written after the last + publish are gone with the machine — so the restored stream resumes + above anything the dead machine may have served. A consumer reconnects with the cursor it already had, sees a gap, and carries on — no offset it holds is ever reused for different data, which is the one thing a resume cannot survive. `recv` allows a forward jump for exactly this reason. @@ -662,6 +717,20 @@ def restore( Pass `replay_published=True` to serve history from the published table. Nothing copies published files back down. + **The log's shape comes from the stream's metadata**: its exact + schema — binary encodings and system columns included, which an + Iceberg schema does not keep — and its `sort_by`. litelink checks them + against the replica or the table, and needs them outright for a table + no litelink 0.10 publish stamped. `schema=` (JSON Schema, as `new` + takes) and `sort_by=` are for a stream whose metadata predates + recording them, or to override; `config=` replaces the restored + log's policy, which is otherwise the replica's, or litelink's default. + + **A retired stream is refused** unless `revive=True`, which continues + it on its next log at exactly the retired end — no fence, since + retiring fixed the end — with the retired log's columns and sort and + `config`. See `Stream.retire`. + ⚠️ **Two writers on one log corrupts it.** The fence stops offsets being reused; nothing stops the machine you are failing over FROM if it is still alive. litelink cannot detect a live writer on another @@ -672,13 +741,68 @@ def restore( # **The metadata first**, because it says which log is current. Without # it this would rebuild a migrated stream's FIRST log and serve that as # though nothing had happened since. - metadata = _metadata.fetch(published, name, s3_options) # ty: ignore[invalid-argument-type] + metadata = ( + _metadata.fetch(published, name, s3_options) # ty: ignore[invalid-argument-type] + if _metadata.remote(published) + else None + ) + if metadata is None: + # A local published location has no copy beside its tables; a box + # that holds the stream has the metadata itself. Read for one + # thing only — whether it is retired — so a revive works there. + local = _metadata.load(root, name) + if local is not None and local.retirement is not None: + metadata = local + + if metadata is not None and metadata.retirement is not None: + if not revive: + retirement = metadata.retirement + raise StreamRetired(name, retirement.at, retirement.end_offset) + + return cls._revive( + name, + root, + metadata, + published=published, + config=config, + s3_options=s3_options, + max_replay=max_replay, + replay_published=replay_published, + group_commit=group_commit, + ) + + # The shape, exactly as the stream recorded it, unless the caller + # says otherwise; a part not recorded is left to litelink to read. + current = None if metadata is None else metadata.current + shape = ( + _log.with_system(_declaration(schema)) + if schema is not None + else None + if current is None + else _metadata.shape(current) + ) + if sort_by is None and current is not None: + sort_by = current.sort_by + + # litelink's own reserves unless the caller set them: one owner. + reserves: dict[str, Any] = { + key: value + for key, value in ( + ("replica_reserve", replica_reserve), + ("published_reserve", published_reserve), + ) + if value is not None + } log = litelink.restore( root, - name if metadata is None else metadata.current.name, + name if current is None else current.name, published=published, s3_options=s3_options, # ty: ignore[invalid-argument-type] binary=binary, + schema=shape, + sort_by=sort_by, + config=config, # ty: ignore[invalid-argument-type] + **reserves, ) if metadata is not None: _metadata.save(root, metadata) @@ -695,6 +819,229 @@ def restore( group_commit=group_commit, ) + @classmethod + def retire( + cls, + name: str = "", + *, + root: str | PathLike[str], + s3_options: object | None = None, + ) -> _metadata.Retirement: + """Finish a stream for good: every row published, nothing more taken. + + streamcast.Stream.retire("trades", root="data") + + Run with the server stopped, as `migrate` is. In order: + + 1. **The published table is made complete.** litelink's `retire` + seals every buffered row and publishes the whole log — the + trailing run a plain `publish` holds back included — so the table + every other machine reads ends exactly where the log does. A stream + that merely stops being written leaves that tail on this disk. + 2. **The log refuses writers**, from any handle, for good. + 3. **The metadata records it** — when, where the log ended, how it was + sorted and the `streamcast_ts` span — locally and beside the + tables, with the log's statistics added to the manifest. That is + everything reviving needs, on a box that never held the log. + + From then on the stream is served READ-ONLY: `new` and `migrate` open + it for reading — subscribers replay and catch up, a live view simply + sees nothing new — every send raises `StreamRetired`, and `serve` + refuses a publisher with 4410. Its maintainers do not run. + + **Undone by `restore(..., revive=True)`**, here or on another box: the + stream continues on a new log starting where this one ended. Retiring + published everything first, so reviving loses nothing. + + Safe to run again: an already retired stream returns its record, and + a retire that died partway finishes. + """ + metadata = _metadata.load(root, name) + if metadata is not None and metadata.retirement is not None: + return metadata.retirement + + current = name if metadata is None else metadata.current.name + finished = False + try: + log: LogHandle = litelink.open(root, current) + except litelink.RetiredError: + # litelink retired it — a retire that died before the metadata + # was written. Only the record is left to make. + log = litelink.open(root, current, read_only=True) + finished = True + except FileNotFoundError: + msg = f"there is no stream {name!r} at {root} to retire" + raise FileNotFoundError(msg) from None + + try: + if metadata is None: + metadata = _metadata.single(name, log) + + if not finished and isinstance(log, litelink.WriteHandle): + if log.config.wal_replication: + with _replicate.retiring(log): + log.retire() + else: + log.retire() + + # The whole log, now all of it published. + statistics = log.column_statistics() + end = log.end_offset() + schema = log.schema + sort_by = tuple(log.sort_by) or None + published = log.published + finally: + log.close() + + start_ts, end_ts = _metadata.span(statistics) + live = metadata.live_log + retirement = _metadata.Retirement( + at=time.time_ns() // 1000, + end_offset=end, + sort_by=sort_by, + start_ts=live.start_ts if live.start_ts is not None else start_ts, + end_ts=end_ts, + ) + # Its statistics join the manifest now, while they can be read: a + # revived stream seals this log without it, maybe on another box. + manifest = _manifest.extend( + _manifest.load(root, name), + _manifest.entry(live.name, live.start_offset, end, schema, statistics), + ) + metadata = dataclasses.replace( + metadata, manifest=_manifest.name(name), retirement=retirement + ) + # The manifest first, then the metadata that points at it (#27). + _manifest.save(root, name, manifest) + if _metadata.remote(published): + _manifest.publish(published, name, manifest, s3_options) # ty: ignore[invalid-argument-type] + + _metadata.save(root, metadata) + if _metadata.remote(published): + _metadata.publish(metadata, published, s3_options) # ty: ignore[invalid-argument-type] + + return retirement + + @classmethod + def _read_only( + cls, + name: str, + root: str | PathLike[str], + log: LogHandle, + metadata: _metadata.Metadata, + *, + s3_options: object | None, + max_replay: int | None, + replay_published: bool, + ) -> Stream: + """A retired stream, opened to be read: it replays, and refuses sends. + + Replays from the published table whatever `replay_published` says: + retiring evicted staging and emptied the buffer, so the table is the + log's only copy, and a replay of the local tiers would find nothing. + `replay_published` is taken, and ignored, so the call that opens a + stream reads the same retired or not. + """ + del replay_published + return cls( + name, + log=log, # ty: ignore[invalid-argument-type] + owns_log=True, + max_replay=max_replay, + floor=_floor(metadata), + retired=_retired(root, metadata), + s3_options=s3_options, # ty: ignore[invalid-argument-type] + replay_published=True, + retirement=metadata.retirement, + ) + + @classmethod + def _revive( + cls, + name: str, + root: str | PathLike[str], + metadata: _metadata.Metadata, + *, + published: str, + config: object | None, + s3_options: object | None, + max_replay: int | None, + replay_published: bool, + group_commit: bool, + ) -> Stream: + """A retired stream continued on a new log, where the retired one ended. + + The retired log is never written again — litelink refuses — so the + stream rotates, as `migrate` does: the next name, the same columns + and sort, starting at exactly the retired end, so the offsets stay + one dense sequence. Everything it needs is in the metadata, so the + retired log need not be on this box. + """ + retirement = metadata.retirement + assert retirement is not None # the caller checked + current = metadata.current + location = current.published or published + successor = metadata.next_name() + declared = _declaration(current.schema) + if (Path(root) / successor).exists(): + # **A revive that died after creating the log** and before saving + # the metadata: carried on from, if it is what that run made. + log = litelink.open(root, successor) + if ( + log.end_offset() != retirement.end_offset + or _log.lowest(log) is not None + ): + log.close() + msg = ( + f"{Path(root) / successor} exists and is not an empty log at " + f"offset {retirement.end_offset}, where {current.name!r} was " + f"retired; nothing was revived" + ) + raise FileExistsError(msg) + + else: + log = litelink.new( + root, + successor, + schema=_log.with_system(declared), + sort_by=retirement.sort_by, + config=config, # ty: ignore[invalid-argument-type] + published=None if not _metadata.remote(location) else location, + s3_options=s3_options, # ty: ignore[invalid-argument-type] + start_offset=retirement.end_offset, + ) + + revived = metadata.revive( + _metadata.describe( + log.name, + retirement.end_offset, + None, + log.schema, + published=log.published, + sort_by=log.sort_by, + ) + ) + try: + _metadata.save(root, revived) + if _metadata.remote(log.published): + _metadata.publish(revived, log.published, s3_options) # ty: ignore[invalid-argument-type] + + except BaseException: + log.close() + raise + + return cls( + name, + log=log, + owns_log=True, + max_replay=max_replay, + floor=_floor(revived), + retired=_retired(root, revived), + s3_options=s3_options, # ty: ignore[invalid-argument-type] + replay_published=replay_published, + group_commit=group_commit, + ) + def __repr__(self) -> str: durable = "durable" if self._log is not None else "live-only" return ( @@ -915,8 +1262,11 @@ def log(self) -> WriteHandle | None: Closed by `aclose` when this `Stream` created it from `root=`+`schema=`, and never when it was passed in — that one is the caller's. + + A retired stream's is opened read-only (`litelink.LocalReadHandle`): + reads work, writes are refused by litelink. """ - return self._log + return self._log # ty: ignore[invalid-return-type] @property def schema(self) -> dict[str, object] | None: @@ -927,6 +1277,16 @@ def schema(self) -> dict[str, object] | None: """ return None if self._shape is None else dict(self._shape) + @property + def retirement(self) -> _metadata.Retirement | None: + """When the stream was retired and where it ended, or None if it is not. + + A retired stream is served read-only: it replays, snapshots and + catches up, and refuses every send with `StreamRetired`. See + `Stream.retire`. + """ + return self._retirement + @property def durable(self) -> bool: """Whether a log is attached, and so whether offsets survive a restart.""" @@ -1113,6 +1473,11 @@ def _submit(self, batch: list[Row]) -> asyncio.Future[list[int | None]]: transaction and its rows commit together, so per-row values would imply a precision the commit does not have. """ + if self._retirement is not None: + raise StreamRetired( + self._name, self._retirement.at, self._retirement.end_offset + ) + now = time.time_ns() codec = self._codec if codec.check is not None: @@ -1339,7 +1704,22 @@ async def serve_publisher(self, connection: Peer) -> None: because that is what the local call does: `send` raises, the caller catches it, and the next call works. Closing would make one bad row cost every good one behind it. + + **A retired stream refuses publishers**, with 4410 (`Close.RETIRED`): + its log is open read-only and takes nothing. The stream decides, from + what it is; subscribers still read its history. """ + if self._retirement is not None: + await connection.close( + Close.RETIRED, + refusal( + "stream_retired", + at=self._retirement.at, + end_offset=self._retirement.end_offset, + ), + ) + return + await connection.send( greeting( stream=self._name, @@ -1516,7 +1896,7 @@ async def serve_subscriber( # number of live scans against one log. await replay.aclose() - async def _resolve(self, requested: int | None) -> tuple[WriteHandle, int] | None: + async def _resolve(self, requested: int | None) -> tuple[LogHandle, int] | None: """The log and offset a replay should start at, or None for live-only. Every refusal here is cheap and answerable before a scan is opened. @@ -1602,7 +1982,7 @@ async def _resolve(self, requested: int | None) -> tuple[WriteHandle, int] | Non async def _replay_from( self, - log: WriteHandle, + log: LogHandle, start: int, frontier: int, where: Predicate | None = None, @@ -1682,6 +2062,30 @@ async def _replay_from( return _prepend(first, stream) +def _open_retired( + root: str | PathLike[str], name: str, *, schema: Mapping[str, object] +) -> tuple[LogHandle, _metadata.Metadata] | None: + """The retired stream's log opened read-only, and its metadata — or None + if the stream is not retired. The declared schema is checked as for a + live stream: a declaration that disagreed would be silently ignored.""" + metadata = _metadata.load(root, name) + if metadata is None or metadata.retirement is None: + return None + + declared = _declaration(schema) + log = litelink.open(root, metadata.current.name, read_only=True) + if list(_log.declared(log.schema)) != list(declared): + found = list(_log.columns(log)) + log.close() + msg = ( + f"the retired stream {name!r} has columns {found}, and this call " + f"declares {declared.names}; declare the columns it has" + ) + raise ValueError(msg) + + return log, metadata + + def _open_or_create( root: str | PathLike[str], name: str, diff --git a/src/streamcast/asgi.py b/src/streamcast/asgi.py index a17fc6f..a6c1e86 100644 --- a/src/streamcast/asgi.py +++ b/src/streamcast/asgi.py @@ -62,7 +62,12 @@ async def lifespan(app): from streamcast._errors import Close, NotReplayable, ProtocolError from streamcast._limits import MAX_BACKLOG, MAX_IN_FLIGHT, MAX_INBOUND, _bound from streamcast._protocol import Publish, parse_subscribe, refusal -from streamcast._server import _DETAIL_CHARS, _routes, _sidecars, _supervisors +from streamcast._server import ( + _DETAIL_CHARS, + _routes, + _sidecars, + _supervisors, +) if TYPE_CHECKING: from collections.abc import AsyncIterator, Iterable, Mapping @@ -278,7 +283,6 @@ class _Mounted: __slots__ = ( "_children", "_maintain", - "_publish", "_sidecars", "_started", "_streams", @@ -290,7 +294,6 @@ def __init__( *, maintain: bool | Maintain = True, replicate: bool = False, - publish: bool = False, max_backlog: int | Mapping[str, int] = MAX_BACKLOG, max_inbound: int | Mapping[str, int] = MAX_INBOUND, max_in_flight: int | Mapping[str, int] = MAX_IN_FLIGHT, @@ -309,13 +312,10 @@ def __init__( # as `app = asgi(...)` — should not do. self._sidecars: list[_Child] = [*_sidecars(self._streams, replicate)] self._children: list[_Child] = [] - self._publish = publish self._started = False def __repr__(self) -> str: - return ( - f"" - ) + return f"" def _start(self) -> None: if self._started: @@ -448,10 +448,6 @@ async def _dispatch(self, peer: _Peer, target: str) -> None: return if isinstance(requested, Publish): - if not self._publish: - await peer.close(Close.BAD_REQUEST, refusal("publish_disabled")) - return - await stream.serve_publisher(peer) return @@ -476,7 +472,6 @@ def asgi( *, maintain: bool | Maintain = True, replicate: bool = False, - publish: bool = False, max_backlog: int | Mapping[str, int] = MAX_BACKLOG, max_inbound: int | Mapping[str, int] = MAX_INBOUND, max_in_flight: int | Mapping[str, int] = MAX_IN_FLIGHT, @@ -512,7 +507,6 @@ def asgi( streams, maintain=maintain, replicate=replicate, - publish=publish, max_backlog=max_backlog, max_inbound=max_inbound, max_in_flight=max_in_flight, diff --git a/tests/test_asgi.py b/tests/test_asgi.py index afdecd8..d943a31 100644 --- a/tests/test_asgi.py +++ b/tests/test_asgi.py @@ -225,15 +225,14 @@ async def test_an_unreplayable_offset_is_refused(self, log): with pytest.raises(streamcast.NotReplayable): await streamcast.connect(f"{base}/trades", offset=9_999) - async def test_publishing_is_off_unless_the_mount_allows_it(self, log): - """The same opt-in `serve` has, and for the same reason.""" + async def test_every_mounted_stream_takes_publishers(self, log): + """No switch, as `serve` has none: a stream that can be written takes + publishers, and one that cannot (a retired one) refuses them itself.""" stream = streamcast.Stream("trades", log=log) app = asgi(stream, maintain=False, replicate=False) async with running(app) as base: - with pytest.raises(streamcast.ProtocolError) as raised: - await streamcast.publish(f"{base}/trades") - - assert "does not accept publishers" in str(raised.value) + async with streamcast.publish(f"{base}/trades") as producer: + assert await producer.send(trade(0)) == 1 async def test_an_http_request_gets_a_404_rather_than_a_traceback(self, log): """A routing mistake in the host app, answered rather than raised.""" @@ -364,7 +363,7 @@ async def test_a_publisher_appends_through_the_mount(self, log): """`serve_publisher` iterates the connection, so this exercises `__aiter__` — the one method a subscription never touches.""" stream = streamcast.Stream("trades", log=log) - app = asgi(stream, maintain=False, replicate=False, publish=True) + app = asgi(stream, maintain=False, replicate=False) async with running(app) as base: async with streamcast.connect(f"{base}/trades") as sub: async with streamcast.publish(f"{base}/trades") as producer: @@ -377,7 +376,7 @@ async def test_a_publisher_appends_through_the_mount(self, log): async def test_send_many_stays_one_transaction(self, log): stream = streamcast.Stream("trades", log=log) - app = asgi(stream, maintain=False, replicate=False, publish=True) + app = asgi(stream, maintain=False, replicate=False) async with running(app) as base: async with streamcast.publish(f"{base}/trades") as producer: offsets = await producer.send_many([trade(i) for i in range(20)]) @@ -386,7 +385,7 @@ async def test_send_many_stays_one_transaction(self, log): async def test_a_rejected_row_answers_and_keeps_the_connection(self, log): stream = streamcast.Stream("trades", log=log) - app = asgi(stream, maintain=False, replicate=False, publish=True) + app = asgi(stream, maintain=False, replicate=False) async with running(app) as base: async with streamcast.publish(f"{base}/trades") as producer: with pytest.raises(streamcast.Rejected) as raised: diff --git a/tests/test_demos.py b/tests/test_demos.py index a0661a1..c7dc19c 100644 --- a/tests/test_demos.py +++ b/tests/test_demos.py @@ -101,9 +101,7 @@ async def test_a_producer_publishes_through_it_to_a_subscriber(self, tmp_path): stream = streamcast.Stream.new( "orders", root=tmp_path, schema=broker.schema("examples/book/schema.json") ) - async with streamcast.serve( - stream, "127.0.0.1", 0, publish=True, maintain=False - ) as server: + async with streamcast.serve(stream, "127.0.0.1", 0, maintain=False) as server: uri = f"ws://127.0.0.1:{server.sockets[0].getsockname()[1]}/orders" async with streamcast.publish(uri) as publication: await publication.send( diff --git a/tests/test_live_schema.py b/tests/test_live_schema.py index c86cde4..93c9fda 100644 --- a/tests/test_live_schema.py +++ b/tests/test_live_schema.py @@ -162,7 +162,7 @@ async def test_binary_columns_work_without_a_log(self, serve): async def test_a_remote_publisher_is_told_why(self, serve): stream = live() - async with serve(stream, publish=True) as uri: + async with serve(stream) as uri: async with streamcast.publish(uri) as producer: with pytest.raises(Exception, match="venue"): # noqa: B017, PT011 await producer.send({**GOOD, "venue": "x"}) diff --git a/tests/test_node.py b/tests/test_node.py index 5ec1433..76d8af8 100644 --- a/tests/test_node.py +++ b/tests/test_node.py @@ -86,7 +86,7 @@ async def test_a_browser_websocket_reads_and_writes_a_stream(tmp_path, serve, no "nested": {"m": {"b": "0102"}}, } - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: process = await asyncio.create_subprocess_exec( node, str(CLIENT), diff --git a/tests/test_otel_example.py b/tests/test_otel_example.py index 9b3d91e..dd57a29 100644 --- a/tests/test_otel_example.py +++ b/tests/test_otel_example.py @@ -150,9 +150,7 @@ async def test_a_broker_restart_costs_only_the_batches_sent_while_it_was_down( async def broker(port: int = 0): # noqa: ANN202 stream = streamcast.Stream.new("t", root=tmp_path, schema=schema) - return await streamcast.serve( - stream, "127.0.0.1", port, publish=True, maintain=False - ) + return await streamcast.serve(stream, "127.0.0.1", port, maintain=False) server = await broker() port = server.sockets[0].getsockname()[1] diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index e598c7b..790e81c 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -54,7 +54,7 @@ class TestOrder: async def test_submitted_rows_are_acknowledged_in_the_order_sent( self, stream, serve ): - async with serve(stream, maintain=False, publish=True) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri) as producer: futures = [await producer.submit({"i": n}) for n in range(500)] offsets = await asyncio.gather(*futures) @@ -66,7 +66,7 @@ async def test_one_publishers_rows_group_into_one_commit( ): """What pipelining is for: a single publisher keeps the writer fed.""" held, commits = hold_commits(stream, monkeypatch) - async with serve(stream, maintain=False, publish=True) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri) as producer: first = await producer.submit({"i": 0}) await asyncio.sleep(0.05) # the server has it; its commit is held @@ -83,7 +83,7 @@ async def test_a_full_window_holds_the_next_submit( self, stream, serve, monkeypatch ): held, _commits = hold_commits(stream, monkeypatch) - async with serve(stream, maintain=False, publish=True) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri, max_in_flight=4) as producer: inflight = [await producer.submit({"i": n}) for n in range(4)] fifth = asyncio.ensure_future(producer.submit({"i": 4})) @@ -101,7 +101,7 @@ async def test_the_servers_bound_holds_a_client_allowed_more( ): """The server's `max_in_flight`: past it, the server stops reading.""" held, commits = hold_commits(stream, monkeypatch) - async with serve(stream, maintain=False, publish=True, max_in_flight=2) as uri: + async with serve(stream, maintain=False, max_in_flight=2) as uri: async with streamcast.publish(uri, max_in_flight=10) as producer: futures = [await producer.submit({"i": n}) for n in range(10)] await asyncio.sleep(0.1) @@ -126,7 +126,7 @@ def test_the_window_must_allow_a_send(self): class TestFailures: async def test_a_refused_row_fails_in_its_place(self, stream, serve): - async with serve(stream, maintain=False, publish=True) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri) as producer: good = await producer.submit({"i": 1}) bad = await producer.submit({"i": "not an integer"}) @@ -139,7 +139,7 @@ async def test_a_refused_row_fails_in_its_place(self, stream, serve): assert await after == 2 async def test_close_waits_for_what_is_in_flight(self, stream, serve): - async with serve(stream, maintain=False, publish=True) as uri: + async with serve(stream, maintain=False) as uri: producer = await streamcast.publish(uri) futures = [await producer.submit({"i": n}) for n in range(50)] await producer.close() @@ -151,7 +151,7 @@ async def test_a_dropped_connection_fails_every_send_in_flight( ): held, _commits = hold_commits(stream, monkeypatch) server_closed = False - async with serve(stream, maintain=False, publish=True) as uri: + async with serve(stream, maintain=False) as uri: producer = await streamcast.publish(uri) futures = [await producer.submit({"i": n}) for n in range(5)] await asyncio.sleep(0.05) diff --git a/tests/test_publish.py b/tests/test_publish.py index 6663e5d..bc14533 100644 --- a/tests/test_publish.py +++ b/tests/test_publish.py @@ -23,7 +23,7 @@ class TestItPublishes: async def test_a_remote_publisher_appends_and_subscribers_see_it(self, serve, log): """The headline, end to end: publish over a socket, receive over another.""" stream = streamcast.Stream("trades", log=log) - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.connect(uri) as sub: async with streamcast.publish(uri) as producer: offset = await producer.send(trade(0)) @@ -37,7 +37,7 @@ async def test_a_remote_publisher_appends_and_subscribers_see_it(self, serve, lo async def test_send_many_is_one_transaction(self, serve, log): """Contiguous offsets, because it is the same call `Stream.send_many` is.""" stream = streamcast.Stream("trades", log=log) - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri) as producer: offsets = await producer.send_many([trade(i) for i in range(20)]) @@ -45,7 +45,7 @@ async def test_send_many_is_one_transaction(self, serve, log): async def test_an_empty_batch_sends_nothing(self, serve, log): stream = streamcast.Stream("trades", log=log) - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri) as producer: assert await producer.send_many([]) == [] @@ -58,7 +58,7 @@ async def test_the_greeting_tells_a_publisher_the_schema(self, serve, log): it is about to send against the shape the server will accept. """ stream = streamcast.Stream("trades", log=log) - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri) as producer: shape = producer.info.schema assert isinstance(shape, dict) @@ -82,7 +82,7 @@ async def test_two_publishers_never_share_an_offset(self, serve, log): this is the behaviour that guard protects. """ stream = streamcast.Stream("trades", log=log) - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async def publisher(tag: int, count: int) -> list[int | None]: async with streamcast.publish(uri) as producer: @@ -105,7 +105,7 @@ async def test_a_batch_stays_contiguous_under_a_racing_publisher(self, serve, lo the log could see half a group. """ stream = streamcast.Stream("trades", log=log) - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async def batches() -> list[list[int | None]]: out = [] @@ -134,29 +134,34 @@ async def singles() -> None: class TestWhatItRefuses: - async def test_publishing_is_off_unless_the_server_allows_it(self, serve, log): - """A server must not become writable on an upgrade. - - Every other refusal is about what a caller asked for. This one is - about what the operator allowed, so it says so rather than pretending - the stream is missing. - """ + async def test_every_served_stream_takes_publishers(self, serve, log): + """No switch on `serve`: whether a stream can be written is the + stream's to say — a retired one refuses publishers (`test_retire`) — + and the same port serves its readers.""" stream = streamcast.Stream("trades", log=log) - async with serve(stream, maintain=False) as uri: # no publish=True - with pytest.raises(streamcast.ProtocolError) as raised: - await streamcast.publish(uri) - - # The wire carries `publish_disabled`; the sentence is built by - # the client, and names the setting to change. - assert "does not accept publishers" in str(raised.value) - assert "publish=True" in str(raised.value) - - # And the stream still serves readers, which is the point of the - # refusal being narrow. + async with serve(stream, maintain=False) as uri: async with streamcast.connect(uri) as sub: - await stream.send(trade(0)) + async with streamcast.publish(uri) as producer: + assert await producer.send(trade(0)) == 1 + assert (await sub.recv())[0] == 1 + def test_an_older_servers_refusal_still_names_the_setting(self): + """A server from before every stream took publishers sends + `publish_disabled`; the client still says what to change.""" + from websockets.exceptions import ConnectionClosed # noqa: PLC0415 + from websockets.frames import Close as CloseFrame # noqa: PLC0415 + + from streamcast._client import _refusal # noqa: PLC0415 + from streamcast._protocol import refusal # noqa: PLC0415 + + closed = ConnectionClosed( + CloseFrame(streamcast.Close.BAD_REQUEST, refusal("publish_disabled")), None + ) + error = _refusal(closed, stream="trades", offset=None) + assert isinstance(error, streamcast.ProtocolError) + assert "publish=True" in str(error) + async def test_a_row_the_schema_refuses_is_rejected_not_committed(self, serve, log): """litelink's own message, and the connection survives it. @@ -165,7 +170,7 @@ async def test_a_row_the_schema_refuses_is_rejected_not_committed(self, serve, l would make one bad row cost every good one behind it. """ stream = streamcast.Stream("trades", log=log) - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri) as producer: with pytest.raises(streamcast.Rejected) as raised: await producer.send({"event_ts": "not a number", "price": 1.0}) @@ -178,7 +183,7 @@ async def test_a_row_the_schema_refuses_is_rejected_not_committed(self, serve, l async def test_a_rejected_batch_commits_none_of_itself(self, serve, log): stream = streamcast.Stream("trades", log=log) - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri) as producer: rows: list[dict[str, object]] = [ dict(trade(0)), @@ -193,7 +198,7 @@ async def test_a_rejected_batch_commits_none_of_itself(self, serve, log): async def test_a_frame_that_is_not_a_row_is_refused(self, serve, log): """Answered rather than closed, like any other bad publish.""" stream = streamcast.Stream("trades", log=log) - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri) as producer: await producer.connection.send('"just a string"') with pytest.raises(streamcast.Rejected, match="not a row"): @@ -205,7 +210,7 @@ async def test_publishing_to_a_stream_that_is_not_served_is_refused( self, serve, log ): stream = streamcast.Stream("trades", log=log) - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: wrong = uri.replace("/trades", "/quotes") with pytest.raises(streamcast.StreamNotFound): await streamcast.publish(wrong) @@ -222,7 +227,7 @@ async def test_it_records_the_last_acknowledged_offset(self, serve, log, tmp_pat """ cursor = tmp_path / "producer.offset" stream = streamcast.Stream("trades", log=log) - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri, cursor=cursor) as producer: assert producer.resumed_from is None, "nothing to resume yet" for i in range(5): @@ -237,7 +242,7 @@ async def test_a_restart_resumes_from_what_it_was_acked_for( """The whole point: a new process knows where it got to.""" cursor = tmp_path / "producer.offset" stream = streamcast.Stream("trades", log=log) - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri, cursor=cursor) as first: for i in range(5): await first.send(trade(i)) @@ -270,7 +275,7 @@ async def test_the_recovery_replay_recovers_the_sequence(self, serve, tmp_path): with handle: stream = streamcast.Stream("trades", log=handle) cursor = tmp_path / "producer.offset" - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri, cursor=cursor) as producer: for seq in range(5): await producer.send({"publisher": "a", "seq": seq}) diff --git a/tests/test_restore_shape.py b/tests/test_restore_shape.py new file mode 100644 index 0000000..b8eb5b6 --- /dev/null +++ b/tests/test_restore_shape.py @@ -0,0 +1,204 @@ +"""A restored log's shape: the schema and `sort_by` the stream recorded, exactly. + +litelink 0.10.1 checks a restore's schema and `sort_by` against the replica or +the published table, and needs them outright for a table no 0.10 publish +stamped — from which an Iceberg schema alone cannot rebuild the log: it keeps +no Arrow field metadata, where a binary column's wire encoding lives. So the +stream's metadata records both, and `Stream.restore` passes them on. +""" + +from __future__ import annotations + +import json +import shutil +from typing import Any + +import litelink +import pytest + +import streamcast +from streamcast import _metadata + +SCHEMA: dict[str, Any] = { + "type": "object", + "properties": { + "event_ts": {"type": "integer"}, + "trace": {"type": "string", "contentEncoding": "base16", "format": "bytes16"}, + "price": {"type": ["number", "null"]}, + }, + "required": ["event_ts", "trace"], +} +SORT_BY = ("event_ts",) + + +def row(i: int) -> dict[str, object]: + return {"event_ts": 1_790_000_000_000_000 + i, "trace": bytes(16), "price": 1.5} + + +class TestTheMetadataRecordsTheSort: + async def test_a_new_stream_records_it(self, tmp_path): + stream = streamcast.Stream.new( + "t", root=tmp_path, schema=SCHEMA, sort_by=SORT_BY + ) + try: + stream.ensure_metadata() + finally: + await stream.aclose() + + metadata = _metadata.load(tmp_path, "t") + assert metadata is not None + assert metadata.live_log.sort_by == SORT_BY + assert json.loads(_metadata.path(tmp_path, "t").read_text())["live_log"][ + "sort_by" + ] == list(SORT_BY) + + async def test_a_file_from_before_it_gets_it_at_the_next_serve(self, tmp_path): + stream = streamcast.Stream.new( + "t", root=tmp_path, schema=SCHEMA, sort_by=SORT_BY + ) + try: + stream.ensure_metadata() + path = _metadata.path(tmp_path, "t") + older = json.loads(path.read_text()) + del older["live_log"]["sort_by"] + path.write_text(json.dumps(older)) + before = _metadata.load(tmp_path, "t") + assert before is not None + assert before.live_log.sort_by is None + + stream.ensure_metadata() # what `serve` does at start + finally: + await stream.aclose() + + after = _metadata.load(tmp_path, "t") + assert after is not None + assert after.live_log.sort_by == SORT_BY + + async def test_the_recorded_shape_is_the_logs_exactly(self, tmp_path): + """Binary encodings and system columns included: what Iceberg loses.""" + stream = streamcast.Stream.new( + "t", root=tmp_path, schema=SCHEMA, sort_by=SORT_BY + ) + try: + stream.ensure_metadata() + assert stream.log is not None + actual = stream.log.schema + finally: + await stream.aclose() + + metadata = _metadata.load(tmp_path, "t") + assert metadata is not None + rebuilt = _metadata.shape(metadata.live_log) + assert rebuilt.equals(actual, check_metadata=True) + + +@pytest.mark.replication +class TestRestoringIt: + async def _lost_box(self, root, bucket, s3) -> None: + """A stream with no WAL replication, published, and then its box gone.""" + stream = streamcast.Stream.new( + "t", + root=root, + schema=SCHEMA, + sort_by=SORT_BY, + published=bucket, + s3_options=s3, + ) + try: + await stream.send_many([row(i) for i in range(5)]) + assert stream.log is not None + while stream.log.seal(flush=True) is not None: + pass + + stream.log.publish(flush=True) + stream.ensure_metadata() + finally: + await stream.aclose() + + shutil.rmtree(root) + + async def test_the_shape_is_passed_from_the_metadata( + self, tmp_path, s3, bucket, monkeypatch + ): + """litelink checks a given shape against the table EXACTLY: this + passes only if the one rebuilt from the metadata is the log's. Both + halves are recorded as passed, since a stamped table would supply them + itself and hide one that was not.""" + await self._lost_box(tmp_path / "a", bucket, s3) + passed: dict[str, object] = {} + real = litelink.restore + + def recording(*args, **kwargs): # noqa: ANN002, ANN003, ANN202 + passed.update(kwargs) + return real(*args, **kwargs) + + monkeypatch.setattr(litelink, "restore", recording) + restored = streamcast.Stream.restore( + "t", root=tmp_path / "b", published=bucket, s3_options=s3 + ) + try: + assert restored.log is not None + assert restored.log.sort_by == SORT_BY + trace = restored.log.schema.field("trace") + assert trace.metadata == {b"streamcast.encoding": b"base16"} + assert await restored.send(row(9)) is not None + finally: + await restored.aclose() + + assert passed["sort_by"] == SORT_BY + schema = passed["schema"] + assert schema is not None + assert schema.field("trace").metadata == {b"streamcast.encoding": b"base16"} # ty: ignore[unresolved-attribute] + + async def test_a_schema_that_disagrees_is_refused(self, tmp_path, s3, bucket): + await self._lost_box(tmp_path / "a", bucket, s3) + retyped = { + **SCHEMA, + "properties": { + **SCHEMA["properties"], + "price": {"type": ["string", "null"]}, + }, + } + with pytest.raises(Exception, match="price"): + streamcast.Stream.restore( + "t", + root=tmp_path / "b", + published=bucket, + s3_options=s3, + schema=retyped, + ) + + async def test_config_replaces_the_restored_logs(self, tmp_path, s3, bucket): + await self._lost_box(tmp_path / "a", bucket, s3) + config = litelink.LogConfig(target_seal_size=123_456) + restored = streamcast.Stream.restore( + "t", root=tmp_path / "b", published=bucket, s3_options=s3, config=config + ) + try: + assert restored.log is not None + assert restored.log.config.target_seal_size == 123_456 + finally: + await restored.aclose() + + +class TestReviving: + async def test_config_reaches_the_new_log(self, tmp_path): + stream = streamcast.Stream.new("t", root=tmp_path, schema=SCHEMA) + try: + await stream.send(row(0)) + finally: + await stream.aclose() + + streamcast.Stream.retire("t", root=tmp_path) + revived = streamcast.Stream.restore( + "t", + root=tmp_path, + published=tmp_path.as_uri(), + revive=True, + config=litelink.LogConfig(target_seal_size=123_456), + ) + try: + assert revived.log is not None + assert revived.log.config.target_seal_size == 123_456 + finally: + await revived.aclose() diff --git a/tests/test_retire.py b/tests/test_retire.py new file mode 100644 index 0000000..34de958 --- /dev/null +++ b/tests/test_retire.py @@ -0,0 +1,264 @@ +"""`Stream.retire`: a stream finished for good, served read-only, and revived. + +A stream that merely stops being written leaves its end on the box — the +trailing run `publish` holds back for compaction — and nothing refuses a late +write or says it is finished. Retiring publishes everything, refuses every +write, and records it in the metadata; `restore(..., revive=True)` undoes it on +a new log, here or on another box. +""" + +from __future__ import annotations + +import shutil +from pathlib import Path + +import litelink +import pytest + +import streamcast +from streamcast import _metadata + +SCHEMA = { + "type": "object", + "properties": {"i": {"type": "integer"}}, + "required": ["i"], +} + + +async def produce(root: Path, count: int = 7, **kwargs: object) -> None: + """A stream with `count` rows, written and then left: the server stopped.""" + stream = streamcast.Stream.new("t", root=root, schema=SCHEMA, **kwargs) # ty: ignore[invalid-argument-type] + try: + await stream.send_many([{"i": i} for i in range(count)]) + finally: + await stream.aclose() + + +class TestRetiring: + async def test_it_publishes_what_stopping_would_leave_behind(self, tmp_path): + await produce(tmp_path) + with litelink.open(tmp_path, "t", read_only=True) as log: + # A plain stop: the rows are on this disk, none of them published. + assert log.published_through() == 0 + + retirement = streamcast.Stream.retire("t", root=tmp_path) + + assert retirement.end_offset == 8 + with litelink.open(tmp_path, "t", read_only=True) as log: + assert log.published_through() == 7, "every row, the tail included" + + async def test_the_metadata_records_it_and_old_builds_refuse_it(self, tmp_path): + await produce(tmp_path) + retirement = streamcast.Stream.retire("t", root=tmp_path) + + metadata = _metadata.load(tmp_path, "t") + assert metadata is not None + assert metadata.retirement == retirement + assert retirement.start_ts is not None + assert retirement.end_ts is not None + # Version 3, so a build that predates retirement refuses the file + # rather than taking the retired log for a migration that died. + assert '"streamcast_metadata": 3' in _metadata.path(tmp_path, "t").read_text() + + async def test_it_is_safe_to_run_again(self, tmp_path): + await produce(tmp_path) + first = streamcast.Stream.retire("t", root=tmp_path) + assert streamcast.Stream.retire("t", root=tmp_path) == first + + async def test_a_retire_that_died_after_litelinks_step_finishes(self, tmp_path): + """litelink retired the log, then the metadata was never written.""" + await produce(tmp_path) + with litelink.open(tmp_path, "t") as log: + log.retire() + + retirement = streamcast.Stream.retire("t", root=tmp_path) + assert retirement.end_offset == 8 + + def test_a_stream_that_is_not_there_is_refused(self, tmp_path): + with pytest.raises(FileNotFoundError, match="no stream 't'"): + streamcast.Stream.retire("t", root=tmp_path) + + +class TestARetiredStreamIsReadOnly: + async def test_it_opens_read_only_and_refuses_a_send(self, tmp_path): + await produce(tmp_path) + streamcast.Stream.retire("t", root=tmp_path) + + for opened in ( + streamcast.Stream.new("t", root=tmp_path, schema=SCHEMA), + streamcast.Stream.migrate("t", root=tmp_path, schema=SCHEMA), + ): + try: + assert opened.retirement is not None + assert opened.end_offset == 8 + with pytest.raises(streamcast.StreamRetired, match="revive=True"): + await opened.send({"i": 99}) + finally: + await opened.aclose() + + async def test_it_is_served_for_reading_and_refuses_a_publisher( + self, tmp_path, serve + ): + await produce(tmp_path) + streamcast.Stream.retire("t", root=tmp_path) + stream = streamcast.Stream.new("t", root=tmp_path, schema=SCHEMA) + + async with serve(stream) as uri: + # The history replays — from the published table, the only copy. + async with streamcast.connect(uri, offset=streamcast.EARLIEST) as sub: + assert [(await sub.recv())[0] for _ in range(7)] == list(range(1, 8)) + + # A publisher is refused by the stream, with when and where. + with pytest.raises(streamcast.StreamRetired) as raised: + async with streamcast.publish(uri): + pass + + assert raised.value.end_offset == 8 + assert raised.value.at is not None + + async def test_it_is_not_maintained(self, tmp_path, serve, monkeypatch): + from streamcast import _server # noqa: PLC0415 + + await produce(tmp_path) + streamcast.Stream.retire("t", root=tmp_path) + stream = streamcast.Stream.new("t", root=tmp_path, schema=SCHEMA) + planned: list[list[str]] = [] + real = _server._supervisors # noqa: SLF001 + + def recording(routes, maintain): # noqa: ANN001, ANN202 + children = real(routes, maintain) + planned.append([child.name for child in children]) # ty: ignore[unresolved-attribute] + return children + + monkeypatch.setattr(_server, "_supervisors", recording) + async with serve(stream): + pass + + assert planned == [[]], "no maintainer for a finished log" + + +class TestReviving: + async def test_restore_refuses_a_retired_stream_without_it(self, tmp_path): + await produce(tmp_path) + streamcast.Stream.retire("t", root=tmp_path) + with pytest.raises(streamcast.StreamRetired, match="revive=True"): + streamcast.Stream.restore("t", root=tmp_path, published=tmp_path.as_uri()) + + async def test_it_continues_on_a_new_log_where_the_old_one_ended( + self, tmp_path, serve + ): + await produce(tmp_path) + streamcast.Stream.retire("t", root=tmp_path) + + revived = streamcast.Stream.restore( + "t", root=tmp_path, published=tmp_path.as_uri(), revive=True + ) + assert revived.retirement is None + assert revived.log is not None + assert revived.log.name == "t-v2" + assert await revived.send_many([{"i": 7}, {"i": 8}]) == [8, 9], "dense" + + metadata = _metadata.load(tmp_path, "t") + assert metadata is not None + assert metadata.retirement is None + assert [(e.name, e.start_offset, e.end_offset) for e in metadata.logs] == [ + ("t", 1, 8), + ("t-v2", 8, None), + ] + # Back to version 2: nothing a build before retirement cannot read. + assert '"streamcast_metadata": 2' in _metadata.path(tmp_path, "t").read_text() + + async with serve(revived, maintain=False) as uri: + async with streamcast.connect(uri, offset=1, catch_up=True) as sub: + got = [(await sub.recv())[0] for _ in range(9)] + + assert got == list(range(1, 10)), "across the seam, from the table" + + async def test_a_revive_that_died_after_creating_its_log_carries_on( + self, tmp_path, monkeypatch + ): + await produce(tmp_path) + streamcast.Stream.retire("t", root=tmp_path) + real = _metadata.save + + def failing(*_: object, **__: object) -> None: + msg = "disk full, mid-revive" + raise OSError(msg) + + monkeypatch.setattr(_metadata, "save", failing) + with pytest.raises(OSError, match="mid-revive"): + streamcast.Stream.restore( + "t", root=tmp_path, published=tmp_path.as_uri(), revive=True + ) + + monkeypatch.setattr(_metadata, "save", real) + revived = streamcast.Stream.restore( + "t", root=tmp_path, published=tmp_path.as_uri(), revive=True + ) + try: + assert revived.log is not None + assert (revived.log.name, revived.end_offset) == ("t-v2", 8) + finally: + await revived.aclose() + + +@pytest.mark.replication +class TestOnAnotherBox: + async def test_retire_here_and_revive_there(self, tmp_path, s3, bucket, serve): + """The planned move: nothing published is lost, and nothing is left.""" + box_a, box_b = tmp_path / "a", tmp_path / "b" + await produce(box_a, published=bucket, s3_options=s3) + streamcast.Stream.retire("t", root=box_a, s3_options=s3) + published_copy = _metadata.fetch(bucket, "t", s3) + assert published_copy is not None + assert published_copy.retirement is not None + shutil.rmtree(box_a) # the old box is gone + + revived = streamcast.Stream.restore( + "t", root=box_b, published=bucket, s3_options=s3, revive=True + ) + assert await revived.send_many([{"i": 7}]) == [8] + async with serve(revived, maintain=False) as uri: + async with streamcast.connect( + uri, offset=1, catch_up=True, s3_options=s3 + ) as sub: + got = [(await sub.recv())[0] for _ in range(8)] + + assert got == list(range(1, 9)) + fetched = _metadata.fetch(bucket, "t", s3) + assert fetched is not None + assert fetched.retirement is None, "flipped in the published copy too" + + async def test_a_stream_without_wal_replication_restores( + self, tmp_path, s3, bucket + ): + """litelink rebuilds a log from its published table when there is no + replica: fenced above anything the old log may have issued.""" + box_a, box_b = tmp_path / "a", tmp_path / "b" + stream = streamcast.Stream.new( + "t", root=box_a, schema=SCHEMA, published=bucket, s3_options=s3 + ) + try: + await stream.send_many([{"i": i} for i in range(5)]) + assert stream.log is not None + while stream.log.seal(flush=True) is not None: + pass + + stream.log.publish(flush=True) + stream.ensure_metadata() + finally: + await stream.aclose() + + shutil.rmtree(box_a) + + restored = streamcast.Stream.restore( + "t", root=box_b, published=bucket, s3_options=s3, published_reserve=1000 + ) + try: + # Above the published end by the reserve the caller chose. + assert restored.end_offset is not None + assert restored.end_offset >= 6 + 1000 + [offset] = await restored.send_many([{"i": 5}]) + assert offset == restored.end_offset - 1 + finally: + await restored.aclose() diff --git a/tests/test_stamp.py b/tests/test_stamp.py index 75fa648..4b38011 100644 --- a/tests/test_stamp.py +++ b/tests/test_stamp.py @@ -173,7 +173,7 @@ async def test_a_row_that_supplies_it_is_refused(self, stamped): async def test_a_remote_publisher_is_told_why(self, serve, stamped): stream = streamcast.Stream("trades", log=stamped) - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri) as producer: with pytest.raises(Exception, match="stamped by the server"): # noqa: B017, PT011 await producer.send({**trade(0), _log.STAMP: 1}) diff --git a/tests/test_types.py b/tests/test_types.py index f843b21..aea166e 100644 --- a/tests/test_types.py +++ b/tests/test_types.py @@ -249,7 +249,7 @@ async def test_binary_arrives_as_text_and_is_stored_as_bytes(self, tmp_path, ser "blob": "AP8A", "nested": {"m": {"b": "0102"}}, } - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri) as producer: assert await producer.send(wire) == 1 @@ -274,7 +274,7 @@ async def test_the_python_publisher_sends_bytes_in_each_columns_encoding( """ stream = stream_at(tmp_path) assert stream.log is not None - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri) as producer: assert await producer.send(row(1)) == 1 assert await producer.send_many([row(2), row(3)]) == [2, 3] @@ -288,7 +288,7 @@ async def test_the_python_publisher_sends_bytes_in_each_columns_encoding( async def test_text_that_is_not_its_encoding_is_rejected(self, tmp_path, serve): stream = stream_at(tmp_path) - async with serve(stream, publish=True, maintain=False) as uri: + async with serve(stream, maintain=False) as uri: async with streamcast.publish(uri) as producer: with pytest.raises(Exception, match="binary in base16"): # noqa: B017, PT011 await producer.send({**row(), "trace_id": "not hex"}) diff --git a/tests/test_writer.py b/tests/test_writer.py index 02895f3..5b756e9 100644 --- a/tests/test_writer.py +++ b/tests/test_writer.py @@ -242,9 +242,7 @@ async def test_a_remote_publisher_is_held_at_the_socket( stream = streamcast.Stream.new("trades", root=tmp_path, schema=SCHEMA) held, _commits = hold(stream, monkeypatch) try: - async with serve( - stream, maintain=False, publish=True, max_inbound=2 - ) as uri: + async with serve(stream, maintain=False, max_inbound=2) as uri: async with streamcast.publish(uri) as producer: futures = [await producer.submit({"i": n}) for n in range(20)] await asyncio.sleep(0.1) diff --git a/uv.lock b/uv.lock index 9c243b3..c17ab97 100644 --- a/uv.lock +++ b/uv.lock @@ -700,7 +700,7 @@ wheels = [ [[package]] name = "litelink" -version = "0.9.0" +version = "0.10.1" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "duckdb" }, @@ -708,12 +708,12 @@ dependencies = [ { name = "pyiceberg", extra = ["pyarrow", "sql-sqlite"] }, { name = "xxhash" }, ] -sdist = { url = "https://files.pythonhosted.org/packages/b7/ed/78869573728d3629c81df24dd3336ab78cf39e5359e2f873a5986e42709c/litelink-0.9.0.tar.gz", hash = "sha256:1cc772ef187126244ab4de12c99dd553b0b398fd686d874e1689e4f05b1fbf2b", size = 334817, upload-time = "2026-10-04T08:26:53.513Z" } +sdist = { url = "https://files.pythonhosted.org/packages/f6/62/93de0c03314752e374dcf2184930b6f740acce00dd284dcbfd2775e1e252/litelink-0.10.1.tar.gz", hash = "sha256:9e22771b3ba7c046fde8234b776768f06e90766e140ee2caa5ff1b900b217c5b", size = 339945, upload-time = "2026-10-04T19:42:47.91Z" } wheels = [ - { url = "https://files.pythonhosted.org/packages/76/d0/ecf290860c88540ef5c83205ac4792a2a4940a7cbf629a623c670c54e7ff/litelink-0.9.0-py3-none-macosx_11_0_arm64.whl", hash = "sha256:907558f6f251da04027b568761eee6e5dd681ab0233bdf0f00edb3b59e65e146", size = 44219326, upload-time = "2026-10-04T08:26:31.605Z" }, - { url = "https://files.pythonhosted.org/packages/fa/f5/e6e38817be74ef44e43c9cd55d553a4aceedbd118e71dc056047d5a01ded/litelink-0.9.0-py3-none-macosx_11_0_x86_64.whl", hash = "sha256:d3a5f5dd26b8ead8d6dd7eea63cdd562dfe77239f1683f4f74c5e43102a44184", size = 46940922, upload-time = "2026-10-04T08:26:37.06Z" }, - { url = "https://files.pythonhosted.org/packages/60/a6/dc7778a650d1c9e4e1509607a5edde406972f25e66764302284860a75ab4/litelink-0.9.0-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:42050220a3aeec55454a76a35a4791e3fbaba5ed00db84b3b17c6f383e3eb8af", size = 56089929, upload-time = "2026-10-04T08:26:43.398Z" }, - { url = "https://files.pythonhosted.org/packages/3e/20/2b813be4090c27816fde13e9eab9aa73bb11ac69129016f47778a8a4ad60/litelink-0.9.0-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:47e6ca3a605280bb9668bc83ab966c0b0c135de6b7da72d8744bdd6102af9ff9", size = 59987930, upload-time = "2026-10-04T08:26:50.854Z" }, + { url = "https://files.pythonhosted.org/packages/14/9d/1a8668e052c70eb7bce1e5de995caa948c6cb892d7bdf302415d2f9a9f70/litelink-0.10.1-py3-none-macosx_11_0_arm64.whl", hash = "sha256:6142daf82e8a65ad823f5e3139af8b4be92b1d007e7fa735607a152d774222a3", size = 44224397, upload-time = "2026-10-04T19:42:34.92Z" }, + { url = "https://files.pythonhosted.org/packages/2e/fe/9d67efa46ffe516c2c22a1b4ea73cf5bc24c2301af1ce25d4813e6bbeaec/litelink-0.10.1-py3-none-macosx_11_0_x86_64.whl", hash = "sha256:e1eb9ab0004b846f45b42232e73719d2bfd0c6c8fcbec13aa99a2aa3df4cea03", size = 46945989, upload-time = "2026-10-04T19:42:38.221Z" }, + { url = "https://files.pythonhosted.org/packages/bf/1b/bafebeae1cb064f73ce04e6cbe46882ec94c51196d2a1a7e8b2a04b2d277/litelink-0.10.1-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:cd637023bece7d3d9d1d3ab84ddf0593ddc7eb2c4da3ee7060e413fc18ad6e51", size = 56094995, upload-time = "2026-10-04T19:42:41.911Z" }, + { url = "https://files.pythonhosted.org/packages/0f/13/d119a3386e5571d6f92289fd9f02fcfd433db7a62052959042c97d403778/litelink-0.10.1-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:801fdc227456a63ace88981e6dfef1e25f860daa532566e37c57cd9d3009fc72", size = 59992999, upload-time = "2026-10-04T19:42:45.292Z" }, ] [[package]] @@ -2003,7 +2003,7 @@ dev = [ [package.metadata] requires-dist = [ - { name = "litelink", specifier = ">=0.9.0,<0.10" }, + { name = "litelink", specifier = ">=0.10.1,<0.11" }, { name = "msgspec", specifier = ">=0.18" }, { name = "starlette", marker = "extra == 'asgi'", specifier = ">=0.40" }, { name = "websockets", specifier = ">=14" },