videoflow.wire package

Submodules

videoflow.wire.redis_payload_store module

The Redis PayloadStore: named, generation-fenced obligations instead of a decrement-only counter, on the keys RFC 0006 BLOB-13 names, with every multi-key step an optimistic WATCH/MULTI/EXEC transaction and no Lua.

Why this module exists

RedisBlobStore (RFC 0002, serialization.py) reclaims an offloaded payload with one counter: EXISTS then DECR then UNLINK at zero. Reviewed against the payload conformance cases that shape has three holes. Two deliveries of one message to the same reader decrement twice (PAY-005). A counter that expires between the EXISTS and the DECR is re-created at -1 and the blob another reader still needs is deleted (PAY-008). A worker that dies after its ack but before its release leaks the blob until its TTL with nothing able to tell that the reader finished (PAY-006). This store replaces the counter with obligations: a SET of reader ids, released by id (idempotent), fenced by the generation minted at put (a stale release touches nothing), reconcilable from a ledger.

Keys (BLOB-13)

vf-blob-<hex>	string	the bytes; the wire BlobRef.ref, unchanged
vf-blobmeta-{vf-blob-<hex>}	HASH	size, digest, generation, content_id, created_at
vf-blobobl-{vf-blob-<hex>}	SET	obligation ids; EXPIRE = latest deadline

The companion keys are hash-tagged onto the blob key: on Redis Cluster a key’s slot is computed from the text between its first { and }, so all three hash to the slot of vf-blob-<hex> and one UNLINK / one transaction can cover them (decision D10). The blob key itself keeps its RFC 0002 spelling, so nothing on the wire changes. RFC 0002’s vf-blobrc-<hex> counter is not hash-tagged (it predates this layout); the fallback below keeps treating it as a single-key operation for that reason.

Lifecycle (BLOB-14)

1. Put writes the blob, then the metadata, then the obligation set. An interrupted put therefore degrades to a TTL-only blob — the safe direction (BLOB-5): nothing can ever be reclaimed early because a record is missing.

2. Release is WATCH obl meta → read the set and the stored generation → MULTI SREM obl id [+ UNLINK of all three keys when the set would become empty and the generation matches] → EXEC. A nil EXEC (something touched the keys since the WATCH) retries, bounded by TRANSACTION_RETRIES with a jittered pause between attempts — eight replicas of a partitioned child release one object within milliseconds of each other, and a bound below the reader count strands the last contenders with an unknown receipt the messenger does not retry (the object then leaks until its TTL). A release naming an older generation is stale and touches nothing. A missing obligation set is never created by a release: the reader applies RFC 0002 counter semantics if vf-blobrc-<hex> exists (BLOB-6) and otherwise leaves the blob to its TTL.

3. Acquire (dlq/<flow>, archive/<flow>, replay readers) adds the id and extends every key’s life to the deadline — never shortens it — so a dead letter’s bytes outlive the run (PAY-014).

4. Reconcile scans the store (SCAN, never KEYS) and reclaims objects whose obligation set is empty, objects that never received their set (a put interrupted after the metadata, older than orphan_grace_seconds, with no RFC 0002 counter) and objects whose only obligations are intent/* ids the ledger no longer requires. Anything it could not read lands in unknown.

5. TTL backstop (BLOB-7): every key carries the contract’s TTL, extended to the horizon when obligations pin the object.

Typed reads (BLOB-15)

read never returns None and never raises for a store that is merely unreachable: a transport failure is TransientFailure (retry — never “malformed bytes”), a nil is Missing, a digest mismatch against the metadata is Corrupt. A key with no metadata (an RFC 0002 publisher, or metadata that expired) reads unverified, and says so.

redis-py facts this module relies on (redis 8.0.1, .venv/.../redis/)

• client.py:1763-1766 — a Pipeline that is watching runs each command immediately and returns its real reply until multi() is called; after multi() commands are queued for execute().

• client.py:1879-1921 — execute() sends MULTI … EXEC as one packet; a nil EXEC reply (a watched key changed) raises WatchError('Watched variable changed.'). client.py:2039-2100 — execute() always reset()``s in ``finally, which returns the connection to the pool.

• client.py:2003-2037 and 1768-1805 — a ConnectionError/TimeoutError while watching is re-raised as a WatchError whose __context__ is the transport error. That is how a lost response to EXEC reaches us, and why the release code inspects __context__ to report unknown instead of retrying blindly.

• client.py:1703-1707 — with client.pipeline() as pipe calls reset() on exit (an UNWATCH is sent if still watching). client.py:2112-2120 — watch() refuses to run after multi(); unwatch() only sends when watching.

• exceptions.py:23-56 — BusyLoadingError and AuthenticationError are ConnectionError subclasses; TimeoutError is separate; WatchError (:69), ResponseError (:57), NoPermissionError(ResponseError) (:97), OutOfMemoryError(ResponseError) (:77). None of them derive from the builtins of the same name.

• connection.py:884-894 — the default retry policy is Retry(NoBackoff(), 0): a transport error surfaces once, redis-py does not retry it for us.

• _parsers/helpers.py:35 — INFO is parsed into a dict with ints (cluster_enabled → 0/1); :911 — CONFIG GET values are str even with decode_responses = False; :933 — SCAN returns (int cursor, [raw keys]). Every other reply here is raw bytes.

• commands/core.py:2997-3033 — expire(name, secs, gt = True) sends EXPIRE key secs GT (Redis ≥ 7.0): set only when longer than the current expiry, and a key without a TTL counts as infinite, so it is never shortened.

• commands/core.py:11815 — client.cluster('KEYSLOT', key); a standalone server answers with a ResponseError (cluster support disabled), which is why the probe first reads INFO cluster.

Redis server facts: an empty SET does not exist (SREM of the last member deletes the key, so “set absent” and “set empty” are one state); a watched key that expires before EXEC aborts the transaction (since 6.0.9; one already expired at WATCH time does not, since 7.0); SCAN may return a key twice and never blocks the server.

videoflow.wire.redis_payload_store.COUNTER_KEY_PREFIX = 'vf-blobrc-'

the fallback contract when no obligation set exists.

Type:

RFC 0002’s reclamation counter (BLOB-5)

videoflow.wire.redis_payload_store.DEFAULT_ORPHAN_GRACE_SECONDS = 60.0

How long a blob with metadata but no obligation set is presumed to be a put still in progress.

videoflow.wire.redis_payload_store.INTENT_PREFIX = 'intent/'

the only obligations reconcile may cancel on its own.

Type:

Publisher intents (intent/<publication_id>)

videoflow.wire.redis_payload_store.MAX_OBJECT_BYTES = 536870912

the largest string value a stock server accepts.

Type:

Redis’ proto-max-bulk-len default

class videoflow.wire.redis_payload_store.RedisPayloadStore(url: str | None = None, client: Any = None, orphan_grace_seconds: float = 60.0, clock: Callable[[], float]=<built-in function time>)

Bases: PayloadStore

• Arguments:

  • url: the Redis URL; defaults to VIDEOFLOW_BLOB_REDIS_URL or the local dev server, exactly as RedisBlobStore.

  • client: an already-built client (a test’s fake). When given, url is ignored.

  • orphan_grace_seconds: how long a blob that has metadata but no obligation set is left alone by reconcile — a put interrupted before its set was written must not be reclaimed while its publisher may still be about to reference it.

  • clock: epoch-seconds source. Obligation deadlines and created_at are in this domain, because they are compared across processes.

Thread-safe: every transaction takes its own pooled connection.

acquire_obligation(ref: ImmutablePayloadRef, obligation_id: str, deadline: float) → DurableReceipt

SADD the id and extend every key’s life to deadline (epoch seconds), never shortening it, in one transaction fenced on the stored generation.

• Raises:

  • LookupError: the object is not stored (expired, evicted, reclaimed or never written), or is a different generation now — as the reference store.

  • TransientFailure (core): the server was unreachable, or the record kept changing under every attempt.

capabilities() → PayloadCapabilities

property client: Any

inventory() → Known[tuple[ImmutablePayloadRef, ...]] | Unknown

put(data: bytes, content_id: str, contract: RetentionContract) → ImmutablePayloadRef

Blob, then metadata, then the obligation set (BLOB-14 step 1). Every key gets the contract’s TTL, extended to the horizon when obligations pin the object (BLOB-14 step 5); a TTL of 0 or less means no expiry. payload.write.after fires once the bytes and their metadata are on the server and before the set is written; obligation.acquire.after once the set is — so a fault schedule can stop a put at exactly the point where it leaves a counterless, TTL-only object (PAY-007).

• Raises:

  • ResourceUnavailable: the server refused the write for memory (OOM under noeviction) — backpressure, never silent eviction.

  • TransientFailure (core): the server was unreachable or slow. Whatever was written before the failure is TTL-only and harmless.

read(ref: ImmutablePayloadRef, reader: str | None = None) → PayloadBytes | TransientFailure | Missing | Corrupt

GET the bytes and verify them against the ref’s digest, or the metadata’s when the ref carries none (BLOB-15). reader only labels diagnostics.

reconcile(ledger: ObligationLedger, operation_id: str) → ReclamationObservation

BLOB-14 step 4. For a key the ledger lists, its tuple is the whole truth: every other obligation is cancelled. For a key it does not list only intent/* obligations are cancelled — reader obligations belong to readers this ledger may not see, and leaking until TTL is the safe error. An object is reclaimed when nothing remains; one with no set at all is reclaimed only past orphan_grace_seconds and only if it has metadata to date it and no RFC 0002 counter. A scan that failed reports its pattern in unknown; a key that could not be read or kept changing is reported there by name.

ref_for_key(key: str) → ImmutablePayloadRef

The ref for a wire key, completed from vf-blobmeta-{key}; the ABC default (unverifiable, unfenced) when there is no metadata.

• Raises:

  • TransientFailure (core): the metadata could not be read; the caller retries.

release_obligation(ref: ImmutablePayloadRef, obligation_id: str, completion_receipt: str) → ReleaseReceipt

The BLOB-14 step 2 transaction. Idempotent by (obligation_id, generation): releasing an id that is not in the set changes nothing and reports the truthful remaining count; a release after the object is gone finds no set and applies the RFC 0002 fallback, which creates nothing.

• Returns: a receipt whose stale means the generation did not match and nothing was touched; unknown means the transaction was sent but its reply was lost, or the keys changed under every attempt — retry, it is safe; remaining is None on the fallback path (counter semantics keep no per-reader record).

renew_obligation(ref: ImmutablePayloadRef, obligation_id: str, deadline: float) → DurableReceipt

videoflow.wire.redis_payload_store.SCAN_COUNT = 200

Keys per SCAN round trip.

videoflow.wire.redis_payload_store.STORE_ID = 'redis'

The store identity written into every ref (ImmutablePayloadRef.store).

videoflow.wire.redis_payload_store.TRANSACTION_RETRIES = 32

Bound on optimistic-transaction retries (BLOB-14 step 2); past it the outcome is unknown. Every contender on one object costs the others a round — eight replicas of a partitioned child release the same key within milliseconds of each other — so the bound must exceed the largest reader set a payload carries, and each retry yields briefly (_RETRY_BACKOFF_SECONDS, jittered) so the contenders do not re-collide in lockstep.

videoflow.wire.redis_payload_store.blob_key_of(companion: str) → str | None

The blob key a companion key is hash-tagged onto, or None when companion is not one.

videoflow.wire.redis_payload_store.counter_key(key: str) → str

vf-blobrc-<hex> — RFC 0002’s counter for the blob (BLOB-5), not hash-tagged.

videoflow.wire.redis_payload_store.metadata_key(key: str) → str

vf-blobmeta-{vf-blob-<hex>} — hash-tagged onto the blob key (BLOB-13).

videoflow.wire.redis_payload_store.obligation_key(key: str) → str

vf-blobobl-{vf-blob-<hex>} — hash-tagged onto the blob key (BLOB-13).

videoflow.wire.redis_payload_store.redis_capabilities_observed(client: Any, max_object_bytes: int = 536870912) → PayloadCapabilities

What this Redis, as configured, can guarantee — read back live, never assumed.

• Arguments:

  • client: a redis.Redis (or a fake modelling config_get, info and cluster).

  • max_object_bytes: the largest object the server accepts (proto-max-bulk-len).

• Returns: a PayloadCapabilities whose durable is Known(True) only when persistence is on (appendonly yes or a save schedule) and the eviction policy is noeviction; evictable is Known(policy !=         'noeviction'); both are Unknown('auth', …) when CONFIG GET is denied (ACL NOPERM, or the command renamed/disabled on a managed offering). persistent_storage stays Unknown('unread'): what backs the data directory (a volume that outlives the pod, or not) is not visible over the wire. atomic_multikey is Known(True) on a standalone server, and on a cluster only when CLUSTER KEYSLOT agrees for the three keys of a probe blob; any failure to observe it is Unknown — the planner treats that as “not offered”.

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.ENVELOPE_OVERHEAD_BYTES = 4096

Bytes the v4 envelope adds around a payload at most (headers, ids, an event timestamp, small metadata): the margin safe_inline_threshold keeps between the inline threshold and the broker’s max_payload.

videoflow.wire.serialization.MAX_ERROR_TEXT_BYTES = 2048

Longest error message/remedy carried in an ABORT envelope. Bounded because an abort marker travels on the terminator path, where an unbounded string from a node’s exception text could otherwise outgrow a broker message.

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 a clean end-of-stream marker with no payload; abort is an abnormal one, carrying the error that killed the emitting node. Both terminators ride the same _eos subject.

videoflow.wire.serialization.MSG_TYPE_TERMINATORS = ('eos', 'abort')

Kinds that mean “no more data from this producer”. Anything that terminates a stream belongs here, so a reader that does not know about a newer terminator still stops rather than waiting forever.

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, resolve_blobs: bool = True) → 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 for either terminator, so a reader that predates MSG_TYPE_ABORT still stops), is_abort (True only for MSG_TYPE_ABORT), error (the failure record an abort carries, else None), span_id, parent_span_id, replica_id, metadata, message (the fully decoded payload — None for a terminator), 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.

• resolve_blobs: False leaves an offloaded payload unfetched: message is None, hydrated is False and blob_ref/blob_inner_type say what to fetch; hydrate_message completes it. Inline payloads are always decoded.

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, error: dict | None = None, inline_threshold: 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, MSG_TYPE_EOS or MSG_TYPE_ABORT. Neither terminator carries a payload.

  • error: the failure record an MSG_TYPE_ABORT marker carries (see videoflow.core.errors.error_to_dict). Ignored for other types.

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

  • inline_threshold: encoded payload size above which the payload is offloaded; None ⇒ MAX_INLINE_PAYLOAD_BYTES. A messenger passes the value it negotiated against the broker’s max_payload (safe_inline_threshold), so an unsafe default cannot bypass the limit.

videoflow.wire.serialization.error_from_proto(proto: Error) → dict

Inverse of error_to_proto; empty optional fields are omitted rather than emitted as ‘’.

videoflow.wire.serialization.error_to_proto(error: dict) → Error

Builds the wire form of an error record (as produced by videoflow.core.errors.error_to_dict). Unknown dispositions degrade to UNSPECIFIED rather than raising: an abort marker’s job is to end a flow cleanly, and it must not itself fail to encode.

videoflow.wire.serialization.hydrate_message(decoded: dict, blob_store: BlobStore) → Any

Completes an envelope decoded with resolve_blobs = False: fetches the offloaded bytes and decodes the inner payload. Returns the message; the caller stores it (decoded['message']) and marks hydrated. A store failure propagates as the store raised it (KeyError for a missing object from RedisBlobStore; a payload store’s typed read outcomes are the messenger’s to classify — transient failures retry, missing or corrupt objects are dead-lettered, BLOB-15).

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.peek_envelope(buf: bytes) → dict

The routing view of an envelope — every key decode_envelope returns except that message is never decoded (None, hydrated False for data): the parse a receiver can afford on its transport thread to decide ownership and replay scope (PART-4, DELIV-16) before a delivery is parked, without paying for the payload it may never process. blob_ref is still reported so an ack-and-skip can release the reader’s share.

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.

videoflow.wire.serialization.safe_inline_threshold(max_payload_bytes: int, inline_threshold: int = 524288) → int

The inline threshold a broker with max_payload_bytes can carry: an envelope whose payload is just under the threshold must still fit with its framing. Returns inline_threshold when it is safe, else the largest safe value; ConfigError is the caller’s to raise when the configured threshold cannot be honoured (PAY-001).

videoflow.wire.serialization.serialized_payload_size(payload: Any) → int

The byte length a payload occupies on the wire once encoded (PAY-020): a decoded ndarray costs its nbytes plus the tensor framing, never the compressed source it was read from. The same encoder the publisher uses, so the estimate equals the measurement.