Collection Streams

An observation over a set returns a page: a bounded number of entities, selected and sorted by the query. Pages suit a screen. They do not suit a reader that needs the whole collection — an export, a search index, a client that keeps a local copy — and they say nothing about what changed after the page was read.

A stream is the third scope of state, beside one entity and a set. It gives an operation the selected entities one at a time, as the storage finds them, and it lets a reader come back later for only what changed.

§Reading without holding

An operation with the stream scope receives a stream of entities instead of an array:

export async function* observation (input: Input, stream: AsyncIterable<{ entry?: Entity }>) {
  for await (const part of stream)
    if (part.entry !== undefined)
      yield toRow(part.entry)
}

Nothing holds the collection whole: not the storage, not the operation, not the reply. A collection of any size is read at the cost of the entity currently in hand. The generic prototype supplies a stream operation that returns the entities as they are.

§Reading again

A complete stream ends with a token. The token stands for the point the reader has reached. A later read that presents it receives what happened in the collection since, and a new token.

A stream therefore yields three kinds of parts:

“The collection” is whatever the query selects. An order that stops matching status==pending has left the collection of pending orders, and a reader of that collection is told it was removed, though the order still exists.

This gives a reader one protocol for two jobs. The first read, without a token, transfers the collection. Every later read, with a token, transfers the difference. A reader that keeps a copy applies both the same way.

§Windows

A reader may ask for a bounded number of entities at a time. Each window ends with a token that continues the collection, and a window that comes back short has reached the end. From there the same token continues with changes.

Windows are ordered by identity or by creation time. These never change for an entity, so an entity stays on one side of a window boundary for the whole read. Sorting by a property that can change would let an entity move across a boundary and be read twice or not at all.

§What a reader can rely on

A token is the mark of a complete stream. A stream that ends without one was cut. The reader asks again with the token it already holds, and loses nothing.

An entity may arrive twice. Once in a window and once as a change, for instance. Every entity carries its VERSION, and a reader keeps the higher one.

A token belongs to its query. It continues the collection it was issued for, and is refused for another selection or order.

A token expires. How long changes can be read back is how long the storage remembers them. A reader whose token has passed that horizon is told so, discards its copy, and reads the collection again.

§What it requires

Reading changes from a token depends on the storage remembering the order in which its writes were committed, and what an entity was before each change. A storage without that history still streams a collection and continues it window by window. It ends a complete stream by saying that there is nothing to continue from, and a reader learns what changed by reading again.

§Streams and events

Events and collection streams both tell an interested party that state changed. They serve different readers.

An event is pushed to components that subscribed in advance, and its payload is a contract. A stream is pulled by a reader, starts from the current state, and describes the collection itself. A component reacts to events. A copy of a collection is kept current with a stream.