arizona_stream (arizona v0.4.0)
View SourceOrdered keyed collection with queued mutations for efficient list diffing.
A stream stores items in a key-indexed map plus an ordered list of keys,
along with a queue of pending operations. Mutations (insert/2,3,
delete/2, update/3, move/3, sort/2, reset/1,2) update the
collection eagerly and append a corresponding op to the queue. The
queue is later drained by arizona_diff:diff/4 to emit minimal patch
operations to the client.
This avoids re-rendering whole lists when only a few items change -- the differ knows exactly which keys to insert, remove, update, or move.
Keys must be unique once rendered, not just as terms: an item is
addressed on the wire and in the DOM by its key's text
(arizona_template:to_bin/1), so keys such as 1 and ~"1" are two
items here but one address to the client, and an op meant for one
patches the other. Use keys of one type.
Complexity -- prefer reset/2 and sort/2 for bulk mutation
items is a map (effectively O(1) per-key access); order is a list with
an O(1) append buffer. Per mutation, over N = stream size:
insert/2(append) andupdate/3-- O(1) amortized, including underon_limit => dropat capacity, where each append also evicts: the eviction pops the front of the order and leaves the append buffer alone, so each key crosses from buffer to front at most oncedelete/2-- O(position of Key in order), keeping the append buffer- positional
insert/3andmove/3-- O(N) list surgery on the order sort/2-- O(N log N);reset/1,2-- O(N)
So K per-item delete/move/insert/3 calls in one event are O(K*N).
For bulk mutation build the new item list once and call reset/2 (a
single op -- the differ still patches only the actual per-item delta), or
reorder everything with one sort/2, instead of K positional calls.
Limit modes
infinity(default) -- no cap, all items remain visible{Limit, halt}-- onceLimititems are present, further inserts are rejected (kept initemsbut not visible){Limit, drop}-- a bounded tail -f buffer: an append (insert/2, or theupdate/3upsert of a missing key) to a full stream evicts the oldest (front-of-order) item from the source to make room, queueing a{delete, Key}for it, so the visible window slides onto the new tail item. Only appends evict -- "oldest" is well-defined only for the append flow; a positionalinsert/3declares a window position rather than an age, and pre-population (new/3,reset/2) sets the whole order at once, so those keephalt's window semantics (items past the limit stay in the source as a hidden tail). An append to such an overfull stream evicts down to the limit, so the stream converges tosize =< Limit.
Pending operation shapes
{insert, Key, Item, Pos}
{delete, Key}
{update, Key, NewItem}
{move, Key, AfterKey} %% AfterKey is `null` for first position
reorder %% from sort/2 when order changes
reset %% reset/1,2 and the genesis of new/1,2,3: reconcile to nowDrain marks -- and the nested-stream memory limit
The queue is drained by the differ but cleared by the live process, which
only reaches streams stored as TOP-LEVEL bindings (clear_stream_pending/2 /
stream_keys/1). A stream nested inside another value -- a field of a parent
stream's item, or a stream inside a map binding -- is out of that reach, so its
queue survives every drain and is re-drained on each re-eval of the enclosing
slot.
Every queued op therefore carries a globally unique stamp. A drain records the
last one it consumed (drain_mark/1) in its snapshot, and the next drain asks
undrained_ops/2 to resume just past it. That makes a re-drain emit exactly the
new ops instead of replaying the whole history -- without it, a replayed
{update, ...} re-rendered its STALE item and emitted a real patch, so op count
and payload grew linearly with the stream's lifetime mutation count (quadratic
per session) and the client applied every historical intermediate value before
the current one.
Resuming is by locating the stamp, never by counting positions: a stream is
an immutable value, so it can have several divergent successors (a view that
keeps a pristine stream and re-derives the rendered one per event -- tab switch,
undo/restore, filter reset). Position counting cannot tell such siblings apart
and would drop a sibling's genuine ops; a stamp a queue does not contain simply
falls back to a full drain. See undrained_ops/2.
What it does not fix: memory. The queue itself keeps growing -- it lives in
app-owned bindings and the differ is pure, so nothing can truncate it. A nested
stream therefore retains every op term (including items a {Limit, drop}
window already evicted) for the lifetime of the view.
Per-cycle CPU does not grow with it: the resume searches backwards from the
newest op and stops at the mark (or below it), so it costs O(ops added this
cycle), not O(history). Returning a single new op measures ~0.1-0.3 us at 100,
1k, 5k and 20k queued ops alike -- flat. (Searching forwards from the head, as
this first did, cost 3.9 / 31 / 136 / 666 us at those sizes.) Draining a rebuilt
queue -- a divergent fork, a reset/1,2 -- is O(that queue), which is the work
of the drain itself rather than of the search.
So the residual cost is retained memory, one op term per mutation. Keep a stream that mutates over a long-lived session as a TOP-LEVEL binding, where the live process clears it after every drain and neither the memory nor the queue grows.
Example
1> S = arizona_stream:new(fun(#{id := Id}) -> Id end).
2> S1 = arizona_stream:insert(S, #{id => 1, name => ~"Alice"}).
3> S2 = arizona_stream:insert(S1, #{id => 2, name => ~"Bob"}).
4> arizona_stream:to_list(S2).
[#{id => 1, name => ~"Alice"}, #{id => 2, name => ~"Bob"}]Summary
Functions
Clears the pending queue on every stream stored under any of the
listed Keys in Bindings. Called by the live process after a render
flushes the queued ops.
Set of keys that differ between two item maps -- added, removed, or value changed. Used by the differ to drive per-item dynamic skipping. Items in Arizona are maps; non-map inputs crash here.
Removes the item with Key and queues a delete op. No-op if Key
is not present.
The mark a drain records in its snapshot: the stamp of the last op in pending,
or none when the queue is empty. Feed it back to undrained_ops/2 on the next
drain of the same slot.
Formats arizona_stream runtime errors into a human-readable message.
Picked up by erl_error:format_exception/3 via the error_info
annotation attached at the raise site.
Returns the item at Key. Errors with missing_stream_key (carrying an
error_info annotation that routes through format_error/2) if not present.
Use get/3 for a default.
Returns the item at Key, or Default if not present.
Like get/3 but the default is computed lazily by a 0-arity fun.
Appends Item to the end of the stream and queues an insert op.
Inserts Item at the given zero-based Pos and queues an insert op.
Moves the item with Key to a new zero-based position and queues a
move op. No-op if Key is not present, or already at that position (no op
is queued). The position is clamped to the last index.
Creates an empty stream with the given key function.
Creates a stream pre-populated with Items. Equivalent to
new(KeyFun, Items, #{}).
Creates a stream pre-populated with Items and the given options.
Every op currently in pending, oldest first.
Empties the stream and queues a reset op.
Replaces all items with NewItems and queues a reset op. The differ
will reuse client-side keys that are still present, only emitting ops
for the actual delta.
Reorders the stream by CompareFun. Queues a reorder op only when
the resulting order actually differs.
Returns the binding keys whose values are streams.
Returns all items in display order as a list.
The ops a drain holding Mark has not consumed yet, oldest first.
Replaces the item at Key with NewItem and queues an update op.
Types
-nominal item() :: term().
-nominal key() :: term().
-nominal mark() :: none | integer().
-nominal opts() :: #{limit => pos_integer() | infinity, on_limit => halt | drop}.
-nominal stream() :: #stream{key :: fun((term()) -> term()), items :: #{term() => term()}, order :: {[term()], [term()]}, pending :: queue:queue(), limit :: pos_integer() | infinity, on_limit :: halt | drop, size :: non_neg_integer()}.
Functions
-spec clear_stream_pending(Bindings, Keys) -> Bindings1 when Bindings :: map(), Keys :: [term()], Bindings1 :: map().
Clears the pending queue on every stream stored under any of the
listed Keys in Bindings. Called by the live process after a render
flushes the queued ops.
-spec compute_item_changed(OldItem, NewItem) -> Changed when OldItem :: map(), NewItem :: map(), Changed :: #{term() => true}.
Set of keys that differ between two item maps -- added, removed, or value changed. Used by the differ to drive per-item dynamic skipping. Items in Arizona are maps; non-map inputs crash here.
Removes the item with Key and queues a delete op. No-op if Key
is not present.
The mark a drain records in its snapshot: the stamp of the last op in pending,
or none when the queue is empty. Feed it back to undrained_ops/2 on the next
drain of the same slot.
-spec format_error(Reason, Stacktrace) -> ErrorInfo when Reason :: term(), Stacktrace :: [tuple()], ErrorInfo :: #{general := iolist()}.
Formats arizona_stream runtime errors into a human-readable message.
Picked up by erl_error:format_exception/3 via the error_info
annotation attached at the raise site.
Returns the item at Key. Errors with missing_stream_key (carrying an
error_info annotation that routes through format_error/2) if not present.
Use get/3 for a default.
-spec get(Stream, Key, Default) -> item() | Default when Stream :: stream(), Key :: key(), Default :: term().
Returns the item at Key, or Default if not present.
-spec get_lazy(Stream, Key, Fun) -> item() | term() when Stream :: stream(), Key :: key(), Fun :: fun(() -> term()).
Like get/3 but the default is computed lazily by a 0-arity fun.
Appends Item to the end of the stream and queues an insert op.
In drop limit mode an append to a full stream first evicts the oldest
(front-of-order) item(s) from the source, so the visible window slides
onto the new tail item (see the moduledoc's limit-modes section).
Crashes with {stream_duplicate_key, Key} (carrying an error_info
annotation that routes through format_error/2) if Key is already
present -- inserting a duplicate key would duplicate it in order and
corrupt the items/order/size invariant. Use update/3 to replace an
existing item.
-spec insert(Stream, Item, Pos) -> Stream1 when Stream :: stream(), Item :: item(), Pos :: non_neg_integer(), Stream1 :: stream().
Inserts Item at the given zero-based Pos and queues an insert op.
Crashes with {stream_duplicate_key, Key} (see insert/2) if Key is
already present. Use update/3 to replace an existing item.
-spec move(Stream, Key, NewPos) -> Stream1 when Stream :: stream(), Key :: key(), NewPos :: non_neg_integer(), Stream1 :: stream().
Moves the item with Key to a new zero-based position and queues a
move op. No-op if Key is not present, or already at that position (no op
is queued). The position is clamped to the last index.
Creates an empty stream with the given key function.
The queue opens with a genesis reset (see new/3).
Creates a stream pre-populated with Items. Equivalent to
new(KeyFun, Items, #{}).
-spec new(KeyFun, Items, Opts) -> stream() when KeyFun :: key_fun(), Items :: [item()], Opts :: opts().
Creates a stream pre-populated with Items and the given options.
The queue opens with a genesis reset -- a reset op -- so the
first render emits the ops for Items. A constructor states a whole
collection, not a delta from one; queueing the items as pending INSERTS
instead described the delta from empty, which is only a truthful log when
the slot the stream lands in is itself empty. Bind a freshly constructed
stream over a slot that already holds one (rather than mutating the stream
already there) and those inserts are all the differ has to go on: the keys
the client still holds but the new stream dropped are never mentioned, and
a key present in both is skipped as a replay -- so a shrink, a reorder and
an in-place content change all diffed to NOTHING and the stale list stayed
on screen. The reset op is the one that already means "reconcile the client
to my current state", so opening with it makes the log truthful for that
caller without costing the incremental path anything: undrained_ops/2
resumes past it, and a re-drain that replays it is idempotent.
Every op currently in pending, oldest first.
Empties the stream and queues a reset op.
-spec reset(Stream, NewItems) -> Stream1 when Stream :: stream(), NewItems :: [item()], Stream1 :: stream().
Replaces all items with NewItems and queues a reset op. The differ
will reuse client-side keys that are still present, only emitting ops
for the actual delta.
The op carries no items: what a kept item changed is measured against the
item the client last rendered, which the differ keeps on its snapshot. The
source as it stood just before the reset is not that item whenever the same
event touched it first -- an earlier update/3 is in the queue this call
discards.
Crashes with {stream_duplicate_key_in_list, Key} when NewItems names a key
more than once -- the bulk counterpart of insert/2's duplicate refusal.
-spec sort(Stream, CompareFun) -> Stream1 when Stream :: stream(), CompareFun :: fun((item(), item()) -> boolean()), Stream1 :: stream().
Reorders the stream by CompareFun. Queues a reorder op only when
the resulting order actually differs.
Returns the binding keys whose values are streams.
Returns all items in display order as a list.
Crashes with {stream_order_stale_key, K, Order, ItemKeys} carrying
error_info when the items <-> order invariant is broken (a key in
order is not in items). The named reason routes through
format_error/2, so the dev error page renders a sentence instead
of the raw {badkey, _}.
The ops a drain holding Mark has not consumed yet, oldest first.
Mark is none (no previous drain) or a drain_mark/1 from one. The op it
names is located in pending and everything up to and including it is dropped;
if it is absent the WHOLE queue is returned, i.e. a full drain.
Locating the mark (rather than counting positions off the head) is what keeps
this exact for a stream with more than one successor. Stamps are minted once,
at one append, onto one queue value, and a queue is only ever extended at the
tail or replaced wholesale -- so any two queues containing the same stamp
descend from the same value and share the identical prefix ending at it.
Dropping that prefix is therefore exactly what the previous drain consumed.
Anything else -- a divergent fork of the same stream, a reset/1,2, a
clear_stream_pending/2, a rebuilt or rolled-back stream -- simply does not
contain the stamp and falls back to a full drain. A queue whose every stamp is
NEWER than the mark needs no fallback: the backwards search has already read
every op by the time it runs out, so its accumulator is that full drain.
The search runs BACKWARDS from the newest op and costs O(ops newer than Mark),
not O(queue length): the mark is normally the previous drain's last op, so a
re-drain walks only what this cycle actually added, however long the history
behind it. Because stamps ascend along a queue, dropping below Mark proves it
is absent and abandons the search there. Both exits are safe by construction --
failing to find the mark yields a full drain, which never skips an op.
-spec update(Stream, Key, NewItem) -> Stream1 when Stream :: stream(), Key :: key(), NewItem :: item(), Stream1 :: stream().
Replaces the item at Key with NewItem and queues an update op.
If Key is not present it is appended as a new item (an upsert): the
differ already renders a queued update of an unseen key as an insert,
so appending here keeps that op consistent with the items/order/size
invariant. Being an append, the upsert evicts in drop limit mode
exactly like insert/2.