Live streams

Record a wss venue, then build a table from the batches captured during that window.

A wss venue emits messages only while a socket is open, so a recorder captures the bytes first. mr-data stream is that recording, and it runs four steps in front of the ordinary build loop. The build that follows reads recorded batches and takes no duration of its own.

The four steps

mr-data stream document register --file connector.json --name coinbase --json
mr-data stream registry create --digest DOCUMENT_DIGEST --json
mr-data stream probe --document-digest DOCUMENT_DIGEST --registry REGISTRY_ID \
  --member BTC-USD --json
mr-data stream capture --document-digest DOCUMENT_DIGEST --registry REGISTRY_ID \
  --member BTC-USD --seconds 300 --watch --json

1. Register the connector document. It states where the venue is, how a message names its own type and event time, and what a subscription frame looks like.
2. Add it to a registry, the reviewed set a recording may name. Pass every digest the recording will use, up to 32.
3. Probe first. A probe connects briefly, reports which message types the document classified and which it did not, and records nothing.
4. Capture or subscribe. capture --seconds N records one bounded window and keeps it. subscribe keeps recording until somebody stops it, checkpointing as it goes.

Registration prints the digest the recipe will name, and mr-data stream document show ID reads one back. mr-data stream status ID --kind probe reads a probe. A document that classifies nothing needs fixing before a capture window goes on it.

Every verb and flag

mr-data stream takes one of nine verbs. A typo refuses naming all nine.

Verb	What it does
document register | document show	Register a connector document, or read one back.
series register	Register the rotating roster a subscription follows.
registry create	Accept a set of documents.
probe	Listen briefly and record nothing.
capture	Record a bounded window and keep it.
subscribe	Keep listening, checkpointing batches a later build continues from.
subscription show | pause | resume | stop	Read or move one subscription.
status	Read a capture or a probe.
recordings	Name what a build would bind, and until when.

Flag	What it does
--file FILE	With document register or series register, the document to register, as a path to a JSON file.
--name NAME	With document register, the name to hold the registered document under.
--digest DIGEST	One registered document digest. With registry create, one of the documents the registry holds, repeated up to 32 times. With series register, the connector document the series rotates members of, which creates one registry over that document and the series together.
--document-digest DIGEST	The connector document to act against, by the digest its registration printed.
--registry REGISTRY	The registry that holds the document, by the identifier its creation printed.
--series SERIES	With subscribe, the series whose roster this subscription follows, by the id the series document declares.
--member MEMBER	One member of the venue to listen to: a channel, a ticker, a market. Repeated once per member, up to 256.
--seconds SECONDS	With capture, how long to record for, at most 600.
--max-events MAX_EVENTS	With capture, stop after this many messages, at most 10,000.
--max-bytes MAX_BYTES	With capture, stop after this many recorded bytes, at most 8,388,608.
--secret NAME	The stored credential the venue authenticates with, by the name mr-data keys set printed. No command names the value, and it never leaves the vault.
--kind capture | probe	With status, which resource the identifier names. A probe and a capture live on separate routes and an identifier does not say which. capture by default.
--watch	With capture, keep reading the record back until it stops, and print what it stopped on. The capture runs under the bound it was given either way.
--json	One machine-readable JSON object instead of human lines.

The 600-second ceiling is the recorder’s own hard bound, and it refuses a request above it rather than lowering it.

Public and keyed venues

An open venue uses stream.wss@1.0.0. A keyed one uses stream.wss@2.0.0, and the recipe still states credential_mode: "none", because the recording authenticates and the build does not.

mr-data keys set venue_key --part key_id=ABC123 --part private_key_pem=@signing.pem --json
mr-data stream capture --document-digest D --registry R --member M --seconds 300 \
  --secret venue_key --json

Store the shaped credential once, part by part, and name it on the recording. The client combines the parts into one envelope and sends the value only in the request body. The venue’s document declares the parts its credential shape requires. If the set does not match, the recorder names the invalid parts. No command flag takes a secret as an argument.

A probe reports an auth_outcome: authenticated, auth_rejected, auth_unconfirmed, clock_skew_suspected and six more.

The connector grammar accepts wss on port 443 and nothing else. It resolves the endpoint’s name and requires every answer to be publicly routable before it opens a socket. It refuses a venue whose terms bar reading it, with the recorded reason, before a connection could exist.

Subscription lifecycle

subscribe starts a subscription that keeps recording until somebody stops it, checkpointing what a build continues from. mr-data stream subscription show ID reads its state:

{
  "status": "…",
  "desired_state": "…",
  "series_id": "…",
  "members_now": 173,
  "members": ["…"],
  "members_not_shown": 0,
  "last_data_at": "…",
  "last_discovery_at": "…",
  "next_discovery_at": "…",
  "next_boundary_at": "…",
  "gap_count": 0,
  "last_gap": null,
  "bound_table_ids": ["…"],
  "checkpoint_sequence": 412,
  "current_cursor": "…"
}

pause, resume and stop move it. status and desired_state are two facts, not one: the second is what somebody asked for, the first is where the subscription actually is.

Each stream advances one durable record, scoped and windowed, and a rotating series keeps one per member. The record holds that stream’s bookmark and a bounded window of the event IDs it most recently accepted. It also counts the IDs the window has evicted. The bookmark carries the anti-replay guarantee: it refuses a batch that does not follow the position the scope is at. The window catches a reconnecting venue re-sending its recent tail. It is not an all-time ledger.

Series-bound members

Where the roster rotates, register a series rather than naming members by hand. Hourly and daily markets rotate like this: contracts expire and the venue lists new ones.

mr-data stream series register --file series.json --digest DOCUMENT_DIGEST --json
mr-data stream subscribe --series SERIES_ID --registry REGISTRY_ID --json

Registering a series creates one registry over that document and the series together. A subscription then names its members outright or names a series, never both. Its members are whatever the series says they are at each moment. next_discovery_at and next_boundary_at on the subscription record say when it next re-reads the roster and when the current window ends.

A capture of one contract of a rotating series pins a digest no stored document has. That digest is the declared document with this contract’s own stream ID in it. The registry accepts that digest only when it is what one of the registry’s own documents becomes with this stream ID substituted.

Bind the recording from a recipe

A stream source states three things: a stream locator whose display_locator is the wss:// endpoint, credential_mode: "none", and connector.stream_connector_document_digest. That digest names the connector document the capture ran under.

{
  "name": "ticker_frames",
  "source_class": "stream",
  "data_classification": "public",
  "locator": {"kind": "stream", "display_locator": "wss://ws.example.com/feed"},
  "rights_claim": {
    "claimed_basis": "terms_of_service_asserted",
    "claim_evidence_digest": "sha256:…"
  },
  "connector": {
    "adapter_id": "stream.wss@1.0.0",
    "credential_mode": "none",
    "stream_connector_document_digest": "sha256:…"
  },
  "limits": {"max_source_bytes": 3145728}
}

mr-data run has no duration flag and will not grow one. A build reads recorded batches and those batches bound it. The recorder fixed the bytes when it recorded them, not when the build started, so the build reproduces.

When no recording exists, the build fails even if the recipe is valid:

mr-data stream recordings --document-digest D --json

It reports every batch a build could read for that venue, the bookmark each one carries, and when each expires. A recording whose expires_at has passed wants replacing rather than naming. A run refuses an aged-out batch before it claims a worker, not midway through a build.

What the transform is handed

The relation is fixed and the recipe does not declare it. A stream source presents to the SQL as the recorder’s own line, every column text, in this order. Two shapes exist and the connector picks one. It writes eight columns, or nine when it declares a member field. member is then fourth, directly after event_time. A connector that says nothing about attribution writes the eight columns it has always written, byte for byte.

Column	What it holds
epoch	The connection this frame arrived on, as text.
event_id	The identity the recorder assigned the frame.
event_time	The declared event time in the kind the connector declares, or the RFC 3339 arrival time when that frame was timed by receipt.
member	Which member of the venue the frame is about, only when the connector declares a member field. A table partitions by it without parsing the message.
message	The provider frame, canonical JSON text.
message_class	snapshot, delta, or whatever else the connector classifies.
message_type	The venue’s own message type.
received_at	RFC 3339 UTC receipt time.
sequence	The frame’s position within its connection, as text.

Read the frame with the engine’s JSON functions:

select received_at,
       message_type,
       cast(json_extract(message, '$.price') as varchar) as price
from ticker_frames

event_time can hold two kinds of value, row by row. A connector whose event_time_field is null times every frame by receipt. One whose event_time_fallback is received_at times only the frames the venue did not stamp. In one relation, a row the venue stamped carries the declared kind, say epoch milliseconds. A row beside it carries an RFC 3339 arrival time. A transform that casts the column blind gets nulls for half of it and no error.

Read received_at when you need the arrival time. When you need the venue’s own timestamp, guard the cast against its possible shapes:

select case
         when event_time like '%-%' then cast(event_time as timestamp)
         else to_timestamp(cast(event_time as bigint) / 1000)
       end as stamped_at
from ticker_frames

A connector that declares an event_time_field and leaves event_time_fallback at refuse, the default, never mixes. Every frame includes the declared kind, or the capture rejected it.

How a stream table grows

A refresh of a recorded stream re-reads nothing. Each run records the bookmark it read up to, and the next run continues from there, an increment. It rewrites no part: it keeps every part of the previous version and adds its own. The table declares no materialization for this, and the worker reads none.

A missed window is a permanent hole

Recordings live for thirty days. A refresh continues from the bookmark the last build reached. Neither can recover messages recorded before that bookmark or messages that were never recorded.

If a venue went unrecorded over a period somebody cares about, say so as a gap in the series. Do not promise a backfill. There is nothing to fetch, and no refresh, backfill or re-run repairs it. Keep the subscription running where continuity matters.

What is not covered

• Interactive-login venues. The grammar accepts a stored credential that the recorder can redeem, not a browser session, form submission or interactive challenge.
• Anything that is not a WebSocket. Server-sent events, long polling, message queues and vendor SDK callbacks have no recorder. Use an ordinary HTTPS source where the venue also publishes a fetchable address.
• A build reading a socket. Nobody can replay a subscription an hour later and receive the same messages, so a connection is a capture role, never an acquisition transport.
• Backfilling before the first capture. History begins when the recording did.

Table-driven rosters

When another live table should choose the members, use a table_members series. The table supplies actual venue member IDs and typed context; discovery seals that projection with each recording segment. It does not match entities or build member URLs. See the complete table-driven stream discovery recipe.

Next

Write a recipe covers the rest of the document. Publish and keep it fresh covers how a stream table refreshes.