feat(stream): retire a stream, serve it read-only, revive it on restore

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,

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=<unset>, cursor=None, cursor_uri=None,
@@ -128,7 +128,7 @@ streamcast.connect(uri, *, offset=<unset>, 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,17 +412,20 @@ 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(
     "trades", root="data", published="s3://market-data/prod", replay_published=True,
 )
 ```
 
-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.

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

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

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.

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)

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

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]] = []
 

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:

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:

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]}"

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

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",