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, expired) and supersedes (retired locally, no broker settlement); 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

evictions

Groups discarded (evicted or expired) so far — read by the messenger’s drop accounting.

has_pending_from(parent_name: str) → bool

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

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

How long the oldest incomplete group has waited (0 when none).

pending_count() → int

Incomplete groups held right now.

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).

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

How long the oldest incomplete group has waited (0 when none).

pending_count() → int

Incomplete groups held right now.

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).

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

How long the oldest incomplete group has waited (0 when none).

pending_count() → int

Incomplete groups held right now.

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.jetstream_backend module

NATS JetStream as a MessagingBackend — the transport half of what NATSMessenger used to be all of.

The split follows the contract in videoflow.backends.messaging: this adapter owns the connection, its asyncio loop thread, the streams and durables it provisions, the pull loops that prefetch deliveries into bounded per-subscription queues, the leases it keeps alive, and the settlement of every delivery token it hands out. It knows nothing about joining inputs, dead-letter policy, payload bytes or end-of-stream drains — those stay in the messenger, which composes this with a payload store.

What the contract buys, concretely:

• A delivery token names the subscription, the message, the broker’s attempt count and the subscription generation. A redelivery of the same message makes the earlier token stale: settling it returns SettleStale without a broker call, so a superseded handle can never TERM the delivery that replaced it (MSG-011).

• Settlement is confirmed or unknown, never assumed. Completed uses ack_sync and reports SettleUnknown when the acknowledgement is not confirmed within the timeout — the caller must not release a payload obligation on that (PAY-004). Terminal requires a durable record reference (DELIV-15).

• Publication outcomes are typed. A deadline that expires after the send is PublicationUnknown; a broker refusal is Rejected with retryable saying whether waiting helps (a full BATCH stream) or not (no stream captures the subject); a lost receipt is never reported as rejection (MSG-013, MSG-014). observe_publication resolves an unknown send by an idempotent re-publish inside the stream’s duplicate window and is Unresolvable beyond it.

• Observations distinguish zero from unknown. observe_subscription is Unknown when the consumer could not be read, and its unresolved count is fed by the server’s own MAX_DELIVERIES advisories — and, through mark_exhausted, by the messenger’s ledger when the retry budget lives there (max_deliver = -1) and a dead letter could not be recorded — so work the broker retains but will never resolve on its own is not reported as “nothing pending” (MSG-012).

• Stop-receive is separate from shutdown. stop_receiving (the messenger’s quiesce) halts the pull loops and hands every parked delivery back at once while the deliveries already handed out stay the receiver’s to settle; shutdown hands back whatever is still held. A SIGTERMed worker runs only the first before it dies, and that is what returns its prefetched inputs to the survivors now instead of after ack_wait (RUN-030).

Everything a NATS API behaviour relies on is cited against nats-py 2.15.0 (.venv/lib/python3.12/site-packages/nats): Msg.ack_sync waits for the server’s reply and marks the message acked only then (nats/aio/msg.py); nak/term/in_progress are fire-and-forget (same file); JetStreamContext.publish raises nats.js.errors.APIError for a server-side refusal and nats.errors.TimeoutError when no PubAck arrives (nats/js/client.py); pull_subscribe creates the durable from config when consumer_info says it does not exist (nats/js/client.py); a JetStream Msg.metadata carries num_delivered and sequence.stream (nats/aio/msg.py).

videoflow.messaging.jetstream_backend.ACK_CONFIRM_SECONDS = 2.0

How long a Completed settlement waits for the server’s ack reply before it is reported SettleUnknown (Msg.ack_sync(timeout), nats-py 2.15.0).

videoflow.messaging.jetstream_backend.FETCH_TIMEOUT_SECONDS = 1.0

How long one pull request waits for a message before the loop re-checks whether it should stop. Also the unit receive_any waits in.

class videoflow.messaging.jetstream_backend.JetStreamMessagingBackend(nats_url: str, flow_id: str, run_id: str, flow_type: str, prefetch: int = 4, fetch_timeout: float = 1.0, ack_confirm_seconds: float = 2.0, keepalive: bool = True, connect_timeout: float = 30.0, byte_budget: int | None = None)

Bases: MessagingBackend

• Arguments:

  • nats_url: the server, e.g. nats://localhost:4222.

  • flow_id / run_id / flow_type: the run this backend serves; every channel and subscription it touches belongs to it.

  • prefetch: per-subscription local prefetch depth (DEFAULT_PREFETCH).

  • fetch_timeout: seconds one pull request waits (FETCH_TIMEOUT_SECONDS).

  • ack_confirm_seconds: how long Completed waits for the broker’s reply.

  • keepalive: whether to extend the lease of every unsettled delivery periodically (ack_wait / 3), as the messenger always has.

  • byte_budget: the most envelope bytes this receiver holds unsettled — parked or handed out — before its pull loops pause (RUN-025; VF_PREFETCH_BYTES). A fetch already in flight when the budget fills still lands, so the resident total may exceed it by one message: the documented atomic-admission tolerance. None: unbounded.

budget_waits() → int

How many times a pull loop paused because the byte budget was full.

capabilities() → MessagingCapabilities

What this broker offers, with the parts that need a read-back reported as observed: max_payload from the connection, replication and storage from the first stream this backend provisioned (ensure_channel), and Unknown('unread') for either until then.

close(owned: Sequence[ChannelId], expected_generation: str) → CleanupObservation

Delete exactly the owned channels’ streams by ownership (topology.delete_run_streams).

property connected: bool

ensure_channel(spec: ChannelSpec, operation_id: str) → VerifiedChannel

Create or reconcile the channel and read back its effective configuration. An immutable mismatch (retention, replicas) raises IncompatibleProfile; “already exists” is never accepted without inspection.

ensure_subscription(spec: SubscriptionSpec, operation_id: str) → VerifiedSubscription

Bind the subscription’s durable (creating it when absent, from a config that reproduces today’s bytes), read it back, and start prefetching into its queue. A data subscription also subscribes to the broker’s MAX_DELIVERIES advisory for its durable, which feeds unresolved.

property loop: AbstractEventLoop

mark_exhausted(subscription: SubscriptionId, stream_sequence: int | None) → None

Count a retained message as unresolved work in observe_subscription without a broker advisory: the messenger’s ledger found its retry budget exhausted and could not record its dead letter (DELIV-15’s pending_handoff), so the broker will keep redelivering it and nothing will resolve it until the record exists. Cleared by the settlement that eventually terminates it, or by the message leaving the stream.

observe_ack_floor(subscription: SubscriptionId) → Known[int] | Unknown

consumer_info().ack_floor.stream_seq of any data durable of the run, bound here or not.

observe_channel(channel: ChannelId) → Known[ChannelObservation] | Unknown

stream_info().state of the channel’s stream: first/last sequence, message count, retention.

observe_publication(envelope: Envelope) → Accepted | Rejected | PublicationUnknown | PublicationUnresolvable

Resolve an earlier PublicationUnknown by an idempotent re-publish: inside the stream’s duplicate window the broker answers duplicate when the first send was stored and stores it now when it was not — either way one accepted copy. Outside the window nothing can be known: Unresolvable.

observe_subscription(subscription: SubscriptionId) → Known[SubscriptionObservation] | Unknown

prefetched(subscription: SubscriptionId) → int

Deliveries fetched from the broker and parked locally for subscription (0 when the backend keeps none).

publish(envelope: Envelope, deadline: float) → Accepted | Rejected | PublicationUnknown | PublicationUnresolvable

Publish with a monotonic deadline; on expiry the outcome is PublicationUnknown, never a guess.

receive(subscription: SubscriptionId, item_credit: int, byte_credit: int, deadline: float) → list[Delivery]

receive_any(subscriptions: Sequence[SubscriptionId], item_credit: int, byte_credit: int, deadline: float) → list[tuple[SubscriptionId, Delivery]]

Waits until deadline (monotonic) for any of the subscriptions’ queues to hold a delivery and returns everything that became ready in that wait — every item asyncio.wait reports done has already been dequeued, so all of them are returned (discarding all but one lost messages once).

renew(token: DeliveryToken) → LeaseObservation

resident_bytes() → int

Envelope bytes this receiver holds unsettled right now (parked or handed out).

set_admission(subscription: SubscriptionId, admit: Callable[[Delivery], bool], on_skip: Callable[[Delivery], None] | None = None) → bool

Ask the backend to run admit on every delivery of subscription before it is parked for receive: a delivery it refuses is settled Completed by the backend at once — so a message this consumer will never process does not occupy its ack window — and, once that settlement is confirmed, handed to on_skip off the backend’s own threads. admit runs on the backend’s thread and must be cheap and must not block. Returns False when the backend keeps no such filter; the caller then decides on the receiving side.

settle(token: DeliveryToken, outcome: Completed | Retry | Terminal, settlement_id: str) → SettleConfirmed | SettleUnknown | SettleStale

shutdown(hand_back: bool = True) → None

Stop the pull loops, hand back every delivery this receiver still holds, flush pending publishes, close the connection and the loop thread.

The hand-back is what makes a scale-down or a rollout cheap: a delivery parked in the prefetch queue, or received and never settled, would otherwise sit leased until ack_wait lapsed — on a replica that no longer exists. A NAK returns it to the broker at once for the survivors. A delivery already settled is left alone (MsgAlreadyAckdError).

• Arguments:

  • hand_back: False closes without the NAKs — what a crashed process looks like to the broker (its leases lapse on their own). Tests simulating a crash pass it; a worker never does.

skipped(subscription: SubscriptionId) → int

Deliveries the admission filter refused and this backend acked on the consumer’s behalf.

start() → None

Connect on a fresh loop thread. Raises BrokerUnavailable when the server cannot be reached.

stop_receiving() → int

Stop admitting input on every bound subscription — the quiesce half of a graceful retirement (RUN-030): the pull loops fetch nothing more, and every delivery parked in a prefetch queue (fetched, never handed to a receiver) is NAKed back to the broker at once, so a survivor picks it up now rather than after ack_wait lapses on a process that is about to die. The deliveries already handed out stay this receiver’s to settle (shutdown hands back what is still held then). Idempotent; a second call hands back nothing more.

• Returns:

  • how many parked deliveries were handed back.

subject_for(channel: ChannelId, kind: str) → str

subscribe_control(callback: Callable[[], None]) → None

Invoke callback (on the loop thread) when the run’s flow-wide stop is published.

supersede(token: DeliveryToken) → bool

Retire a delivery token locally — the receiver now holds a newer attempt of the same logical message — without any broker settlement. Returns whether the token was live. A later settle of the retired token is SettleStale. The default is a no-op for adapters whose tokens are already fenced by attempt and generation.

videoflow.messaging.jetstream_backend.channel_spec_for(flow_id: str, run_id: str, node: str, flow_type: str, profile: str, required: Sequence[SubscriptionId] = (), realtime_buffer: int = 1, batch_max_msgs: int = 10000, replicas: int = 1, max_bytes: int | None = None) → ChannelSpec

The channel a node’s output stream must provide, from the flow type — today’s stream settings, as a spec. replicas is the copies a replicated broker keeps (STREAM-14); max_bytes caps the stream’s storage (a capacity fixture); both default to today’s unset values.

videoflow.messaging.jetstream_backend.consumer_credit(nb_tasks: int, partitioned: bool, item_credit: int = 1, prefetch: int = 4) → int

STREAM-15: the broker-side max_ack_pending a subscription needs so every replica can hold item_credit inputs in processing plus prefetch parked, without one replica’s un-acked work starving another’s — a shared (competing) durable multiplies by the replica count, a per-replica (partitioned) durable does not. Provisioning and the worker call this so the durable created and the durable bound agree (MSG-017).

videoflow.messaging.jetstream_backend.stream_config_for_spec(spec: ChannelSpec, generation: str | None = None) → nats.js.api.StreamConfig

The JetStream StreamConfig a channel spec means — byte-identical to today’s for a default spec.

videoflow.messaging.jetstream_backend.subscription_spec_for(subscription: SubscriptionId, ack_wait_seconds: float, max_deliver: int, credit: int, byte_credit: int = 0) → SubscriptionSpec

A subscription spec: credit is the broker-side max_ack_pending this subscription is bound with.

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).

The transport itself — connection, loop thread, streams, durables, prefetch, leases, settlement, publication outcomes — lives in videoflow.messaging.jetstream_backend.JetStreamMessagingBackend behind the MessagingBackend contract; this class composes it with a payload store and keeps what is the messenger’s to decide: trace and sequence bookkeeping, join assembly, the end-of-stream drain, the delivery-policy ladder, dead-lettering, and when a payload obligation is released (only after a confirmed settlement). Every delivery is a DeliveryToken from the backend, so a superseded handle can never terminate the attempt that replaced it, and every payload is fetched on the receiving thread after ownership is decided, never on the broker loop.

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, delivery_policy: dict | None = None, backend: MessagingBackend | None = None, payload_store: PayloadStore | None = None, blob_reader_ids: list[str] | None = None, runtime: FlowRuntime | None = None, replayable: bool = False, prefetch_bytes: 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 (the RFC 0002 counter store).

  • 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.

  • backend: the MessagingBackend to compose; a JetStreamMessagingBackend on nats_url when None.

  • payload_store: an obligation-keeping PayloadStore (RFC 0006). Used instead of blob_store when given: puts acquire the reader obligations in blob_reader_ids and the publisher’s intent, releases happen by reader id after a confirmed settlement (BLOB-14).

  • blob_reader_ids ([str]): the reader obligations every put acquires (VF_BLOB_READER_IDS): <child> per competing child, <child>/p<i> per partitioned replica.

  • runtime: the node’s FlowRuntime ledger (RFC 0006 CTRL-4): terminators and received sets, the outbox, attempt counts, the terminal log, pending dead-letter handoffs, open groups. A memory-backed one when None — every record then lives and dies with this process, and the messenger knows it (runtime.durable_shared()).

  • replayable (bool): this producer mints ids from its source offset (MSGID-6) and checkpoints the last accepted one; a live source mints {node}:{epoch}:{n} (MSGID-5). Under the switch only.

  • prefetch_bytes (int): the most envelope bytes the default backend holds unsettled before it stops fetching (VF_PREFETCH_BYTES, RUN-025); None leaves it unbounded.

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.

checkpoint(state: bytes) → None

One ledger write: the node’s state and the group it was updated with (last_input_key), so a replacement’s restored state and the inputs it must still see describe the same committed prefix (RUN-022). The group is remembered as covered: if it is redelivered to this process’s replacement, it is acknowledged without being handed to the node again.

This is the immediate form, for a node whose input produces no output (a sink, a leaf): its checkpoint is the commit. A node with an output hands its checkpoint to publish_message instead (RuntimeContext), and _publish commits state and output together. A superseded owner is refused (RUN-023); a ledger that dies with the process is refused too, rather than faking durability (RUN-022).

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. The action is decided by DeliveryPolicy.action_for from how the error classified and how many times the broker has delivered it — not by the flow type alone. This messenger only executes the verdict.

The difference that matters: a poison message is dead-lettered on its first failure rather than burning four attempts on its way to the same place, and a worker-fatal error naks without dead-lettering, because the message is fine and this worker is not. A delivery is terminated only against a durable record: the dead letter’s id when one was accepted, an entry in this node’s terminal log otherwise (DELIV-15); a dead-letter publish that failed keeps the delivery alive for a later attempt.

join_status() → dict[str, Any]

What the join is doing right now (RUN-006): how many groups wait for a missing member, how long the oldest has waited, and whether the policy would ever give up on them (missing = 'wait' never does — a node in that state is waiting, which the health seam must not report as healthy processing progress). Cancellation is the control stop: quiesce/the flow-wide stop ends the wait and hands the halves back.

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.

pending_count() → int

Messages waiting for this node across all its parents — locally prefetched, held in incomplete join groups, or still on the broker. Parents whose broker query failed contribute only their local queue; use pending_observation where “failed” must not read as “zero”.

pending_observation() → Known[int] | Unknown

pending_count as an observation: Unknown as soon as any parent’s broker query failed, because a total that silently omits one parent is exactly the “zero pending” that declared a stalled node idle. Feeds ProgressDeadline, which neither resets nor trips on Unknown.

publication_stats: dict[str, int]

accepted, duplicate, unknown, dropped, rejected.

Type:

Publication outcomes by kind

publish_abort(error: Any) → None

Publishes an abnormal end-of-stream carrying why this node died.

Rides the same _eos subject as a clean EOS, which is the point: it reuses the per-replica EOS consumers and the provisioning interest anchor unchanged, so a marker published by a dying node is still retained and still reaches every downstream replica. Its dedup id is distinct from the clean marker’s so a node that aborts is never mistaken for one that finished.

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

Publish one data output. metadata may carry the node’s pending checkpoint under the reserved CHECKPOINT_METADATA_KEY (the task puts it there, RuntimeContext.checkpoint); it is taken off here, committed with the output in one ledger write, and never reaches the wire.

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.

quiesce() → None

Stop admitting input (SIGTERM, a scale-down, a rollout): what is already held by the task still settles normally, and what the adapter prefetched but never handed out goes back to the broker at once, so a survivor picks it up now rather than after ack_wait lapses on a process that is about to die (RUN-030). Idempotent. A quiesced process publishes no EOS: its replacement continues the stream (stop_reason).

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: entry} with exactly one entry per real parent of this node. Each entry carries message, metadata, event_ts, and two termination flags: is_stop_signal (that parent ended cleanly) and is_abort (it died). An aborted entry also carries abort_origin and abort_error — the originating node and its error record — so the failure can be reported and relayed downstream rather than degenerating into a hang.

restore_checkpoint() → bytes | None

The last checkpointed state, and the group it covers: a redelivery of that group is acknowledged, not re-applied (receive_message) — after its committed output, if the checkpoint carries one, is confirmed on the broker.

resume_offset() → int

For a replayable producer: the last accepted source offset in the ledger, so the source resumes from the next one after a restart (MSGID-6). 0 when nothing was accepted (or the node is live).

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.

stop_reason() → str | None

Which stop check_for_termination reports, one of the STOP_* constants, or None while the flow runs. Default: the control stop whenever a termination was signalled — the only kind a bare messenger has.

subscription_status() → dict[str, Known[Any] | Unknown]

Per parent, the backend’s observation of this node’s data subscription — available, leased, unresolved (retained but exhausted: never “all done”, MSG-012), dropped — or Unknown when it could not be read.

take_drops() → dict[str, int]

Inputs this messenger gave up on since the last call, by reason, that no caller above the Messenger seam could have seen: a retry budget exhausted (exhausted), bytes that could not be decoded (undecodable), a join group evicted (join_evicted), a live publication the broker refused (publish_discarded).

terminal_entries() → list[dict]

The node’s terminal log (DELIV-15): every delivery it ended without a dead letter, from the ledger.

videoflow.messaging.obligations module

The runtime’s own obligation ledger (RFC 0006 BLOB-14 step 4): what a worker derives, at start and periodically, from durable records it already keeps and from the broker’s own facts — no test-built ledger, no counter.

For every publication a parent of this node made (the parent’s outbox in the run ledger: the payload refs, the reader obligations the put acquired, the accepted stream sequence), the ledger decides which readers still owe a release:

• a publication the channel evicted (its sequence is below the channel’s first retained sequence, observe_channel) will never reach a reader that has not taken it — nobody owes it anything (PAY-012);

• otherwise a reader owes its obligation while its ack floor on that channel is below the sequence (observe_ack_floor); a floor at or past it means the reader settled the message, and a release that never happened (a crash after a confirmed ack, PAY-006) is what reconciliation cancels;

• an intent/<publication_id> is owed while the publication is unresolved; a definitely refused one owes nothing (PAY-007, PAY-013).

Anything that could not be observed — a floor, a channel — is required: an unobservable release is not a release. The ledger is authoritative only for the reader ids it knows and for publishers’ intents; an archive’s or a dead-letter pin’s obligation (archive/*, dlq/*) is never cancelled by it.

class videoflow.messaging.obligations.RuntimeObligationLedger(runtime: FlowRuntime, backend: MessagingBackend, channels: Sequence[str])

Bases: ObligationLedger

• Arguments:

  • runtime: the node’s ledger (its store is shared with its parents’ outboxes).

  • backend: the transport, for channel retention and ack floors.

  • channels: the publishing nodes whose obligations to judge — this node’s

    parents, and itself for its own intents.

authoritative(obligation_id: str) → bool

Whether this ledger may cancel obligation_id when it does not require it.

required_obligations() → Mapping[str, tuple[str, ...]]

ref key -> obligation ids still required.

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.

Ownership is a separate question from naming, and this module answers it too. Names are hyphen-joined and hyphens are legal inside every part, so a stream name alone cannot say where the run ends and the node begins: vf-f-r-x-n is node x-n of run r and node n of run r-x. Two things make ownership exact anyway. Every stream and durable is created with owner labels in its JetStream metadata (videoflow.io/flow-id, run-id, node, kind, generation; RFC 0006 STREAM-14) that teardown compares verbatim. For streams that predate the labels, the stream’s own data subject is the authority: subjects are dot-delimited and sanitize never lets a dot through, so vf.f.r.x-n and vf.f.r-x.n are distinct strings — teardown reads the run out of the subject token, never out of a name prefix. Provisioning also reads every stream and consumer back after creating it and reports the fields the broker did not honour, because a REALTIME stream the broker silently kept at INTEREST retention is a flow with the wrong semantics, not a provisioned one.

videoflow.messaging.topology.DEFAULT_ITEM_CREDIT = 1

Inputs one replica holds in processing at a time.

videoflow.messaging.topology.DEFAULT_MAX_ACK_PENDING = 8

The broker-side max_ack_pending provisioning used before RFC 0006 (STREAM-15 replaced it with consumer_credit); kept as the reviewed value the conformance negative controls reproduce.

videoflow.messaging.topology.DEFAULT_PREFETCH = 4

Deliveries a worker parks locally per subscription beyond the one it is processing, so the next input is already in hand when the current one settles.

videoflow.messaging.topology.DEFAULT_READ_BACK_TIMEOUT_SECONDS = 30.0

How long read_back_streams gives the broker, connect included, before every stream it was asked about is reported Unknown('timeout').

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.DLQ_RETENTION_SECONDS = 604800

the forensic and replay horizon a dead letter’s payload obligation must cover too (BLOB-14 step 3).

Type:

How long dead letters are retained (STREAM-8)

videoflow.messaging.topology.LEGACY_BIND_CREDIT = 6

The max_ack_pending a worker bound its durable with before RFC 0006 (its local queue depth plus two); kept for the same reason.

class videoflow.messaging.topology.VerifiedConsumer(requested: nats.js.api.ConsumerConfig, effective: nats.js.api.ConsumerConfig | None, mismatches: tuple[str, ...] = ())

Bases: object

VerifiedStream’s counterpart for a durable consumer.

effective: nats.js.api.ConsumerConfig | None

mismatches: tuple[str, ...] = ()

requested: nats.js.api.ConsumerConfig

class videoflow.messaging.topology.VerifiedStream(requested: nats.js.api.StreamConfig, effective: nats.js.api.StreamConfig | None, mismatches: tuple[str, ...] = ())

Bases: object

What the broker holds after _ensure_stream: the config that was asked for, the config read back (None when the read-back itself failed), and each requested field the two differ on, rendered 'field: requested X, effective Y'. Empty mismatches means the broker gave the flow exactly what it asked for.

effective: nats.js.api.StreamConfig | None

mismatches: tuple[str, ...] = ()

requested: nats.js.api.StreamConfig

videoflow.messaging.topology.connect_options_for(timeout: float, fail_fast: bool, error_cb: Callable[[BaseException], Awaitable[None]]) → dict[str, Any]

nats.connect keyword arguments for a short-lived observation connection.

fail_fast (the operator’s machine) disables reconnects; note that nats-py 2.15.0 still cycles the server pool on a refused connect (nats/aio/client.py _select_next_server loops until a server answers, and max_reconnect_attempts = 0 never discards one), so the caller bounds the connect with asyncio.wait_for and reads the refusal out of error_cb — which is also where a server-side permissions violation on a JetStream API subject arrives (client.py _process_err), while the request itself merely times out. Without fail_fast (an in-cluster entrypoint whose broker may still be starting) the client’s own retry schedule applies: 60 attempts two seconds apart.

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, generation: str | None = None) → 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). Under RFC 0006 the durable is labelled with its consuming node as owner.

videoflow.messaging.topology.consumer_credit(nb_tasks: int, partitioned: bool, item_credit: int = 1, prefetch: int = 4) → int

STREAM-15: the broker-side max_ack_pending a subscription needs so every replica can hold item_credit inputs in processing plus prefetch parked, without one replica’s un-acked work starving another’s — a shared (competing) durable multiplies by the replica count, a per-replica (partitioned) durable does not. Provisioning and the worker call this so the durable created and the durable bound agree (MSG-017).

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

videoflow.messaging.topology.credit_admits(effective_credit: int, partitioned: bool, item_credit: int = 1, prefetch: int = 4) → int

The inverse of consumer_credit: how many replicas an effective max_ack_pending (read back from the broker) lets hold item_credit inputs in processing plus prefetch parked at the same time. A plan that admits more replicas than this is incompatible with the durable as it stands — the surplus replicas start but never hold an input (RUN-024): the credit must be re-derived and updated, not the replicas added. A per-replica (partitioned) durable serves one replica whatever its credit.

async videoflow.messaging.topology.delete_run_streams(nc: nats.aio.client.Client, flow_id: str, run_id: str, node_names: Iterable[str] | None = None, generation: str | None = None) → CleanupObservation

Teardown: delete the streams this run owns — by exact ownership, never by name prefix — and report truthfully what happened.

Ownership, per stream (_owned_stream): a stream carrying RFC 0006 owner labels is this run’s iff the labels equal (flow_id, run_id) verbatim (and generation, when given); an unlabelled stream is this run’s iff its data subject names this run token for token. Neither can mistake run r-x for run r the way startswith('vf-{flow}-{run}-') did, and neither needs the graph: node_names ([spec.name for spec in specs]) only narrows the deletion to those nodes’ streams, for a caller tearing down its own provisioning.

What is not attributed — and is therefore left standing, and not reported, since nothing on it says whose it is: an unlabelled stream provisioned with custom subjects by something other than this module. Nothing in-tree does that.

The flow’s dead-letter stream is deliberately never one of them. Teardown runs in a finally on success, failure, stall and Ctrl-C alike, so deleting the DLQ here destroyed exactly the evidence an operator wants after a failed run. It ages out on its own retention instead, and videoflow dlq purge removes it on purpose. (It carries flow-level labels with no run id, and its subject vf.{flow}._dlq.> is not a data subject, so neither rule can match it.)

• Returns:

  • CleanupObservation: removed are confirmed gone (a stream that was already gone counts), remaining are owned streams whose delete failed, and complete is False whenever anything remains or the inventory could not be read — a listing failure deletes nothing and is not “nothing to delete”. Re-running is safe, and is the remedy.

videoflow.messaging.topology.dlq_stream_config(flow_id: str, replicas: int = 1) → nats.js.api.StreamConfig

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

The dead-letter stream, scoped to the flow rather than the run.

Everything else about a run is disposable and is deleted with it; dead letters are the opposite — they are the forensic record of what went wrong, and they are most wanted precisely after a run that failed and was torn down. Run scoping meant delete_run_streams destroyed the evidence on the way out and the stream’s week-long retention never applied to anybody. The run id lives in the subject instead, so entries stay attributable and filterable.

videoflow.messaging.topology.dlq_subject_filter(flow_id: str, run_id: str | None = None, node_name: str | None = None) → str

A subject wildcard selecting dead letters for inspection: the whole flow, one run of it, or one node of one run. Used by videoflow dlq.

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

vf.{flow}._dlq.{run}.{node} — filterable by run, by node, or by both.

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, generation: str | None = None) → 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, generation: str | None = None) → 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.join_item_credit(parents: Sequence[str], join_policy: dict | None, flow_type: str) → int

The per-replica processing credit a node needs: one input group for a single-parent node; for a join, its policy’s working set — every half of every pending group plus the group being assembled (JoinPolicy.working_set) — because a credit smaller than that lets an adversarial parent ordering fill the durables with halves that can never complete (RUN-005). Provisioning and the bind derive it identically.

videoflow.messaging.topology.max_deliver_for(flow_type: str, max_retries: int = 3, delivery: dict | None = None, ledger_budget: bool = False) → int

Broker-side delivery cap for one node’s durables.

Derived from the node’s effective DeliveryPolicy rather than from the flow type alone, because delivery is overridable per node: an at-least-once sink in a REALTIME flow needs a cap above 1 or its retries would be silently impossible, and a best-effort node in a BATCH flow should not be retried at all. Provisioning and the messenger both call this so the durable they create and the durable they bind agree.

• Arguments:

  • flow_type: supplies the preset.

  • max_retries: deployment-level retry count (VF_MAX_RETRIES).

  • delivery: the node’s own override, as a dict (NodeSpec.delivery).

  • ledger_budget: the retry budget is enforced by the runtime ledger’s attempt counts (RFC 0006 STREAM-15, decision D11: only with a durable, shared runtime store), so an at-least-once durable is provisioned with max_deliver = -1 — the broker never strands a message; a worker-fatal redelivery never counts. Best-effort durables keep their cap of 1 either way.

videoflow.messaging.topology.observation_failure(error: BaseException, reported: Sequence[str] = ()) → tuple[str, str]

(reason, detail) for a failed observation of the broker, in the Unknown vocabulary: auth (credentials refused, or a permissions violation the error callback saw while the request timed out), unreachable (no server answered — a refused connect shows up as a timeout with a ConnectionRefusedError in reported), timeout (the server is there and slow), malformed (an answer the client could not interpret).

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

videoflow.messaging.topology.profile_mismatches(profile: str, effective: nats.js.api.StreamConfig) → tuple[str, ...]

The ways a stream’s effective configuration contradicts a messaging profile — the semantic half of a read-back, beside the field-level _mismatches. A field can match the request and still not carry the profile (the request asked for the wrong shape), and a stream nobody here provisioned can carry the profile with fields no request ever set (a byte limit instead of a message limit), so the two checks are separate. Each finding reads '<field>: <effective value> <what that means>'; empty means the stream carries the profile.

reliable_work needs retention that keeps a message until it is acknowledged (INTEREST or WORK_QUEUE) and DiscardPolicy.NEW at the limit — OLD on a full stream evicts an unacknowledged message, which is exactly the loss the profile forbids. live_latest needs LIMITS retention, DiscardPolicy.OLD (the freshest message wins) and a bound on the backlog (max_msgs, max_bytes or max_age), or “latest” is whatever fits on the disk. durable_control and replay_archive are not stream-level promises on JetStream (the planner rejects them before a stream exists), so they have no findings here.

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 | None = None, generation: str | None = None, replicas: int = 1, ledger_budget: bool = False) → 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. Every resource is read back after creation (_ensure_stream/_ensure_consumer), so a broker that did not honour a requested field is reported rather than trusted.

• Arguments:

  • nc: a connected nats client.

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

  • max_ack_pending: broker-side credit per durable. None derives it from the consuming node’s replica count and join working set (consumer_credit, STREAM-15).

  • generation: provisioning generation recorded in the owner labels (RFC 0006), so delete_run_streams can be asked to remove only what this provisioning created. None leaves that label out.

  • replicas: stream copies to request (a replicated broker profile’s jetstream_replicas); read back and reported like every other field.

  • ledger_budget: provision at-least-once durables with max_deliver = -1 because the workers’ runtime ledger enforces the retry budget (max_deliver_for); the workers derive the same from the same store.

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.

async videoflow.messaging.topology.quiet_error_cb(_e: BaseException) → None

Silences the client’s per-attempt error logging; the read-back reports the failure itself.

videoflow.messaging.topology.read_back_streams(nats_url: str, flow_id: str, run_id: str, node_names: Sequence[str], flow_type: str, timeout: float = 30.0, replicas: int = 1, generation: str | None = None, fail_fast: bool = True) → dict[str, VerifiedStream | Unknown]

read_back_streams_async over a short-lived connection of its own — for a worker checking its channels before it opens, or the provision entrypoint checking what it just created — so nothing reaches into a messenger’s live connection. A connect that fails within timeout reports every node as the same Unknown (observation_failure); nothing is ever inferred.

• Arguments:

  • node_names: the nodes whose output streams to read.

  • flow_type / replicas / generation: what provisioning requested, so VerifiedStream.mismatches is meaningful.

  • timeout: overall bound, connect included.

  • fail_fast: no reconnects (the operator’s machine, a worker whose broker is already up); False lets the client retry a broker that is still starting, as the in-cluster provision Job does.

async videoflow.messaging.topology.read_back_streams_async(js: nats.js.JetStreamContext, flow_id: str, run_id: str, node_names: Sequence[str], flow_type: str, replicas: int = 1, generation: str | None = None, reported: Sequence[str] = ()) → dict[str, VerifiedStream | Unknown]

Read the streams of the named nodes back from a JetStream context, keyed by node. Each entry is a VerifiedStream whose requested is what provisioning asks for under this flow type (stream_config_for) and whose mismatches are the requested fields the broker does not honour, or an Unknown saying why nothing can be said: missing (the stream does not exist — a definite observation, kept apart from the unobservable ones), or timeout / auth / unreachable / api / malformed (observation_failure).

stream_info returns the applied configuration (nats-py 2.15.0, nats/js/manager.py stream_info -> api.StreamInfo.config), and a stream that does not exist is nats.js.errors.NotFoundError (err_code 10059).

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, generation: str | None = None, replicas: int = 1) → nats.js.api.StreamConfig

• Arguments:

  • generation: provisioning generation recorded in the owner labels (RFC 0006); None leaves that label out.

  • replicas: copies the stream keeps (a replicated broker profile’s jetstream_replicas); 1 leaves the field to the server default, so a single-server request is exactly what it always was.

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

Prefix shared by every stream of one run. Diagnostic only: it is also the prefix of every stream of run {run}-x, so delete_run_streams never selects on it — ownership comes from owner metadata or from the subject tokens (subject_owner). Enumerated by backends.identity.derived_names.

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

videoflow.messaging.topology.subject_owner(subject: str) → tuple[str, str, str] | None

The (flow, run, node) tokens a data or EOS subject encodes — sanitized, exactly as subject_for/eos_subject_for wrote them — or None for any other subject (control, DLQ, wildcards, foreign).

This is the one place a run boundary is exact without metadata: subject tokens are dot-delimited and sanitize never lets a dot through, so vf.f.r.x-n (run r, node x-n) and vf.f.r-x.n (run r-x, node n) are different strings although both hyphen-join to the stream name vf-f-r-x-n. delete_run_streams attributes streams that predate owner labels with it.

videoflow.messaging.topology.verify_channel_profiles(read_back: Mapping[str, VerifiedStream | Unknown], requests: Sequence[ProfileRequest], *, unknown_is_fatal: bool, where: str, config_mismatches: bool = False) → None

Bind explicit profile requests to what the broker actually holds. For every request whose channel was read back: a stream whose effective configuration contradicts the profile (profile_mismatches) is a definite rejection; a stream that does not exist is a broker problem (provisioning has not run, or ran against another run id); a stream that could not be observed is fatal only when unknown_is_fatal — an explicit request must not pass on an unread guarantee — and a logged warning otherwise. Requests for channels the caller did not read back are not this caller’s to judge.

• Arguments:

  • config_mismatches: also count the field-level VerifiedStream.mismatches (a num_replicas below the requested count, a clamped limit). The provision entrypoint knows the full request and passes True; a worker, which does not know the replica count asked for, judges the profile only.

  • where: provision / worker <node>, for the message.

• Raises:

  • IncompatibleProfile: a stream definitely does not carry its requested profile.

  • BrokerUnavailable: a requested channel’s stream does not exist.

  • UnobservableState: a requested channel could not be read back and unknown_is_fatal.