videoflow.backends.memory package

Reference in-memory backends: the executable specification.

Each module here implements one contract faithfully enough that the conformance suite can run every model-level case against it — retention limits, leases and redelivery, dedup windows, obligation ledgers, compare-and-swap ownership, MIG geometry — under a fake clock and the fault barriers in videoflow.backends.faults. They are not test doubles that return canned answers: a real adapter must produce the same outcomes for the same schedule, which is what makes the same conformance runner usable for every provider.

Submodules

videoflow.backends.memory.allocation module

The in-memory AcceleratorAllocationBackend: a cluster of GPU nodes as data, with the concurrency protocol the real adapters must honour.

Modelled deliberately:

• Reads can fail. fail_reads('pods') makes the pod listing unavailable: the inventory then reports completeness='partial' with no occupancy, and a plan that needs occupancy is Infeasible('occupancy unknown') — never “idle”. fail_reads('nodes') makes inventory() itself Unknown.

• Ownership is a server-side compare-and-swap. Every node carries a resource_version; a write names the version it read, and loses when a concurrent writer bumped it. Two reservers paused between their read and their write (owner.read.after / owner.update.before barriers) race exactly as two kubectl clients would.

• Readiness is correlated with the operation. A node’s mig.config.state carries the generation of the write it answers; a stale success for an earlier geometry does not complete a new one.

• Foreign workloads and retained workloads hold devices. Geometry is never destroyed while any workload the model knows about still uses the node.

• Shared configuration is never deleted. The last owner out restores the pointer and leaves a tombstoned entry-less map, under CAS.

• Every mutation is audited, so a zero-mutation assertion is a list check.

class videoflow.backends.memory.allocation.MemoryAllocationBackend(nodes: Sequence[NodeFixture], clock: FakeClock | None = None, authority: str = 'device-plugin', mig_apply_seconds: float = 0.0, log: ObservationLog | None = None)

Bases: AcceleratorAllocationBackend

• Arguments:

  • nodes: the cluster.

  • authority: which real adapter this instance stands in for; decides the capabilities advertised and whether geometry can be changed.

  • mig_apply_seconds: fake-clock delay before a geometry write reaches state=success (0 = immediately on the next observe).

apply_pending_geometry() → None

Advance every geometry write whose fake-clock delay has elapsed (tests call this after clock.advance).

bind_workload(node: str, workload_id: str, units: int) → None

A workload (ours or foreign) now holds units devices on node.

bindings(claim_id: str, workload_id: str) → WorkloadBindings

capabilities(environment: Mapping[str, Any]) → AllocationCapabilities

fail_geometry(node_name: str) → None

Make the manager report state=failed for every geometry write on node_name (a permanent preparation failure).

fail_reads(kind: str, reason: str | None = 'timeout') → None

Make nodes / pods / config reads fail (None restores them).

inventory(scope: Mapping[str, Any]) → Known[InventorySnapshot] | Unknown

mark_workload_ready(claim_id: str) → None

mutations(kinds: Sequence[str] | None = None) → list[tuple[str, str, str, str]]

node(name: str) → NodeFixture

observe(claim_id: str) → Known[ClaimObservation] | Unknown

plan(requests: Sequence[WorkloadRequest], snapshot: InventorySnapshot) → FeasiblePlan | Infeasible

pointer() → str

reconcile(claim_id: str, desired: str, expected_generation: str) → ClaimObservation

release(claim_id: str, operation_id: str, expected_generation: str, keep_workloads: bool = False) → ReleaseObservation

reserve(plan: FeasiblePlan, operation_id: str, expected_generation: str | None) → ClaimObservation

shared_config() → tuple[dict[str, dict[str, int]], int, bool]

unbind_workload(node: str, workload_id: str) → None

class videoflow.backends.memory.allocation.NodeFixture(name: str, product: str, gpu_count: int, memory_gib: float, labels: dict[str, str]=<factory>, owner: str | None = None, owner_epoch: str | None = None, resource_version: int = 1, mig_config: str = 'all-disabled', mig_state: str = 'success', mig_state_generation: int = 0, mig_layout: dict[str, int]=<factory>, workloads: dict[str, int]=<factory>, restore_record: str | None = None)

Bases: object

One GPU node. labels follow GPU Feature Discovery’s vocabulary.

classification() → str

The same per-node rule the cluster reader applies to GPU Feature Discovery labels (deploy.cluster.classify_gfd_labels), so a fixture and a real node with the same labels are judged identically — plus geometry this backend applied itself, which a real node would advertise as slices.

gpu_count: int

labels: dict[str, str]

memory_gib: float

mig_config: str = 'all-disabled'

mig_layout: dict[str, int]

mig_state: str = 'success'

mig_state_generation: int = 0

name: str

owner: str | None = None

owner_epoch: str | None = None

product: str

resource_version: int = 1

restore_record: str | None = None

workloads: dict[str, int]

videoflow.backends.memory.allocation.pack_whole_devices(requests: Sequence[tuple[str, int]], free: Mapping[str, int], eligible: Mapping[str, AbstractSet[str]] | None = None) → dict[str, str] | None

Exhaustive per-host packing for small inventories: each request needs count whole devices on one host. Returns workload -> host, or None when no assignment exists — the independent oracle for “aggregate capacity hides per-node fragmentation” (three hosts with two free each cannot host three requests of two? they can; two hosts with three free each cannot host three requests of two). eligible narrows the hosts a workload may take (its hard constraints); a workload absent from it may take any host.

videoflow.backends.memory.clock module

A controllable clock for deterministic models: monotonic and wall time advance together, only when asked. Timers fire on advance.

class videoflow.backends.memory.clock.FakeClock(start: float = 1000.0, wall_start: float = 1700000000.0)

Bases: object

advance(seconds: float) → None

Move time forward, firing every timer due on the way in order.

at(seconds_from_now: float, callback: Callable[[], None]) → None

monotonic() → float

now() → float

time() → float

videoflow.backends.memory.messaging module

The in-memory MessagingBackend: a faithful model of a JetStream-like broker (and, in core_only mode, of Core NATS) under a fake clock.

What it models deliberately, because the conformance cases turn on it:

• Retention. limits channels keep at most max_msgs messages per stream — data, EOS and control share the slot unless per_subject_limits is set — and either evict the oldest (recording a drop for every subscription that had not settled it) or reject the publish. interest channels keep a message until every required subscription has settled it.

• Leases and redelivery. A delivery is leased for ack_wait_seconds; an expired lease redelivers with attempt + 1 until max_deliver (-1 = unlimited). Past the cap the message is unresolved: retained, undeliverable, and reported as such — never as “zero pending”.

• Credit. item_credit is the server-side cap on leased messages per subscription, shared by every replica competing on it; byte_credit bounds a single receive.

• Dedup. A publication_id seen within the window is accepted as a duplicate and stored once; beyond it a retry is a new message.

• Ambiguous acceptance. pause_acceptance holds publications so a caller’s deadline expires with the outcome unknown; resume then stores them (late acceptance) unless cancel_publication removed them first.

• Truthful observation. fail_observation makes observe_subscription return Unknown; nothing in this model turns a failed read into zero.

• Core-only mode: no retention, no leases, bounded client queues with slow-consumer drops, nothing for a subscription that was not attached at publish time.

• Subjects. A subscription sees only the messages of its own kind — a data subscription never receives a terminator and an eos one never receives data — exactly as a JetStream durable filters to one subject of the node’s stream. A dead letter (kind == 'dlq') is stored on the origin node’s channel for inspection but belongs to the flow’s DLQ retention, so it never occupies, evicts or fills the run channel’s slot.

• Archive. An archived envelope outlives close of its run’s channels (replay is a new execution against retained history, not the working queue).

class videoflow.backends.memory.messaging.MemoryMessagingBackend(clock: FakeClock | None = None, core_only: bool = False, latest_per_key: bool = False, durable_control: bool = False, archive: bool = False, archive_horizon_seconds: float = 3600.0, dedup_window_seconds: int = 120, replication_factor: int = 1, persistent: bool = False, max_payload_bytes: int = 8388608, mixed_retention: bool = False, client_queue_limit: int = 64, log: ObservationLog | None = None)

Bases: MessagingBackend

• Arguments:

  • clock: the fake clock every timer reads.

  • core_only: model Core NATS (no retention, no leases, client queues).

  • latest_per_key: advertise and implement per-key latest-value slots.

  • durable_control / archive: advertise the optional profiles (MSG-018, MSG-024).

  • client_queue_limit: core-only slow-consumer bound per subscription.

  • log: an observation log to record publish/deliver/settle/drop events into.

cancel_publication(channel_id: ChannelId, publication_id: str) → bool

Definitely cancel a held publication. False when it was not held (it may already be stored).

capabilities() → MessagingCapabilities

channel_ids() → list[ChannelId]

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

Remove exactly the owned channels; report what could not be confirmed removed.

commit_control(channel_id: ChannelId, key: str, expected_version: str | None, value: bytes) → bool

control_state(channel_id: ChannelId, key: str) → tuple[bytes | None, str | None]

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

fail_observation(subscription: SubscriptionId, reason: str | None = 'timeout') → None

Make observe_subscription return Unknown(reason) until called with None.

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

The highest stream sequence below which every visible message is settled for subscription.

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

The channel’s retained range (ChannelObservation); Unknown when the adapter cannot read it.

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

What the backend can say later about envelope.publication_id; Unresolvable when it keeps no ledger.

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

pause_acceptance(channel_id: ChannelId) → 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]

renew(token: DeliveryToken) → LeaseObservation

replay(channel_id: ChannelId, event_id: str) → Envelope | None

The archived envelope for event_id while within the archive horizon; None once expired.

resume_acceptance(channel_id: ChannelId) → list[Accepted | Rejected | PublicationUnknown | PublicationUnresolvable]

Store every held publication (late acceptance); returns their outcomes in order.

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

stored(channel_id: ChannelId) → list[Envelope]

subscription_ids(channel_id: ChannelId) → list[SubscriptionId]

The logical subscriptions bound on a channel — the inventory a teardown or a replacement worker reads.

videoflow.backends.memory.messaging.make_channel(flow_id: str, run_id: str, node: str, profile: str, retention: str, required: Sequence[SubscriptionId] = (), max_msgs: int = 10000, max_bytes: int | None = None, max_age_seconds: float | None = None, overflow: str = 'evict_oldest', dedup_window_seconds: int = 120, replicas: int = 1, persistence: bool = False, per_subject_limits: bool = False, owner_labels: Mapping[str, str] | None = None) → ChannelSpec

Convenience constructor used by tests and by the composition of the reference runtime.

videoflow.backends.memory.messaging.make_envelope(channel: ChannelId, publication_id: str, body: bytes = b'x', event_id: str | None = None, kind: str = 'data', partition_key: str | None = None, event_ts: float | None = None, source_epoch: str | None = None, source_offset: int | None = None, headers: Mapping[str, str] | None = None) → Envelope

videoflow.backends.memory.messaging.make_subscription(channel: ChannelId, consumer: str, competing: bool = True, partition: int | None = None, ack_wait_seconds: float = 30.0, max_deliver: int = 4, item_credit: int = 8, byte_credit: int = 67108864, kind: str = 'data') → SubscriptionSpec

videoflow.backends.memory.mig_geometry module

An independent MIG geometry oracle.

videoflow/deploy/mig.py plans layouts with a deliberately simplified card model (compute slices only, no placement positions). This table is the independent check the conformance cases evaluate a plan against: memory slices are counted separately from compute slices, and every profile carries the start positions the hardware allows, so a layout that fits by slice count but cannot be placed is rejected here — and a layout the planner rejects for the wrong reason is caught as well. Values are transcribed from the NVIDIA MIG user guide (“Supported MIG profiles” tables); the family key is a substring of the GFD nvidia.com/gpu.product label. Keep this file boring: no imports from deploy/, nothing shared with the planner it audits.

class videoflow.backends.memory.mig_geometry.Family(key: str, compute_slices: int, memory_slices: int, memory_gib: float, profiles: Mapping[str, videoflow.backends.memory.mig_geometry.Profile])

Bases: object

compute_slices: int

key: str

memory_gib: float

memory_slices: int

profiles: Mapping[str, Profile]

class videoflow.backends.memory.mig_geometry.Profile(name: str, compute_slices: int, memory_slices: int, memory_gib: float, max_instances: int, starts: tuple[int, ...])

Bases: object

compute_slices: int

max_instances: int

memory_gib: float

memory_slices: int

name: str

starts: tuple[int, ...]

videoflow.backends.memory.mig_geometry.check_layout(product: str, layout: Mapping[str, int]) → list[str]

Problems with placing layout (profile name -> instance count) on one card of product: memory over capacity, compute slices over capacity, per-profile instance caps, and — the check the planner lacks — whether the instances can actually be placed at legal, non-overlapping positions.

videoflow.backends.memory.mig_geometry.family_for(product: str) → Family | None

videoflow.backends.memory.mig_geometry.smallest_profile(product: str, memory_gib: float) → Profile | None

videoflow.backends.memory.payload module

The in-memory PayloadStore: objects with digests and generations, named obligations instead of a counter, TTL and byte-budget physics under a fake clock.

Two tiers are modelled because the design package insists they be told apart:

• tier='durable': an object under obligation is never expired or evicted; acquiring an obligation extends the object’s life to the obligation deadline; a put over budget is refused (backpressure) rather than admitted by eviction.

• tier='evictable': the Redis volatile-lru shape (the compose redis-small fixture; the dev server before RFC 0006) — every object has a TTL that fires regardless of obligations, and under memory pressure the least-recently-used TTL-bearing object is evicted even if a reader still needs it. That is the finding the reliable profile must reject; this model makes it observable.

Release is atomic and idempotent by (obligation_id, generation); a stale generation cannot delete a newer object at the same key. An object whose obligation set is absent — a put that died before attaching it, or a set that expired — is never reclaimed by a release (BLOB-14 step 2: the missing set is neither created nor treated as “everyone finished”); reconcile reclaims such an orphan only once it is older than orphan_grace_seconds, as the Redis store does, so a put still in progress is not mistaken for a leak.

Barriers, in put: payload.write.after fires once the bytes are stored and before any obligation is attached; obligation.acquire.after fires once the contract’s obligations are attached. A crash between them leaves exactly the counterless, TTL-only object the lifecycle promises (PAY-007).

class videoflow.backends.memory.payload.MemoryPayloadStore(clock: FakeClock | None = None, tier: str = 'durable', max_bytes: int | None = None, forward_unchanged: bool = True, log: ObservationLog | None = None, store_id: str = 'memory', orphan_grace_seconds: float = 60.0)

Bases: PayloadStore

• Arguments:

  • tier: 'durable' or 'evictable' (see the module docstring).

  • max_bytes: byte budget; None for unlimited.

  • forward_unchanged: when True, put of bytes whose content_id and digest already exist returns the existing reference (reference forwarding: an unchanged frame traverses metadata stages as one object).

  • orphan_grace_seconds: how long an object with no obligation set is left alone by reconcile — a put interrupted before its obligations were attached must not be reclaimed while its publisher may still be about to reference it.

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

capabilities() → PayloadCapabilities

corrupt(key: str) → None

delete(key: str) → None

expire_obligation(key: str, obligation_id: str) → None

Model one obligation record vanishing; an emptied set no longer exists, as on the server.

expire_obligation_set(key: str) → None

Model the whole obligation set expiring or being evicted while the bytes remain (PAY-008).

fail_inventory(reason: str | None) → None

fail_reads(reasons: list[str]) → None

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

object_count() → int

obligations(key: str) → dict[str, float]

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

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

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

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. An object is reclaimed when nothing remains of a set it had; one that never had a set (or whose set expired) is reclaimed only past orphan_grace_seconds, as the Redis store does.

ref_for_key(key: str) → ImmutablePayloadRef

The ref for a wire key (BlobRef.ref), completed from the store’s own metadata when it keeps any: size, digest and generation as recorded at put, so a reader that only has the key can still verify what it reads and release against the right generation. A store without metadata for the key (an RFC 0002 publisher, or metadata that expired) returns a ref with empty digest and generation — readable, unverifiable, and released against nothing.

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

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

stored_bytes() → int

truncate(key: str, size: int) → None

Model a truncated object: the bytes are cut short, the metadata still names the original.

class videoflow.backends.memory.payload.StaticLedger(required: Mapping[str, tuple[str, ...]])

Bases: ObligationLedger

A ledger built from a mapping, for tests and simple runtimes.

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

ref key -> obligation ids still required.

videoflow.backends.memory.runtime_store module

RuntimeStore implementations that need no server: an in-memory one for models, and a file-directory one that survives a process restart on a single host (the local engine’s default), each with compare-and-swap semantics.

class videoflow.backends.memory.runtime_store.FileRuntimeStore(root: str)

Bases: RuntimeStore

One JSON file per key under root (<key>.json with slashes encoded), versioned by a counter inside the file, with an fcntl lock per key for compare-and-swap across the processes of one host. Durable across process restarts; not shared across hosts.

append(log: str, record: bytes) → int

capabilities() → RuntimeCapabilities

cas(key: str, expected_version: str | None, value: bytes) → bool

delete(key: str, expected_version: str | None) → bool

get(key: str) → tuple[bytes | None, str | None]

log_entries(log: str) → list[bytes]

Every record appended to log, oldest first (empty for an unknown log).

scan(prefix: str) → list[tuple[str, bytes, str]]

class videoflow.backends.memory.runtime_store.MemoryRuntimeStore

Bases: RuntimeStore

append(log: str, record: bytes) → int

capabilities() → RuntimeCapabilities

cas(key: str, expected_version: str | None, value: bytes) → bool

delete(key: str, expected_version: str | None) → bool

get(key: str) → tuple[bytes | None, str | None]

log_entries(log: str) → list[bytes]

Every record appended to log, oldest first (empty for an unknown log).

scan(prefix: str) → list[tuple[str, bytes, str]]