Streaming#

Streams are the ingestion layer: durable, append-only event logs with HTTP endpoints, exactly-once loading into the warehouse, and automatic wiring to the models that read them.

Declaring a Stream#

from interlace import stream

@stream(
    "orders",
    schema={"order_id": "string", "customer_id": "integer", "total": "double"},
    idempotency_key="order_id",
    retention="7d",
)
def orders(event):
    return event    # placeholder — the function body is never executed

Field types: int/integer/bigint, double/float/decimal, text/string/varchar, bool/boolean, timestamp (ISO strings), json.

Publishing Events#

With the daemon running, POST /streams/{name} accepts one event or a batch:

curl -X POST localhost:8000/streams/orders \
  -H 'content-type: application/json' \
  -d '{"order_id": "o-1001", "customer_id": 42, "total": 99.5}'

{ "accepted": 1, "deduplicated": 0, "last_offset": 1, "quarantined": 0 }

The event is durable before the response returns — appended to a write-ahead-logged SQLite log at .interlace/streams.db. Loading into the warehouse is asynchronous.

Idempotency#

If the stream declares an idempotency_key, re-sending an event with a seen key is acknowledged but not re-appended (deduplicated counts it). Retrying webhooks is therefore always safe. Deduplication is per stream and lasts as long as the event lives in the log (i.e. until retention trims it).

Backpressure#

If the warehouse falls far behind (100,000 unmaterialised events on one stream), publishing returns 429 with a retry hint rather than accepting unbounded lag.

Consuming events#

GET /streams/{name}/events is a Server-Sent Events tail of the same durable log, for a consumer that is not an Interlace model. Warehouse materialisation does not use this cursor.

curl -N 'localhost:8000/streams/orders/events?after=0'

Each data frame is {offset, ts, payload, idempotency_key, headers} and its SSE id is the offset, so a reconnect sends Last-Event-ID and resumes. With no cursor the tail starts at the current head (live only). The server blocks until an event is appended; it does not poll the log. A comment frame opens the stream, and another follows 15 seconds of quiet so proxies keep the connection.

Sending a frame does not acknowledge it. Pass group to take that consumer group’s lease (a second subscriber gets 409 until the first disconnects) and, unless you also pass a cursor, resume from the group’s committed offset. The first frame is event: lease with {group, token, committed_offset}. Ack while the tail is still open:

curl -X POST localhost:8000/streams/orders/commit \
  -H 'content-type: application/json' \
  -d '{"group": "billing", "offset": 42, "token": "<from the lease frame>"}'

A stale token is 400. Disconnect releases the lease and leaves the committed offset where the last successful commit put it. <name>__quarantine is the same tail for rows the drift policy diverted.

From Log to Warehouse#

The daemon micro-batches: a flusher coalesces bursts (50 ms), then loads events in 5,000-row batches into streams.<name>, which carries the declared columns plus:

Data and the stream’s watermark move in one transaction — the watermark lives in the warehouse (streams._watermarks), so it commits atomically with the data. A crash leaves either the old watermark (events re-read, stage overwritten — no duplicates) or the new one; it never double-loads or drops events (exactly-once into the warehouse). Poll GET /streams/{name} and compare head (accepted) with watermark (materialised) to see events land.

Consuming Streams in Models#

Reference the stream table like any other:

-- models/order_facts.sql
SELECT order_id, customer_id, total
FROM streams.orders

The daemon enqueues models that read a stream — plus their descendants — after each flush, with watermark-keyed idempotent runs so repeated flushes at the same position debounce into one run.

Schema Drift#

on_schema_drift controls what happens when an event doesn’t match the declared schema. Validation happens before durability, per request:

• reject (default) — any undeclared field or type mismatch fails the whole request with 400. Missing declared fields are fine (they load as NULL).
• evolve — unknown fields are accepted and become new columns on the warehouse table (types inferred; existing rows read as NULL). Incompatible changes to declared fields still fail — evolution never hides breakage.
• quarantine — conforming rows are accepted; violating rows are diverted to a shadow table streams.<name>__quarantine with error and payload columns, and counted in the response’s quarantined. The request succeeds.

Retention#

retention bounds the durable log, not the warehouse table: an event is trimmed from the log only once it is both materialised and older than the window. The sweep runs on the daemon’s scheduler loop. Without retention, the log keeps everything. interlace reset --yes (or POST /reset) clears the log and the streams landing tables entirely — a fresh start, not a retention sweep.

Postgres CDC#

cdc: in interlace.yaml names a postgres connection, a replication slot, a publication, and the tables. interlace serve reads pgoutput and appends each change to the declared @stream. _change is insert, update, or delete. Every value arrives as text, so the stream schema uses string fields and a downstream model casts them. The stored LSN advances only after those offsets have been flushed to the warehouse.

interlace init --template cdc scaffolds a Docker Postgres with the slot and publication, the stream, and a full_merge replica that keeps the latest change per key and drops deletes. Reading the slot needs Postgres 14 or newer.

Observability#

interlace streams        # per stream: head, watermark, pending, drift mode, retention

GET /streams, GET /streams/{name} (includes the last 20 payloads), and stream.flushed events on the live event feed cover the same from the API and web UI.

Reverse-ETL Terminals#

Streaming is the inbound edge; terminal materialisations are the outbound one. A model with materialise: table or materialise: file produces no owned snapshot and no environment view — instead its resolved query result is delivered to an external destination each time it builds. Terminals are the other half of the platform’s I/O boundary, so they’re worth knowing about here even though they’re declared on ordinary SQL models.

/* interlace:
  materialise: table
  target: crm.main.customer_scores
  strategy: merge
  key: customer_id
*/
SELECT customer_id, score FROM customer_value

• Files — materialise: file with format: parquet | csv | json and path, written via a DuckDB COPY. ${date}, ${datetime}, and ${workspace} expand in path when the file is written.
• External tables (reverse ETL) — materialise: table with target: <alias>.<schema>.<table>, where alias is a database wired in through the project’s attach: config (Postgres, SQLite, another DuckDB). strategy picks delivery: full (DELETE all + INSERT — the live table is never dropped, so grants and readers survive), append, or the keyed merge / full_merge. Column drift follows schema.columns (additive by default, or reject / ignore). incremental works here too.
• Environment-gated — terminals fire in prod only by default, so a dev apply builds and fingerprints the model but skips delivery. Widen with environments: [dev, prod].

The full field reference is in the SQL models guide.

Next Steps#

• REST API — the daemon, auth, and events
• Dependencies — how stream consumers are wired