videoflow.messaging package

Submodules

videoflow.messaging.grouping module

Input-group assembly for multi-parent (join) nodes, factored out of the messenger so the two grouping strategies are testable without a broker:

• TraceGroupAssembler — groups by exact trace_id match: inputs that descend from the same originating message of a single upstream producer (diamond topologies). This is the historical behavior.

• TimeGroupAssembler — groups by event time: inputs whose event_ts fall within a tolerance of each other, regardless of lineage. This is how streams from independent producers (multiple cameras, sensors) are fused, with optional quorum emission (>= k of N parents at timeout) and per-parent “collect” windows for high-rate sensor parents that join many-to-one.

An assembler is fed decoded envelope entries (see videoflow.wire.serialization.decode_envelope) paired with their broker ack handles. It owns the pending buffers and resolves the handles of anything it discards (evicted, superseded, expired); handles of everything it emits travel out unresolved inside the ReadyGroup for the task loop to ack/fail after processing — preserving ack-after-process semantics end to end.

This module also owns the internal record types for those entries. decode_envelope returns a plain dict and must keep doing so — it is re-exported through the frozen videoflow.serialization shim, where a dead-lettered payload recorded under the old module path still has to decode. So the dict is adapted once, where messaging first receives it (NATSMessenger._pull_loop), into an EnvelopeEntry, and everything from there down the join/EOS path uses typed attributes. A collected window is a genuinely different record — its message/metadata/event_ts are lists over a parent’s samples and it has no lineage of its own — so it is a separate CollectEntry rather than an EnvelopeEntry with lying types.

All methods are called from the single task thread; no locking is needed.

class videoflow.messaging.grouping.CollectEntry(message: list = <factory>, metadata: list = <factory>, event_ts: list = <factory>, is_stop_signal: bool = False)

Bases: object

The window of samples a collect parent contributes to one time group: every sample within that parent’s window of the group’s time, oldest first.

Deliberately not an EnvelopeEntry — message, metadata and event_ts are lists here, and the window has no single trace id or seq of its own (which is why last_input_info reports None for both).

event_ts: list

is_stop_signal: bool

message: list

metadata: list

class videoflow.messaging.grouping.EnvelopeEntry(trace_id: str, seq: int, event_ts: float | None, message: Any, metadata: dict | None, is_stop_signal: bool, type: str = '', producer_name: str = '', flow_id: str = '', run_id: str = '', span_id: Any = None, parent_span_id: Any = None, replica_id: int = 0, blob_ref: str | None = None)

Bases: object

One decoded envelope as the messaging layer carries it: the exact field set videoflow.wire.serialization.decode_envelope returns, as attributes.

Built at the messenger’s receive boundary via from_decoded and never mutated afterwards — an assembler that supersedes a redelivery swaps the whole entry rather than editing one.

blob_ref: str | None

event_ts: float | None

flow_id: str

classmethod from_decoded(decoded: dict[str, Any]) → EnvelopeEntry

Adapts a decode_envelope result to an EnvelopeEntry.

• Arguments:

  • decoded (dict): the envelope dict from videoflow.wire.serialization.decode_envelope.

• Returns:

  • The equivalent EnvelopeEntry. Only the fields the join path actually reads are required; the rest are tolerated as absent (carrying their declared default) so a caller can hand over a partial envelope.

• Raises:

  • KeyError: if trace_id or seq is missing — without them a group has no identity to assemble on.

is_stop_signal: bool

message: Any

metadata: dict | None

parent_span_id: Any

producer_name: str

replica_id: int

run_id: str

seq: int

span_id: Any

trace_id: str

type: str

class videoflow.messaging.grouping.GroupAssembler(node_name: str, parent_names: list[str], policy: JoinPolicy)

Bases: object

Base interface: feed entries with add, expire with sweep, drain with pop_ready.

add(parent_name: str, entry: EnvelopeEntry, handle: Any) → None

has_pending_from(parent_name: str) → bool

Whether any buffered state still holds a message from this parent (EOS drain check).

pop_ready(now: float | None = None) → ReadyGroup | None

sweep(now: float | None = None) → None

Apply the policy’s timeout to pending groups (evict or stage for quorum emission).

videoflow.messaging.grouping.GroupEntry = videoflow.messaging.grouping.EnvelopeEntry | videoflow.messaging.grouping.CollectEntry | None

a normal entry, a collected window, or None for a parent missing from a quorum emission.

Type:

What one parent contributes to a ReadyGroup

class videoflow.messaging.grouping.ReadyGroup(trace_id: str, seq: int, event_ts: float | None, entries: dict[str, EnvelopeEntry | CollectEntry | None], handles: list)

Bases: object

A fully assembled input group, ready to hand to the node.

• Attributes:

  • trace_id: identity the node’s output will carry forward (a parent trace id for trace groups; a minted time-window id for time groups).

  • seq: deterministic representative sequence number (stable across redelivery of the same group, for output dedup).

  • event_ts: the group’s event time (min over members), carried forward.

  • entries: {parent_name: entry} where entry is an EnvelopeEntry, None for a parent missing from a quorum emission, or a CollectEntry for a collect parent’s window.

  • handles: every unresolved ack handle backing this group, in no particular order.

class videoflow.messaging.grouping.TimeGroupAssembler(node_name: str, parent_names: list[str], policy: JoinPolicy)

Bases: GroupAssembler

Groups by event time. A message from a synchronized parent joins the pending group whose time is nearest to its event_ts (and within tolerance_ms) among groups not already holding that parent; otherwise it seeds a new group. Messages from collect parents are buffered per parent and attached as a list at emission time (every sample within that parent’s window of the group’s time).

A group is ready when every sync parent is present — held for the largest collect window first, so trailing high-rate samples can arrive — or when the policy timeout expires and at least quorum sync parents are present (missing parents emit as None). A timed-out group below quorum is evicted per the missing policy.

Messages without an event_ts (pre-v3 upstream) fall back to their arrival time — correct enough for co-located low-latency flows, but real deployments should stamp at the producer.

add(parent_name: str, entry: EnvelopeEntry, handle: Any) → None

has_pending_from(parent_name: str) → bool

Whether any buffered state still holds a message from this parent (EOS drain check).

pop_ready(now: float | None = None) → ReadyGroup | None

sweep(now: float | None = None) → None

Apply the policy’s timeout to pending groups (evict or stage for quorum emission).

class videoflow.messaging.grouping.TraceGroupAssembler(node_name: str, parent_names: list[str], policy: JoinPolicy)

Bases: GroupAssembler

Groups by exact trace_id. A group is ready when every parent’s half with the same trace id has arrived; a group that outlives the policy timeout is evicted per the missing policy, and the oldest group is evicted (as drop) beyond max_pending.

add(parent_name: str, entry: EnvelopeEntry, handle: Any) → None

has_pending_from(parent_name: str) → bool

Whether any buffered state still holds a message from this parent (EOS drain check).

pop_ready(now: float | None = None) → ReadyGroup | None

sweep(now: float | None = None) → None

Apply the policy’s timeout to pending groups (evict or stage for quorum emission).

videoflow.messaging.grouping.make_assembler(node_name: str, parent_names: list[str], policy: JoinPolicy) → GroupAssembler

videoflow.messaging.nats_messenger module

NATS JetStream-backed implementation of videoflow.core.engine.Messenger.

One JetStream stream per node (subject vf.{flow_id}.{node.name}); a node’s messenger publishes only its own output there. Each real parent gets its own durable pull consumer, named after the consuming node so that replicas of the same consuming node (nb_tasks > 1) share one durable name (competing consumers / load balancing), while distinct children of the same parent get distinct durable names (each gets its own full copy — broadcast fan-out).

Runs its own asyncio event loop on a background thread so the rest of the framework (videoflow.core.task) can stay synchronous, matching the blocking Queue.get()-style calls the local engine used to make.

class videoflow.messaging.nats_messenger.NATSMessenger(node: Node, parent_names: list[str], nats_url: str, flow_id: str, flow_type: str, run_id: str, blob_store: BlobStore | None = None, replica_id: int = 0, ack_wait: int = 60, max_retries: int = 3, eos_quiescence_ms: int = 500, nb_tasks: int = 1, partition_by: str | None = None, join_policy: dict | None = None, envelope_version: int | None = None, blob_readers: int | None = None, blob_ttl_seconds: int | None = None)

Bases: Messenger

• Arguments:

  • node: the videoflow.core.node.Node this messenger is bound to.

  • parent_names ([str]): the real parents of node, by .name.

  • nats_url (str): e.g. nats://localhost:4222.

  • flow_id (str): shared across every node in the flow.

  • flow_type (str): videoflow.core.constants.REALTIME or BATCH — controls the stream retention/discard policy used for node’s own output stream.

  • blob_store: optional videoflow.wire.serialization.BlobStore for payloads over the inline size threshold.

  • blob_readers (int): how many downstream reads each message this node publishes receives (Σ over children of nb_tasks if partitioned else 1, computed by the compiler); enables refcounted blob reclamation (BLOB-5). None disables it (blobs are TTL-only).

  • blob_ttl_seconds (int): TTL for offloaded payloads; None picks the flow-type default (3600s realtime / 86400s batch, BLOB-7).

  • join_policy (dict): serialized videoflow.core.policies.JoinPolicy controlling how multi-parent input groups are formed (by trace id or by event time) and expired; defaults per flow type when unset. The policy’s max_pending bounds how many not-yet-complete groups are held in memory before the oldest is evicted.

ack_inputs() → None

Acknowledge the input group last returned by receive_message — the node processed it successfully (and, for a processor, already published its output).

check_for_termination() → bool

Returns true if a flow-wide termination signal has been received on the control channel. Used by videoflow.core.task.ProducerTask to stop pulling new input even before it naturally reaches StopIteration.

close() → None

Release any broker resources held by the messenger. Default: no-op.

fail_inputs(exc: BaseException) → None

The node raised while processing the last input group. REALTIME drops it (no redelivery — freshest wins). BATCH nak’s it for redelivery until it exhausts max_deliver, then dead-letters it and terminates it so it stops being redelivered.

last_input_info() → dict[str, dict | None] | None

Per-parent envelope info (event_ts, metadata, trace_id, seq) for the input group last returned by receive_message; None entries for parents missing from a quorum emission. None for producers.

last_input_key() → str | None

A stable identity for the input group last returned by receive_message, derived from its trace_id + seq — used as an idempotency key by a sink. The same logical event yields the same key across redelivery/restart.

publish_message(message: Any, metadata: dict | None = None) → None

Publishes this node’s own output message. Depending on the flow’s configured retention policy (REALTIME vs BATCH), this may drop the message if downstream consumers are behind.

publish_stop_signal() → None

Publishes a termination marker on this node’s own subject. Unlike publish_message, this is never dropped regardless of retention policy — every downstream consumer must observe it exactly once.

receive_message() → dict

Blocks until this node has received a complete input: one message from every real parent, all derived from the same upstream originating event (see the trace_id propagation scheme in the concrete implementation) — or until every parent has signaled termination.

• Returns:

  • a dict {parent_name: {"message": ..., "metadata": ..., "is_stop_signal": bool}} with exactly one entry per real parent of this node.

set_output_event_timestamp(value: float) → None

Set the event time (epoch seconds) stamped on this node’s next published output (via RuntimeContext.set_event_timestamp) — when the underlying real-world event was captured. Producers of time-sensitive data (cameras, sensors) should set this; downstream nodes inherit their input group’s event time automatically. Default: no-op.

set_output_partition_key(value: Any) → None

Set the partition key attached to this node’s next published output (via RuntimeContext.set_partition_key), so a downstream partitioned node can route by a business key. Default: no-op.

videoflow.messaging.topology module

Broker topology: naming, JetStream stream/consumer configuration, and up-front provisioning of a flow’s streams and durable consumers.

Everything that decides how a message routes — subject names, stream names, durable names, and the retention/discard policies that give REALTIME vs BATCH their semantics — lives here, so the messenger, the compiler, the manifests, and the provisioning entrypoint all agree on one source of truth.

Naming is scoped by flow_id and run_id so that re-running or redeploying a flow gets a fresh set of streams instead of colliding with the previous run’s durables.

videoflow.messaging.topology.DEFAULT_MAX_RETRIES = 3

Default number of times a BATCH message is retried (redelivered) after the first delivery attempt before it is dead-lettered. max_deliver = retries + 1.

videoflow.messaging.topology.DEFAULT_REALTIME_BUFFER = 1

REALTIME keeps only the freshest N messages per node and never blocks the producer (a new publish evicts the oldest). BATCH uses INTEREST retention so acked messages are freed, bounding the backlog to unacked messages; a full stream then rejects new publishes (DiscardPolicy.NEW), which the publisher turns into blocking backpressure instead of silent loss.

videoflow.messaging.topology.consumer_config_for(flow_id: str, run_id: str, consumer_node_name: str, parent_node_name: str, ack_wait: int = 60, max_deliver: int = 1, max_ack_pending: int = 8) → nats.js.api.ConsumerConfig

Durable pull-consumer config for one (child, parent) edge. Filters to the parent’s data subject so EOS markers (on the _eos subject of the same stream) are handled by a separate per-replica consumer instead. max_deliver is 1 for REALTIME (no redelivery — freshest wins) and retries + 1 for BATCH; max_ack_pending bounds how many un-acked messages the broker will hand out before it stops delivering (this is the server-side half of prefetch bounding).

videoflow.messaging.topology.control_subject_for(flow_id: str, run_id: str) → str

async videoflow.messaging.topology.delete_run_streams(nc: nats.aio.client.Client, flow_id: str, run_id: str) → None

Best-effort teardown: delete every stream belonging to this run.

videoflow.messaging.topology.dlq_stream_config(flow_id: str, run_id: str) → nats.js.api.StreamConfig

videoflow.messaging.topology.dlq_stream_name(flow_id: str, run_id: str) → str

videoflow.messaging.topology.dlq_subject_for(flow_id: str, run_id: str, node_name: str) → str

videoflow.messaging.topology.durable_name_for(consumer_node_name: str, parent_node_name: str) → str

videoflow.messaging.topology.eos_anchor_config(flow_id: str, run_id: str, node_name: str) → nats.js.api.ConsumerConfig

Provision-time durable on a node’s EOS subject that exists purely to create interest, so an EOS marker is retained by the BATCH (INTEREST-retention) stream no matter when it is published.

The real EOS consumers are per-process (uuid-suffixed) durables created in each worker’s setup — they cannot be pre-provisioned, so without this anchor an EOS published by a fast-finishing parent before a slow-starting child registers its EOS consumer is silently discarded (no interest at publish time), and the child then waits for EOS forever: the flow never terminates. The anchor is never fetched from and never acks, so the marker stays retained for any number of late-created consumers (their default DeliverPolicy.ALL replays it); the run’s stream teardown deletes the anchor with everything else. No inactive_threshold: it must not be reaped while the run is alive.

videoflow.messaging.topology.eos_anchor_durable_name_for(node_name: str) → str

The provision-time interest anchor on a node’s EOS subject (see eos_anchor_config).

videoflow.messaging.topology.eos_consumer_config(flow_id: str, run_id: str, consumer_node_name: str, parent_node_name: str, instance_id: str, inactive_threshold: int = 3600) → nats.js.api.ConsumerConfig

Per-replica durable pull-consumer for a parent’s EOS subject. inactive_threshold lets the server clean it up automatically some time after the flow ends, so per-process (uuid-suffixed) EOS consumers don’t accumulate.

videoflow.messaging.topology.eos_durable_name_for(consumer_node_name: str, parent_node_name: str, instance_id: str) → str

videoflow.messaging.topology.eos_subject_for(flow_id: str, run_id: str, node_name: str) → str

videoflow.messaging.topology.max_deliver_for(flow_type: str, max_retries: int = 3) → int

REALTIME never redelivers (freshest wins; a failed frame is dropped), so max_deliver is 1. BATCH redelivers up to max_retries times, then the message is dead-lettered.

videoflow.messaging.topology.partitioned_durable_name_for(consumer_node_name: str, parent_node_name: str, replica_id: int) → str

async videoflow.messaging.topology.provision_flow(nc: nats.aio.client.Client, specs: list[NodeSpec], flow_id: str, run_id: str, flow_type: str, max_retries: int = 3, ack_wait: int = 60, max_ack_pending: int = 8) → None

Idempotently create every stream and durable consumer a flow needs, before any worker publishes. Required for BATCH: under INTEREST retention a message published with no registered consumer interest is discarded immediately, so the durables must exist first.

• Arguments:

  • nc: a connected nats client.

  • specs: list of videoflow.core.compiler.NodeSpec.

async videoflow.messaging.topology.provision_flow_connect(nats_url: str, specs: list[NodeSpec], flow_id: str, run_id: str, flow_type: str, connect_options: dict[str, Any] | None = None, **kwargs: Any) → None

Connects to NATS, provisions, and drains — a self-contained entrypoint.

• Arguments:

  • connect_options: extra kwargs for nats.connect. The default (retry forever) is right for the in-cluster provision Job, whose broker may still be starting; a local run passes fail-fast options instead so an unreachable broker reports itself rather than hanging.

videoflow.messaging.topology.provision_flow_sync(nats_url: str, specs: list[NodeSpec], flow_id: str, run_id: str, flow_type: str, connect_options: dict[str, Any] | None = None, timeout: float | None = None, **kwargs: Any) → None

Synchronous wrapper for callers outside an event loop (the local engine, the init entrypoint).

• Arguments:

  • timeout: overall bound in seconds, or None (the default) to wait indefinitely. nats.connect retries an unreachable server forever — allow_reconnect/max_reconnect_attempts only govern reconnects after a successful connect — so a caller that would rather report the problem than block must set this.

• Raises:

  • TimeoutError when timeout elapses first.

videoflow.messaging.topology.sanitize(value: str) → str

videoflow.messaging.topology.stream_config_for(flow_id: str, run_id: str, node_name: str, flow_type: str, subjects: list[str] | None = None, realtime_buffer: int = 1, batch_max_msgs: int = 10000) → nats.js.api.StreamConfig

videoflow.messaging.topology.stream_label_selector(flow_id: str, run_id: str) → str

Prefix shared by every stream of one run — used for teardown by name prefix.

videoflow.messaging.topology.stream_name_for(flow_id: str, run_id: str, node_name: str) → str

videoflow.messaging.topology.subject_for(flow_id: str, run_id: str, node_name: str) → str