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', 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]: 'sum' | 'min' | 'max' | 'count' | (acc: number, value: number) => number }): 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(predicate: (event: EventForSchema<S>) => boolean): LiveView<S>
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
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.
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, and a handler that throws mid-batch leaves only the
rows up to the throw committed. 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, and a
handler that throws mid-fan-out leaves the entire batch committed
(the chunk is already appended). Both leave length and ingested
mutually consistent after a throw; they differ only in how much of
the batch is committed. 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'.
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.
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>