A mutable, append-optimized event buffer for live ingestion — the entry
point to the live layer. push validates rows against the same schema
as TimeSeries, retention (maxEvents / maxAge) bounds the buffer,
and subscribers react on the 'event', 'batch', 'columns' and
'evict' channels; snapshot to the batch layer at any time with
toTimeSeries().
new LiveSeries({ name, schema, retention: { maxEvents: 10_000 } })Constructor
new LiveSeries(options: LiveSeriesOptions<S>)
Properties
Static methods
fromJSONstatic
fromJSON(input: TimeSeriesJsonInput<S> & { parse?: TimeZoneOptions }, options?: Omit<LiveSeriesOptions<S>, 'name' | 'schema'>): LiveSeries<S>
LiveSeries.fromJSON({ name, schema, rows })Static
factory: builds a fresh LiveSeries from a JSON snapshot
envelope, parsing each row through parseJsonRow.
The retention/grace/ordering options on the second argument are passed through to the constructor; pass them when you want the reconstructed series to behave like its original (e.g. on a client reconnecting and rehydrating from a server snapshot).
Use parse: { timeZone } when JSON timestamps are local-
calendar strings — same semantics as TimeSeries.fromJSON.
Methods
aggregate
aggregate(sequence: Sequence, mapping: M): LiveAggregation<S, readonly [ColumnDef<'interval', 'interval'>, AggregateColumns<ValueColumnsForSchema<S>, M>]>
at
at(index: number): EventForSchema<S> | undefined
atOrAfter
atOrAfter(key: KeyLike): EventForSchema<S> | undefined
live.atOrAfter(new Time(t))Event with the exact key, or the nearest later one.
atOrBefore
atOrBefore(key: KeyLike): EventForSchema<S> | undefined
live.atOrBefore(new Time(t))Event with the exact key, or the nearest earlier one.
bisect
bisect(key: KeyLike): number
live.bisect(new Time(t))Insertion index for key
in the sorted live buffer (binary search; O(log N)). Same shape
as Array.prototype semantics: returns the lowest index where
an event with key could be inserted while preserving order.
clear
clear(): void
cumulative
cumulative(spec: { [K in string]: CumulativeAccumulator }): LiveView<readonly [S[0], ReplaceSmoothedColumn<ValueColumnsForSchema<S>, Targets>]>
diff
diff(columns: Target | readonly Target[], options?: { drop?: boolean }): LiveView<readonly [S[0], ReplaceSmoothedColumn<ValueColumnsForSchema<S>, Target>]>
eventRate
eventRate(): number
Events per second over the current buffer. Computed as
length / (timeRange / 1000). Returns 0 if the buffer is
empty or holds a single event (no time span to divide by).
Mirrors LiveView.eventRate; available directly on
LiveSeries for the buffer-as-window pattern where the user
doesn't want a separate windowed view.
every
every(predicate: (event: EventForSchema<S>, index: number) => boolean): boolean
live.every(e => e.get('healthy'))True when every event matches.
fill
fill(strategy: LiveFillStrategy | LiveFillMapping<S>, options?: { limit?: number }): LiveView<S>
filter
filter(test: Expr<'boolean' | 'any'> | (event: EventForSchema<S>) => boolean): LiveView<S>
Keeps events that pass test: a function, or an expression such as col('value').gt(0).
find
find(predicate: (event: EventForSchema<S>, index: number) => boolean): EventForSchema<S> | undefined
live.find(e => e.get('value') > 0)First event matching the predicate, or undefined.
first
first(): EventForSchema<S> | undefined
includesKey
includesKey(key: KeyLike): boolean
live.includesKey(new Time(t))True when an event with an exactly matching key exists.
last
last(): EventForSchema<S> | undefined
map
map(fn: (event: EventForSchema<S>) => EventForSchema<S>): LiveView<S>
on
on(type: 'event', fn: EventListener<S>): () => void
on(type: 'batch', fn: BatchListener<S>): () => void
on(type: 'evict', fn: EvictListener<S>): () => void
on(type: 'columns', fn: ColumnsListener<S>): () => void
live.on('columns', (batch) => draw(batch.column('cpu')))Fires once per push that adds rows, with just those rows as a
TimeSeries, so a consumer can read whole columns (column(name),
keyColumn(), toColumns(), …) instead of one Event per row.
On the default (strict, time-keyed) series no Event is built and
no row is copied: the batch shares the arrays the series stores, and
wrapping them costs well under a microsecond a push. (A partition
sub-series gathers its rows into fresh arrays; a pushSeries batch of
8 rows or fewer is copied on the way in.) Holding on to a batch keeps
those arrays in memory after retention drops the rows from the series.
After pushSeries, the batch shares the pushed series' arrays, so it
changes only if the caller writes into them (see pushSeries). Other
series (reorder / drop, non-time keys) build
the batch from the added events, sorted by key.
Fires right after 'batch' and before 'evict'. A batch is
immutable: later pushes and retention never change it. Partition
sub-series fire it too, once per routed batch.
partitionBy
partitionBy(by: ByCol, options?: LivePartitionedOptions<K>): LivePartitionedSeries<S, K, ByCol>
Type-parameter order is <ByCol, K> (column name first, then
partition value type) so the explicit-arg form
partitionBy<'host'>('host') binds the literal to ByCol.
That preserves backwards compatibility with the v0.15.0 V8
workaround pattern (which used the explicit-arg form to force
column-literal narrowing through the fused-rolling typing
chain) — and matches what callers usually want when they reach
for the explicit form: declare the partition column.
K (the partition VALUE type) typically narrows from
groups; an explicit <ByCol, K> second arg is rare. If a
caller wants an explicit value union, the natural form is
partitionBy('host', { groups: [...] as const }).
pctChange
pctChange(columns: Target | readonly Target[], options?: { drop?: boolean }): LiveView<readonly [S[0], ReplaceSmoothedColumn<ValueColumnsForSchema<S>, Target>]>
push
push(rows: RowForSchema<S>[]): void
pushJson
pushJson(rows: readonly (JsonRowForSchema<S> | JsonObjectRowForSchema<S>)[], parse?: TimeZoneOptions): void
live.pushJson(rows)Bulk JSON-shape ingest: takes
an array of JsonRowForSchema<S> (or the object-form variant),
parses each row through parseJsonRow (translates null
cells to undefined, parses the key into the right
Time/TimeRange/Interval instance), then dispatches to
LiveSeries.pushMany.
Closes the wire→push safety hole: a JsonRowForSchema<S> is
structurally typed against the schema (column count, value
shapes, null permissibility), so a column added or renamed in
the schema breaks the call site at compile time. The previous
live.push(row as never) workaround swallowed mismatches.
Pass a TimeZoneOptions second argument to disambiguate
local-calendar timestamp strings — same semantics as
TimeSeries.fromJSON's parse option, just inlined as
a sibling argument because pushJson has no input envelope
to attach a parse: key to.
```ts
live.pushJson(rows);
live.pushJson(rows, { timeZone: 'Europe/Madrid' });
```
pushMany
pushMany(rows: readonly RowForSchema<S>[]): void
live.pushMany(rows)Array-form counterpart to
LiveSeries.push: takes a single ReadonlyArray<RowForSchema<S>>
instead of variadic args. Behavior is identical — same per-row
validation, same 'event' / 'batch' / 'evict' listener
semantics, same retention pass at the end.
Reach for pushMany over push(...rows) when ingesting a
snapshot or any large rows array — variadic spread allocates a
stack frame per element and can blow on multi-thousand-row
snapshots. push(...rows) itself is now a thin wrapper around
this method, so behavior between the two is intentionally
identical.
For JSON-shape rows arriving over the wire, prefer
LiveSeries.pushJson — it accepts the JSON envelope
(nulls, raw timestamps) and parses through parseJsonRow. For data
that already comes in columns (typed arrays off a socket or a
worker), build a batch with TimeSeries.fromColumns and push it with
LiveSeries.pushSeries: no row tuples, and on a default series
no per-row check.
Commit granularity differs by backing. On the Event[]
backing each row is appended then its 'event' fires, so a
handler observes length grow row-by-row (1, 2, …) within one
pushMany. On the chunked columnar backing (top-level strict
time-keyed series) the whole batch is appended as one chunk
before any 'event' fires, so a handler sees the full post-batch
length for every event of the batch. This is intrinsic to
all-or-nothing columnar append — per-row commit would reintroduce
the per-row Event cost the chunked backing exists to avoid. The
cross-backing contract callers can rely on: every
successfully-ingested row fires exactly one 'event', in order,
before 'batch'/'evict'.
Listener errors are isolated ([PND-LIVFIX]). A listener that
throws does not stop the push: every row is still committed, the
remaining listeners still run, retention still runs, and 'batch' /
'evict' still fire. The first error is rethrown to the caller once
the push has completed, so length, ingested and every subscriber
(a derived filter() view included) agree afterwards. A push made
from inside a listener is queued and runs after the current push,
in arrival order, so emission stays monotonic.
pushSeries
pushSeries(series: TimeSeries<S>): void
live.pushSeries(TimeSeries.fromColumns({ name: 'ticks', schema: live.schema, columns: { time, price, size } }))Pushes a batch TimeSeries as one batch: the same checks, ordering,
retention and 'event' / 'batch' / 'evict' listeners as
LiveSeries.pushMany of its rows, and the same all-or-nothing
rule (a refused batch stores nothing; the one exception, shared with
pushMany, is a partitionBy router error on a default series,
which comes after the batch is stored). It is how column-shaped data
gets into a live series: build the batch with
TimeSeries.fromColumns (typed arrays in) and push it. "Its rows"
assumes the batch's rows and columns agree, which every validating
constructor guarantees (TimeSeries.fromEvents doesn't check).
Where it is faster. On a default series (ordering: 'strict',
time key) the batch's columns are stored as they are, with no
per-row check and no Event per row, so a large batch costs about
what fromColumns costs: 100k rows in batches of 1,600 took 1.8 ms
against pushMany's 18.1 ms (scripts/perf-live-push-series.mjs).
Two things bring back an Event per row and most of pushMany's
cost: a listener on 'event' or 'batch' (a window(), filter()
or other view built on the series is one; ~1.3-1.5x), and the other
storage mode (ordering: 'drop' / 'reorder', or a timeRange /
interval key), where the batch's rows go through pushMany's row
path (~1x). Results are the same either way. For one row at a time
use push; for batches of a few rows whose arrays are made for that
push (about 8 or fewer), pushMany: there the arrays and a
TimeSeries per batch cost more than they save.
Matching. The batch's key kind must be the live series' (the
key column's name may differ). Value columns are matched by name, in
any order, and must have the live column's kind; a missing or extra
column is refused. (The parameter's type asks for this series'
schema, so in TypeScript the by-name matching matters for a batch
whose schema is only known at run time, passed with a cast.) A gap
in a column the live schema marks required is refused, naming the
first such row, whatever the batch's own schema says. Rows inside the
batch are already in order (a TimeSeries is sorted); against the
rows held, ordering applies as in pushMany.
The batch's buffers are shared, not copied. On a default series
the live series keeps the batch's key and number columns, and
fromColumns keeps the Float64Arrays it was given, so the live
series ends up holding your arrays. Don't write to, reuse, pool or
transfer a pushed Float64Array while its rows are still in the
series. Rows are not checked again, so a write changes rows already
stored and can leave their keys out of order (lookups such as
atOrBefore then go wrong, and a later toTimeSeries() throws),
while a row already read with at() keeps its old value. Transferring
its buffer (postMessage(msg, [buffer])) or shrinking a resizable one
breaks those rows: their values read back as missing, or reads throw.
If you reuse buffers, push buf.slice(0, n) (or a number[], which
fromColumns copies). A push made from inside a listener is queued
and reads the batch when it runs, whatever the storage mode, so leave
its arrays alone until then.
Two kinds of batch are copied for you. Any batch of 8 rows or fewer: for
arrays made fresh for the push, copying costs about what deciding
whether to copy would, and keeps less memory (arrays pushed before pay
~1.5 µs a push for the copy). And a batch that would keep much more
memory alive than its rows need, so a small batch never keeps a big
parent alive: one whose arrays sit in buffers more than twice the size
its rows need, or whose text column carries more than twice as many
distinct strings as it has rows (the copy then keeps only the strings
its rows use). A slice() of a bigger series is like that, since it
shares its parent's buffers and strings, and so is one re-keyed with
asTime({ at: 'center' }). Columns that are views into one buffer about
the batch's size (one worker message holding every column) are kept as
they are.
NaN and other non-finite numbers in a batch from fromColumns
are gaps, because that is how fromColumns reads them (pushMany
refuses one instead): in an optional column they are stored as
missing, in a required one they are refused.
An empty batch is checked for shape and otherwise ignored. A push
made from inside a listener is queued and runs after the current
push, like pushMany's.
rate
rate(columns: Target | readonly Target[], options?: { drop?: boolean }): LiveView<readonly [S[0], ReplaceSmoothedColumn<ValueColumnsForSchema<S>, Target>]>
reduce
reduce(mapping: M, options?: LiveRollingOptions): LiveReduce<S, readonly [S[0], AggregateColumns<ValueColumnsForSchema<S>, M>]>
Streaming counterpart to batch series.reduce(mapping).
Reduces over the source's current buffer — every push
adds to the reducer state, every retention eviction removes.
The snapshot at any moment is the reduction over what's
currently retained.
Same mapping shape as aggregate / rolling; same trigger
options as rolling. The "window" here is implicit — it's
whatever the source retains. For an explicit time-bounded
window, use rolling(duration, mapping, opts) instead.
Returns a LiveSource<Out> whose schema is
[time, ...mappingColumns]. Composes with the rest of the
live operator surface.
rolling
rolling(window: RollingWindow, mapping: M, options?: LiveRollingOptions): LiveRollingAggregation<S, readonly [S[0], AggregateColumns<ValueColumnsForSchema<S>, M>]>
rolling(fusedMapping: FM & FusedMappingValid<FM>, options?: LiveRollingOptions): LiveFusedRolling<S, readonly [S[0], FusedRollingColumns<S, FM>]>
Keyed-form fused multi-window rolling. Maintains N windows in one ingest pass over a single shared deque; emits one merged event per trigger boundary with all windows' columns concatenated.
Use this form when declaring multiple time-windows over the
same source — { '1m': statsMapping, '200ms': samplesMapping }.
Single-window cases keep using the (window, mapping, opts)
shape — both are equivalent for one window, but the legacy
shape is clearer.
Constraints: time-based windows only (object keys are
duration strings); per-window cadence is not supported (single
trigger applies to all windows; users wanting per-window
cadence fall back to two separate rolling() calls). See
PLAN.md "Fused multi-window rolling" for the full rationale.
sample
sample(strategy: SampleStrategy): LiveView<S>
Bounded-memory stream sampling. Thins the event stream going to
downstream consumers without affecting this LiveSeries's own
length, at(i), listeners, or stats() counters.
v0.17.0 ships stride only on the live side — { stride: N },
deterministic 1-in-N, uniform-over-time. Reservoir sampling is
snapshot-side only on this release (TimeSeries.sample); see
SampleStrategy for the rationale (live reservoir's
Algorithm R replacement produces non-prefix evictions; the
existing live-eviction protocol is cutoff-based, so bridging
needs an exact-removal eviction channel arriving with the
streaming RFC's LiveChange model).
Returns a LiveView<S> so the chainable surface
(filter, rolling, reduce, select, …) is immediately
available downstream of the sample.
Multi-entity bias trap. Pre-partition live.sample({stride: N})
applied to a structured input stream (e.g., events arriving in
round-robin host order) silently keeps the same subset of
partitions and drops the rest. The safe shape is to chain after
partitionBy(...), which thins each partition's stream
independently:
// Safe by construction — per-partition counter is implicit
live.partitionBy('host').sample({ stride: 10 }).rolling('5m', m);
Same multi-entity consideration applies to rolling / aggregate /
fill / diff / rate / cumulative / pctChange / reduce:
every stateful live operator silently mixes data across entities
on a multi-entity stream unless scoped per-partition first.
Reducer outputs downstream of sample reflect the sampled
stream; multiply by stride to estimate true counts.
live.stats().ingested continues to count true throughput
upstream of any sample.
select
select(keys: Keys): LiveView<readonly [S[0], PickSelectedColumns<ValueColumnsForSchema<S>, Keys[number]>]>
some
some(predicate: (event: EventForSchema<S>, index: number) => boolean): boolean
live.some(e => e.get('healthy'))True when at least one event matches.
stats
stats(): { earliestTs?: number; evicted: number; ingested: number; latestTs?: number; length: number; rejected: number }
Pipeline stats snapshot — cumulative counters since construction plus current buffer state. Cheap O(1).
ingested: total events accepted (after validation +#insert). Never decreases.evicted: total events removed from the buffer — by retention OR by an explicitLiveSeries.clearcall. Both paths fire the'evict'listener; this counter matches that same fan-out. Never decreases.rejected: total events silently rejected (drop-mode out-of-order arrivals). Strict / reorder modes throw on rejection — those don't count here.length: current buffer size (=this.length).earliestTs/latestTs: timestamps of buffer ends, or undefined if the buffer is empty.
Use case: long-running pipelines that want headline counters
without wiring live.on('batch'/'evict') listeners by hand.
The gRPC experiment's manual-counter pattern is exactly this
shape.
timeRange
timeRange(): number
Time span of the current buffer — last.begin() - first.begin()
in milliseconds. Returns 0 if the buffer is empty or holds a
single event. Useful for the "how much data am I holding right
now?" question that buffer-as-window users ask.
O(1) — reads first/last directly.
toJSON
toJSON(options?: { rowFormat?: 'array' }): TimeSeriesJsonOutputArray<S>
live.toJSON()JSON-shape snapshot of the current
buffer, suitable for sending over a WebSocket or any
JSON.stringify-friendly transport. Sugar over
live.toTimeSeries().toJSON(...).
Defaults to rowFormat: 'array' (tuple rows). Pass
{ rowFormat: 'object' } for schema-keyed object rows. The
return type narrows on the option so consumers don't need to
cast result.rows.
Pairs with LiveSeries.fromJSON for snapshot
reconstruction; pairs with LiveSeries.pushJson for
incremental wire ingest.
toJSON(options: { rowFormat: 'object' }): TimeSeriesJsonOutputObject<S>
live.toJSON()JSON-shape snapshot of the current
buffer, suitable for sending over a WebSocket or any
JSON.stringify-friendly transport. Sugar over
live.toTimeSeries().toJSON(...).
Defaults to rowFormat: 'array' (tuple rows). Pass
{ rowFormat: 'object' } for schema-keyed object rows. The
return type narrows on the option so consumers don't need to
cast result.rows.
Pairs with LiveSeries.fromJSON for snapshot
reconstruction; pairs with LiveSeries.pushJson for
incremental wire ingest.
toJSON(options?: { rowFormat?: JsonRowFormat }): TimeSeriesJsonOutputArray<S> | TimeSeriesJsonOutputObject<S>
live.toJSON()JSON-shape snapshot of the current
buffer, suitable for sending over a WebSocket or any
JSON.stringify-friendly transport. Sugar over
live.toTimeSeries().toJSON(...).
Defaults to rowFormat: 'array' (tuple rows). Pass
{ rowFormat: 'object' } for schema-keyed object rows. The
return type narrows on the option so consumers don't need to
cast result.rows.
Pairs with LiveSeries.fromJSON for snapshot
reconstruction; pairs with LiveSeries.pushJson for
incremental wire ingest.
toObjects
toObjects(): readonly (Readonly<Record<string, EventKey | ColumnValue | undefined>>)[]
live.toObjects()Returns the current buffer as an
array of schema-keyed object rows — same shape as
TimeSeries.toObjects. Useful when callers want to read
by column name rather than tuple position; not the input form
to pushMany (which takes tuples).
toRows
toRows(): readonly NormalizedRowForSchema<S>[]
live.toRows()Returns the current buffer as an
array of normalized typed-row tuples — the same shape
pushMany(rows) accepts. Codec-agnostic: each cell carries its
native runtime value (Time/TimeRange/Interval keys,
undefined for missing data, raw scalars for everything else),
so JSON.stringify is one option but not the only one — the
tuple is also what protobuf / msgpack consumers want before
encoding. For a wire-ready snapshot use LiveSeries.toJSON.
toTimeSeries
toTimeSeries(name?: string): TimeSeries<S>
window
window(size: RollingWindow): LiveView<S>