videoflow.wire package

Submodules

videoflow.wire.serialization module

Wire format used to move messages between nodes over the message broker (videoflow.messaging.nats_messenger.NATSMessenger). Not used by anything running purely in a single local process — it’s the boundary format for bytes that cross a network/process boundary.

The wire is a single, language-neutral protobuf envelope (videoflow.v1.Envelope, envelope version 4). Its payload is a typed protobuf message:

• Tensor for arrays, including video frames;

• Value for structured scalars/maps/lists — and a Value may nest a Tensor (tensor_value), so a mixed container like a (frame_index, frame) tuple has a neutral encoding (spec/PROTOCOL.md WIRE-15);

• any vendor proto by its fully-qualified name.

Arbitrary Python objects are never put on the wire: there is no code-executing fallback codec, because deserializing attacker-controlled bytes that way is remote code execution (see spec/rfcs/0001). A payload type with no built-in encoding registers one via register_payload_encoder; an unknown payload_type on decode is handed back as opaque RawPayload bytes and never deserialized.

decode_envelope reads the envelope; a legacy msgpack (v2/v3) envelope is refused with a clear error rather than decoded. Runs are version-homogeneous — streams are run-scoped — so a single run never mixes versions.

class videoflow.wire.serialization.BlobStore

Bases: object

Interface for the external blob store used for payloads over MAX_INLINE_PAYLOAD_BYTES. Not tied to any particular broker — Redis is a convenient default (large string values, simple TTL-based expiry) even when NATS is the primary messaging broker.

Reclamation (RFC 0002): put_with_readers/release default to TTL-only behaviour so a subclass that only implements put/get keeps working; a store that can refcount overrides both.

get(ref: str) → bytes

Resolves a reference previously returned by put() back into bytes.

put(data: bytes, ttl_seconds: int = 3600) → str

Stores data and returns an opaque reference string that get() can resolve later.

put_with_readers(data: bytes, readers: int, ttl_seconds: int = 3600) → str

Stores data expecting exactly readers downstream reads, enabling the store to reclaim the blob once every reader has released it (BLOB-5).

Default implementation ignores readers and delegates to put() — a store with no reclamation support degrades to plain TTL expiry.

release(ref: str) → None

One downstream reader is finished with ref (its message was acked, BLOB-6). Default no-op.

videoflow.wire.serialization.COMPATIBLE_ENVELOPE_VERSIONS = (4,)

Versions this build can decode. The legacy msgpack wire (v2/v3) has been removed; such envelopes are refused on decode.

videoflow.wire.serialization.DEFAULT_BLOB_TTL_SECONDS = 3600

Blob lifetime when nothing plumbs an explicit TTL. Serialization is flow-type agnostic, so the flow-aware defaults (3600s realtime / 86400s batch, BLOB-7) live with the messenger that knows the flow type; this is only the fallback.

videoflow.wire.serialization.DEFAULT_ENVELOPE_VERSION = 4

a language-neutral videoflow.v1.Envelope (protobuf). Overridable per run via VF_ENVELOPE_VERSION only to a version this build speaks.

Type:

The sole envelope version

videoflow.wire.serialization.EMITTABLE_ENVELOPE_VERSIONS = (4,)

Versions this build can emit.

videoflow.wire.serialization.MAX_INLINE_PAYLOAD_BYTES = 524288

Payloads whose serialized size (in bytes) exceeds this threshold are written to a BlobStore instead of being inlined in the broker message. Large uncompressed video frames (a 1080p RGB frame is ~6.2MB) would otherwise blow past a typical broker’s per-message size limit (NATS defaults to a 1MB max_payload).

videoflow.wire.serialization.MSG_TYPE_DATA = 'data'

Message kinds carried in the envelope type field. data is a normal payload; eos is an end-of-stream marker with no payload.

videoflow.wire.serialization.PAYLOAD_ENTRY_POINT_GROUP = 'videoflow.payload_types'

Entry-point group for third-party payload codecs. Registration normally happens as a side effect of importing the component module (the worker imports it via VF_NODE_CLASS anyway); the group covers host-side tools such as videoflow debug decode, which must understand a vendor payload without knowing which package defines it.

videoflow.wire.serialization.PAYLOAD_TENSOR = 'videoflow.v1.Tensor'

v4 protobuf payload-type identifiers (the payload_type field of an Envelope). Proto messages use their descriptor FQN; these three are the ones the codec special-cases. Any other FQN round-trips as an opaque RawPayload.

class videoflow.wire.serialization.RawPayload(payload_type: str, data: bytes)

Bases: object

An opaque, already-encoded payload: a payload_type FQN and its raw bytes. Produced when decoding a message whose payload_type is not registered (so a forwarding/storing node need not understand it), and accepted by the encoder so such a payload can be re-published losslessly (PROTOCOL.md WIRE-9).

data

payload_type

class videoflow.wire.serialization.RedisBlobStore(url: str | None = None)

Bases: BlobStore

Uses a Redis server purely as a large-value TTL cache, independent of whether Redis is used for messaging.

get(ref: str) → bytes

Resolves a reference previously returned by put() back into bytes.

put(data: bytes, ttl_seconds: int = 3600) → str

Stores data and returns an opaque reference string that get() can resolve later.

put_with_readers(data: bytes, readers: int, ttl_seconds: int = 3600) → str

Stores data expecting exactly readers downstream reads, enabling the store to reclaim the blob once every reader has released it (BLOB-5).

Default implementation ignores readers and delegates to put() — a store with no reclamation support degrades to plain TTL expiry.

release(ref: str) → None

One downstream reader is finished with ref (its message was acked, BLOB-6). Default no-op.

videoflow.wire.serialization.decode_envelope(buf: bytes, blob_store: BlobStore | None = None) → dict

Decodes wire bytes back into a dict with keys producer_name, flow_id, run_id, trace_id, seq, event_ts (None when absent), type, is_stop_signal (derived: True iff type == MSG_TYPE_EOS), span_id, parent_span_id, replica_id, metadata, message (the fully decoded payload — None for EOS), and blob_ref (the blob store reference the payload was resolved from, or None when the payload was inline — lets the caller release the blob after the message is acked, BLOB-6). Only the protobuf v4 envelope is supported; a legacy msgpack (v2/v3) envelope is refused.

videoflow.wire.serialization.derive_message_id(flow_id: str, run_id: str, producer_name: str, trace_id: str, seq: int, msg_type: str) → str

Deterministic, content-derived message id. Two publishes of the same logical message (e.g. a processor that crashed after publishing but is re-run and recomputes the same output for the same input group) produce the same id, so JetStream’s Nats-Msg-Id de-duplication drops the retry copy. It is therefore essential that the inputs here are stable across retries — in particular seq must be carried forward from the input group, not a local wall-clock or attempt counter.

videoflow.wire.serialization.encode_envelope(producer_name: str, flow_id: str, run_id: str, trace_id: str, seq: int, msg_type: str, metadata: dict | None, payload: Any, span_id: str = '', parent_span_id: str = '', replica_id: int = 0, event_ts: float | None = None, blob_store: BlobStore | None = None, version: int | None = None, blob_readers: int | None = None, blob_ttl_seconds: int | None = None) → bytes

Encodes a full wire message and returns the bytes to publish to a broker subject.

• Arguments:

  • msg_type: MSG_TYPE_DATA or MSG_TYPE_EOS. EOS carries no payload.

  • run_id: the per-run identifier that scopes this flow execution.

  • span_id / parent_span_id: hex ids for log/trace correlation (optional).

  • replica_id: index of the emitting replica (0 for single-task nodes); distinguishes EOS markers from different replicas of one node.

  • event_ts: event time of the message in epoch seconds — when the underlying real-world event was captured — minted by the producer and carried forward unchanged; time-aligned joins group on it.

  • version: envelope version to emit. The only supported version is 4 (protobuf); defaults to DEFAULT_ENVELOPE_VERSION.

  • blob_readers: how many downstream reads an offloaded payload will receive; enables refcounted blob reclamation (BLOB-5). None ⇒ TTL-only blobs.

  • blob_ttl_seconds: TTL for an offloaded payload (and its counter); None ⇒ DEFAULT_BLOB_TTL_SECONDS.

videoflow.wire.serialization.make_blob_store(url: str) → BlobStore

Builds the blob store for url, dispatching on its scheme.

• Arguments:

  • url: blob store URL, e.g. redis://localhost:6379/0.

• Returns: a ready BlobStore.

• Raises:

  • ValueError: the URL has no scheme, or no store is registered for it. The message names the known schemes and register_blob_store.

videoflow.wire.serialization.register_blob_store(scheme: str, factory: Callable[[str], BlobStore]) → None

Registers a BlobStore factory for a URL scheme. factory receives the full URL (not just the remainder) and returns a ready store.

Third-party packages may register either by calling this on import, or by declaring an entry point in the videoflow.blob_stores group — the latter is what lets a store be selected purely by configuration, with nothing in the flow importing the package that provides it.

• Arguments:

  • scheme: URL scheme without the separator, e.g. 's3'. Case-insensitive: normalized here to match how make_blob_store parses a URL, so register_blob_store('S3', ...) is reachable from s3://bucket/key.

  • factory: callable taking the blob URL and returning a BlobStore.

videoflow.wire.serialization.register_payload_encoder(python_type: type, encoder: Callable[[Any], Tuple[str, bytes]]) → None

Registers an encode rule mapping instances of python_type to a (payload_type, serialized_bytes) pair on the v4 wire — the encode-side complement of register_payload_type.

Rules are checked in registration order, and only after every built-in check (ndarray, RawPayload, protobuf message, and the JSON-like types that become Value). That ordering is deliberate and worth preserving: it lets a vendor type get a real wire type instead of a TypeError, while making it impossible for any registration — even one matching object — to change how a built-in payload encodes. Those mappings are fixed by spec/PROTOCOL.md §4.4 and proven by the golden vectors, so they are not an extension point. This is the only way to give a type with no built-in encoding a place on the wire — there is no code-executing fallback (see spec/rfcs/0001).

The corollary: a type that is already encodable (a dict subclass, say) is encoded as the built-in it resembles, and its registered rule never runs.

Pair this with register_payload_type (or a decoder that understands the payload_type string) so the receiving side can decode what you emit.

• Arguments:

  • python_type: the type to match with isinstance.

  • encoder: callable taking the payload and returning (payload_type, serialized_bytes).

videoflow.wire.serialization.register_payload_type(message_cls: type[Message]) → None

Register a protobuf message class so envelopes carrying its FQN decode to an instance of it (rather than an opaque RawPayload). The well-known videoflow types are pre-registered; vendors register their own payload messages.

videoflow.wire.serialization.registered_blob_store_schemes() → list[str]

The URL schemes a blob store is currently registered for, sorted.