Broker

Functions

.broker.publish

Publish data for table. This only serialize the data once and publish to all subscribers.

.broker.subscribe

Subscribe a handle to topics and return the replay bound: the value of tick[0; 0] at the moment the handle was registered.

The registration and the bound are one atomic step under the same lock lpt takes, so a subscriber never misses or double-counts a message:

• every message logged before the call is at or below the bound and comes from replay;
• every message published after it is buffered for the handle and written right after the reply to the subscribe request, in order, before any later message.

Hand the returned bound to the subscriber (see .tick.subscribe in the cookbook). Do not read tick[0; 0] in a separate statement: a message published in between would be in neither the replay nor the live stream.

When called for a handle other than the calling connection (from the REPL, a job, another connection), the subscription goes live immediately.

.broker.subscribeFiltered

Register a subscriber on one topic with an optional per-handle row filter, and return the replay bound exactly as .broker.subscribe does. Empty values means no filter (same as unfiltered subscribe for that topic).

.broker.unsubscribe

Unsubscribe from a table for handle.

.broker.validateSeq

Fix a corrupted sequence file by removing a torn trailing message, and return the number of valid messages in the file.

Only a structurally incomplete tail is removed: a short header, a short payload, or a length larger than the rest of the file. With must_deserialize, a complete message that cannot be decoded is reported in the log and kept, along with everything after it (replay skips such a message and applies the rest).

Accepts whole-file gzip sequence logs (.seq.gz or gzip magic) without gunzip-to-temp.

.broker.validateSeqStrict

Strict variant of validateSeq. Returns an error on any corrupt or truncated frame instead of silently truncating. Opens the file read-only. Also accepts whole-file gzip sequence logs.

.broker.eod

Publish an EOD message to all subscribers.

.broker.list

List all topics with subscriber handles.

Slow subscribers

Every subscriber gets its own outbound queue and writer thread, so .broker.publish and lpt never wait on a slow or stalled consumer. A backlog costs the publisher memory, not time.

By default the queue has no bound. Bounds are a policy checked after each publish, set from the embedding API (ChiliEngine in Python):

A shed subscriber is disconnected; its .handle.onDisconnected callback can reconnect and recover. stats() reports queue_depth_by_handle and queue_bytes_by_handle so a backlog can be watched before it matters.

Subscriptions of a disconnected, closed or shed handle are removed from .broker.list automatically.

Other Functions

replay

Replay a sequence file. Accepts whole-file gzip (.seq.gz or gzip magic); skip uses read-and-discard so non-seekable streams work. Tolerates torn trailing records — stops at the last valid frame with a warning.

tick

Fetch-and-add on an internal counter: add inc under a process-wide write lock, then return the new (post-increment) value. Counters are independent per index (0..1023). Safe under concurrent inbound IPC connections — concurrent tick calls on the same index do not lose updates.

Supported contract (may be depended on):

• Return value is always the value after applying inc.
• Atomic across concurrent connections / eval threads.
• Negative inc is allowed (used for release / reset patterns).

Use tick[index; 0] / tick(index, 0) to read the current value without changing it. Indices are not reserved for the tickerplant: tick[0; …] is conventionally the message counter, but other slots may be used as application counters (for example a gateway admission gate).

tock

Set a tick counter to an absolute value (not an increment). Prefer this when reseeding after validateSeq instead of the two-step tick[0; neg tick[0; 0]]; tick[0; n] pattern.

lpt

Atomic tickerplant update under one lock: write (`upd; table; data) to handle, .broker.publish, then advance a tick counter. Stock .tick.upd is lpt[table; data; 0; .tick.msgHandle].

Third argument:

• int tick_index — log and publish data as given, then tick[tick_index; 1].
• sym/str stamp_col — under the same lock, stamp that column with tick[0; 0] + i (u64) per row, log/publish the stamped frame, then tick[0; count data].

Fourth argument handle is the log-file handle, bound inside the lock (pass .tick.msgHandle or any other sequence handle, e.g. a per-table tplog).

Returns the new counter value at the tick slot used.