Streams

A streaming query reads a queue, decides something per message or per time bucket, and pushes the result to another queue. You write it as an operator chain, the chain runs in your process, and its state lives in the same PostgreSQL that holds the messages. Two shapes exist: a window plus a reducer, which folds buckets and emits one message per closed window, and a gate, which makes an allow or deny decision on every message.

There is no changelog topic, no state store and no job manager to deploy. A stream is a worker holding a partition lease.

// Illustrative, not extracted from a test.
import { Queen, Stream } from 'queen-mq'

const url = 'http://localhost:6632'
const queen = new Queen({ url })

const handle = await Stream
  .from(queen.queue('orders'))
  .filter(msg => msg.data.amount > 0)
  .windowTumbling({ seconds: 60, idleFlushMs: 5000 })
  .aggregate({ count: () => 1, sum: m => m.amount })
  .to(queen.queue('orders.per_minute'))
  .run({ queryId: 'orders.per_minute', url, batchSize: 50 })

let handle = Stream::from(q.queue(&src))
    .window_tumbling(2)
    .idle_flush_ms(500)
    .aggregate_count("count")
    .aggregate_sum("sum", |r| r.number("amount"))
    .to(q.queue(&sink))
    .run(
        &q,
        RunOptions::new(&query)
            .reset(true)
            .batch_size(50)
            .max_wait(Duration::from_millis(300)),
    )
    .await
    .unwrap();

Each closed window is one ordinary message on the sink queue, one per window and per key, whose payload is the value that left the chain. Closed windows need no results API: anything that reads a queue reads them.

Open windows have one. POST /streams/v1/state/get returns the live accumulators for a query and partition, by exact key, by key prefix, or by ripeness, at read-only access level. Because the state is rows in the same PostgreSQL, there is no instance to locate first and no metadata service to run.

The four windows

• Tumbling: fixed, non-overlapping buckets, windowTumbling({ seconds: 60 }).
• Sliding: overlapping windows of size hopping every slide, windowSliding({ size: 3600, slide: 60 }).
• Session: one window per key, extending while events for that key keep arriving within gap seconds, windowSession({ gap: 1800 }).
• Cron: wall-clock aligned buckets, windowCron({ every: 'day' }), where every is second, minute, hour, day or week.

Every window carries the same four event-time options and the same idle-flush timer, described next.

Event time

Without an eventTime extractor a window buckets on the broker’s createdAt, and the close trigger is the newest createdAt in the batch just popped. Pass eventTime, a function returning a Date, an epoch-millisecond number or an ISO string, and four knobs come with it. All four are per window operator, all four exist in the JavaScript, Python, Go and Rust SDKs.

A window whose partition goes quiet would otherwise never close, because the clock only advances when a message arrives. idleFlushMs is the timer that closes it: the runner wakes on that interval, asks POST /streams/v1/state/get for the partitions it has touched in the last five minutes and the rows whose windowEnd is already ripe, and commits a cycle carrying the state deletes and the sink pushes with no source ack. It defaults to 5000 ms for tumbling and sliding, 1000 ms for session, and 30000 ms for cron. Pass 0 to disable it, and a window then closes only when the next message for that partition arrives.

Gating

.gate(fn) is the second shape. Instead of folding a batch, it runs fn(value, ctx) on each message in source order, where ctx.state is that key’s state row, mutable. Return true and the message flows downstream, the mutated state is committed, and the message counts toward the ack. Return false and the mutation is discarded, because it did not happen.

// Illustrative, not extracted from a test. A fixed budget per key, no refill.
await Stream
  .from(queen.queue('api-requests'))
  .gate((req, ctx) => {
    ctx.state.remaining ??= 10
    if (ctx.state.remaining < 1) return false
    ctx.state.remaining -= 1
    return true
  })
  .to(queen.queue('api-allowed'))
  .run({ queryId: 'api-budget', url })

A deny stops the batch and keeps the lease. The runner commits the cycle with release_lease: false and ack.count set to the number of messages allowed before the deny, so the denied message and everything after it stay leased. When the lease expires they are redelivered in their original partition order. If the very first message is denied there is no cycle at all: no state write, no push, no ack, and the lease simply times out.

That is what makes a rate limiter or a circuit breaker expressible without a deferred queue, and the ordering is preserved by construction rather than by a retry policy. The broker side is exact about it: a partial ack of K walks K frames forward over the real queen.log_segments ranges instead of adding K to the cursor, so offsets already removed by retention are skipped rather than counted.

Three constraints, all enforced when the chain is built:

• At most one .gate() per stream. Chain two streams if you need two decisions.
• .gate() and a window plus reducer are mutually exclusive in one stream. The window model assumes the whole batch is consumed atomically, which is exactly what gating breaks.
• .keyBy() must come before the gate. The state key defaults to the source partition id, and for a rate limiter that is the setup you want: partition the source queue by the limit key and add no keyBy at all.

Why it is exactly-once

A cycle is four steps: pop a batch from the source under a lease, read the state rows for that partition, run your operators, commit. The commit is one request and one PostgreSQL transaction carrying three things: the state upserts and deletes, the sink pushes, the ack of the source batch.

All three land or none does. A crash mid-cycle leaves the source cursor where it was, the lease expires, and the batch comes back to rebuild the same window from the same input. A response lost on the wire cannot double-emit either: the sink push happens before the ack inside the transaction, and the ack aborts on an expired lease, so a retried cycle finds the lease gone and its sink write rolls back with it.

Sink pushes are not a second write path. queen.log_streams_cycle_v1 calls queen.log_push_one_v1 inline, in the same transaction, and that is the same allocator POST /api/v1/push and POST /api/v1/transaction go through. One code path is why a window result gets a real offset, a real deduplication probe against the sink queue’s window, and its partition’s ordering, rather than a lookalike write that a consumer can tell apart later.

Two details make the cycle cheap rather than merely correct. The procedure takes an array of cycle elements and wraps each one in its own savepoint, so a failing element rolls back alone while the others commit in the same transaction; the broker route submits one element per request today. And concurrent cycles and idle flushes on the same (query, partition) state shard serialise on a PostgreSQL advisory lock taken before either writes, so a flush timer firing mid-cycle waits rather than interleaving.

foreach is the exception. It calls your function and acks after it returns, so an effect outside PostgreSQL is at-least-once. An effect that must not repeat belongs on a sink queue with a consumer of its own.

Keys, workers and state

State rows are keyed by query, partition and key, and partition leases are exclusive: one worker at a time writes a partition’s state, so no worker locks against another. Parallelism is the partition count, exactly as it is for consumers in one group. keyBy changes the state key. Keep it aligned with the partition, or repartition through an intermediate queue first.

A partition carrying open stream state is never reclaimed underneath you. Maintenance deletes an empty, long-inactive partition only when nothing references it, and a surviving row in queen_streams.state vetoes that delete, so a keyed window on a queue that has gone quiet for a month still has its partition when the next message arrives. See retention.

Closed windows delete their own rows, and nothing else does: retention never touches stream state. It is reclaimed by dropping the query, or by rerunning the chain with reset on. reset is also the answer to the HTTP 409 you get when you redeploy a changed chain under a name already registered, because old accumulators and a new fold shape do not mix.

Two things are called state, and the order between them is not a preference

ctx.state in an operator and KV state are both state in the same PostgreSQL, and only one of them is atomic with the cycle.

Inside a stream, the state primitive is state_ops.

The atomicity you get from state_ops is free: queen.log_streams_cycle_v1 commits the state writes, the sink pushes and the ack together, and there is nothing to configure and nothing to get wrong. The KV is newer and more visible, so it is the one that gets reached for by default, and reaching for it silently gives up the only guarantee streaming already handed you.

The KV earns a place inside a stream for one job, and it is the job state_ops cannot do: state that crosses partitions or queries. queen_streams.state is keyed by query and partition, so a marker on a business id whose events fall on different partitions, or a budget shared across every partition of a query, is not expressible in it. Those are the cases, and there are no others.

The naming is part of the defence. The handle is queen.kv and tx.kv, deliberately never state, so that the two things cannot be confused at the call site: ctx.state is the row the cycle owns, and anything spelled kv is not.

A KV call made from inside an operator is an ordinary call. It has the guarantees of the standalone KV and none in addition, so if the cycle rolls back and replays, that write has already happened. Anything that must be atomic with the ack belongs in state_ops, which is why the rule above comes first rather than last.

Pacing is not expressible inside an operator

This is the thing somebody will try, so it is written here rather than discovered.

A cycle carries no kv and no timers array. A stream cycle is the one actor in the product that takes a blocking advisory lock, and the total lock order puts queen.kv and queen.log_timers outside it, so a cycle that touched them would hold the two outermost spaces while blocking on an advisory lock. The rule that forbids it is the same one that makes the whole order acyclic, and it is not an omission that a later array would fix.

The consequence is concrete. The inverse rate limiter, an incr with a max deciding admission and a timer moving the rejected work later instead of dropping it, is written in a flat consumer, not in an operator. From a stream you emit onto a queue and the timer goes in the consumer of that queue. That is one extra line of architecture and no loss of any guarantee, but found on your own it looks like an arbitrary limit, so here it is with its reason.

Gating remains the shape a stream does have for pacing: .gate() denies in place, keeps the denied tail leased and in order, and needs no second queue. It discards or defers by lease. It does not reschedule.

One operator chain, four kinds of window or one gate, one transaction per cycle, and results that arrive on a queue like any other message.