vgi.function_storage

Module overview

Storage for VGI function state.

This module provides a storage protocol and implementation for sharing state across worker processes in distributed VGI function execution.

Protocol

FunctionStorage: Unified protocol for all VGI state storage needs.

Implementations

FunctionStorageSqlite: SQLite-backed storage (local/subprocess transport).

FunctionStorageAzureSql: Azure SQL Database-backed storage (cloud deployments).

See vgi.function_storage_azure_sql for details.

FunctionStorageCfDo: Cloudflare Durable Object-backed storage (edge deployments).

See vgi.function_storage_cf_do for details.

function attach_catalog_bytes

source

attach_catalog_bytes(attach_plaintext: bytes | None) -> bytes | None

Strip the framework shard-UUID prefix from a full attach plaintext.

The framework unwraps an attach to uuid(16) || catalog_bytes; function bodies see only catalog_bytes (what the catalog returned). Returns None when there is no attach.

class BoundStorage

source

Methods

method transaction

source

transaction(transaction_opaque_data: bytes) -> TransactionBoundStorage

Return a transaction-scoped storage view.

Used for state that the user expects to be stable across multiple statements in one SQL transaction (e.g. Kafka topic watermarks, for snapshot-isolation reads).

method queue_push

source

queue_push(items: list[bytes]) -> int

Add work items to the queue and register the invocation.

method queue_push_batches

source

queue_push_batches(batches: list[pa.RecordBatch]) -> int

Serialize and push RecordBatches as work items.

method queue_pop

source

queue_pop() -> bytes | None

Atomically claim one work item from the queue.

method queue_pop_batch

source

queue_pop_batch() -> pa.RecordBatch | None

Pop and deserialize one work item as a RecordBatch.

method queue_clear

source

queue_clear() -> int

Clear all remaining work items and unregister the invocation.

method state_get

source

state_get(ns: bytes | FrameworkNS, key: bytes) -> bytes | None

Read one key’s value (or None).

method state_get_many

source

state_get_many(
ns: bytes | FrameworkNS,
keys: list[bytes],
) -> list[bytes | None]

Batched non-destructive read.

method state_put

source

state_put(ns: bytes | FrameworkNS, key: bytes, value: bytes) -> None

Upsert one (key, value).

method state_put_many

source

state_put_many(
ns: bytes | FrameworkNS,
items: list[tuple[bytes, bytes]],
) -> None

Batched atomic upsert.

method state_scan

source

state_scan(
ns: bytes | FrameworkNS,
*,
start: bytes | None = None,
end: bytes | None = None,
reverse: bool = False,
limit: int | None = None,
) -> Iterable[tuple[bytes, bytes]]

Non-destructive scan of (key, value) in one namespace.

Ordered by key bytes (reverse=True for descending), bounded to the half-open range [start, end) and capped at limit. Returns an iterable (the cloudflare-do backend streams it in pages).

method state_drain

source

state_drain(ns: bytes | FrameworkNS) -> Iterable[tuple[bytes, bytes]]

Atomic scan-and-delete of every (key, value) in one namespace.

Returns an iterable; consume it fully (beginning to iterate claims the whole namespace on the cloudflare-do backend).

method state_delete

source

state_delete(
ns: bytes | FrameworkNS,
keys: list[bytes] | None = None,
*,
start: bytes | None = None,
end: bytes | None = None,
) -> int

Delete by key list, by half-open [start, end) range, or wipe all.

keys and the range are mutually exclusive. See FunctionStorage.state_delete for the full contract.

method execution_clear

source

execution_clear() -> int

Wipe ALL state and log rows for this execution across every namespace.

method state_append

source

state_append(ns: bytes | FrameworkNS, key: bytes, item: bytes) -> int

Append an item to the (ns, key) log; return the assigned ordinal.

Idempotency covers transport-layer retries only (HTTP retry on CfDo, pymssql driver-level retry on Azure SQL). Caller-level retries — re-invoking state_append for the same logical record after it returned — produce duplicate rows. See the underlying FunctionStorage.state_append for the full contract.

method state_log_scan

source

state_log_scan(
ns: bytes | FrameworkNS,
key: bytes,
*,
after_id: int = -1,
limit: int | None = None,
) -> list[tuple[int, bytes]]

Yield (id, value) pairs for (ns, key) with id > after_id.

See FunctionStorage.state_log_scan for the full contract.

method counter_get

source

counter_get(ns: bytes | FrameworkNS, key: bytes) -> int

Read the int64 counter (0 if absent).

method counter_add

source

counter_add(ns: bytes | FrameworkNS, key: bytes, delta: int) -> int

Atomically add delta; return the new value. See FunctionStorage.

method counter_set

source

counter_set(ns: bytes | FrameworkNS, key: bytes, value: int) -> None

Overwrite the counter with value.

method counter_delete

source

counter_delete(ns: bytes | FrameworkNS, key: bytes) -> None

Delete the counter (no-op if absent).

method pack_int_key

source

pack_int_key(i: int) -> bytes

Sugar: encode an int as 8-byte little-endian for use as state_* key.

The common case for table_buffering state_id, aggregate group_id, window partition_id is an int. This canonicalizes the encoding so every caller produces the same bytes for the same int.

method serialize_record_batch

source

serialize_record_batch(batch: pa.RecordBatch) -> bytes

Serialize a RecordBatch to Arrow IPC stream bytes.

method deserialize_record_batch

source

deserialize_record_batch(data: bytes) -> pa.RecordBatch

class FrameworkNS

source

Bases: bytes, enum.Enum

Description

Framework-reserved storage namespaces.

All members start with b"_vgi/"; user code may NOT pass a bytes namespace with that prefix to BoundStorage.state_* — the reserved prefix is checked at every entry point. Framework code threads a member of this enum instead; the wrappers accept either form and normalise to plain bytes downstream.

Adding a new entry: keep it ASCII-only, snake_case, prefixed _vgi/. Don’t rename existing entries — names are persisted in sqlite / Azure SQL / CfDo rows on disk and an unbounded backfill would be required.

Attributes

attribute BUFFERING_INIT

attribute STREAMING_FINALIZE

attribute TIO_STATE

attribute AGGREGATE_STATE

attribute AGGREGATE_WINDOW_PARTITION

attribute STREAMING_SESSION

class FunctionStorage

source

Bases: Protocol

Description

Storage protocol for VGI distributed function execution.

Two access patterns:

Unified state_* - Composite-key K/V over (scope_id, ns, key). The catch-all family for per-execution state, per-transaction state, per-group aggregate state, and any other “this caller picks the namespace” pattern. Read-modify-write singletons via state_get_many / state_put_many; non-destructive enumeration via state_scan; atomic scan-and-delete via state_drain; targeted or namespace-wide deletion via state_delete; cross-namespace teardown via execution_clear.

Work Queue - Atomic FIFO work distribution. Producer pushes, workers atomically claim. Distinct from state_* (destructive consume, not key-addressable).

Idempotency: a concern of the remote (HTTP) tier only. The CfDo backend generates an internal attempt_id per call so a retried state_put_many is a silent no-op and a retried state_drain returns the prior values. The local SQLite tier is a single connection per process with no network retries, so it carries no replay-detection (and no idempotency columns).

Eviction / lifecycle. Every scope-keyed table (function_state, function_state_log, function_counter) is reclaimed for a scope by execution_clear — called at operator teardown for execution-scoped state and on commit/rollback for transaction-scoped state. Beyond that, each backend differs: the CfDo DO self-evicts via an orphan-horizon alarm (idle DO → deleteAll); Azure SQL relies on cleanup_old_entries, an age-based sweep over the created_at column that must be scheduled externally (so every age-managed table needs a created_at); the local SQLite tier is durable with no auto-eviction — long-lived, attach-scoped data (e.g. an accumulate collection) is the consumer’s responsibility to bound (ttl / max_row_size / explicit clear). The test_execution_clear_covers_all_scope_keyed_tables audit pins that every scope-keyed table is wiped by execution_clear. (Follow-up: some worker.py teardown paths call execution_clear without a try/except, so a cleanup exception there can still leak — to be hardened separately.)

Methods

method queue_push

source

queue_push(
execution_id: bytes,
items: list[bytes],
*,
shard_key: str = ‘’,
) -> int

Append work items to the queue.

There is no registration step — the queue tracks only the items themselves (matching the Durable Object).

Parameters

execution_id

Unique identifier for the function invocation.

items

List of serialized work item bytes.

shard_key

Routing key for the CF DO backend; ignored by SQLite / Azure backends. Set automatically by BoundStorage from the caller’s attach_opaque_data / auth context.

Returns

Number of items added.

method queue_pop

source

queue_pop(execution_id: bytes, *, shard_key: str = ‘’) -> bytes | None

Atomically claim one work item from the queue.

Parameters

execution_id

Unique identifier for the function invocation.

shard_key

Routing key for the CF DO backend; ignored by SQLite / Azure backends. Set automatically by BoundStorage from the caller’s attach_opaque_data / auth context.

Returns

Serialized work item bytes, or None if the queue is empty or the execution_id was never pushed. There is no registration, so the backend does not distinguish a never-pushed id from a drained queue — both return None (matching the Durable Object).

method queue_clear

source

queue_clear(execution_id: bytes, *, shard_key: str = ‘’) -> int

Clear all remaining work items for the execution.

Parameters

execution_id

Unique identifier for the function invocation.

shard_key

Routing key for the CF DO backend; ignored by SQLite / Azure backends. Set automatically by BoundStorage from the caller’s attach_opaque_data / auth context.

Returns

Number of items deleted.

method state_get_many

source

state_get_many(
scope_id: bytes,
ns: bytes,
keys: list[bytes],
*,
shard_key: str = ‘’,
) -> list[bytes | None]

Batched non-destructive read of values keyed by (scope_id, ns, key).

Returns a list parallel to keys with the stored bytes for hits and None for misses. Single-call so cloud backends (CfDo) can serve a 100-key request as one HTTP roundtrip.

Parameters

scope_id

Caller’s scope identifier (typically execution_id for per-query state, transaction_opaque_data for txn-scoped state).

ns

Caller-chosen namespace bytes; the storage doesn’t interpret.

keys

List of binary keys to look up.

shard_key

CF DO routing key; ignored by SQLite/Azure backends.

Returns

List parallel to keys of stored values or None.

method state_put_many

source

state_put_many(
scope_id: bytes,
ns: bytes,
items: list[tuple[bytes, bytes]],
*,
shard_key: str = ‘’,
) -> None

Batched atomic upsert of (key, value) pairs in one namespace.

Atomic per backend’s single-statement isolation: either every item in the batch is written, or none are. Existing values for the same (scope_id, ns, key) are overwritten.

Remote backends (CfDo) carry an internal attempt_id so an HTTP retry is detected as a replay and silently no-ops. Local backends (SQLite) are a single connection per process with no network retries, so they need no replay-detection.

method state_scan

source

state_scan(
scope_id: bytes,
ns: bytes,
*,
start: bytes | None = None,
end: bytes | None = None,
reverse: bool = False,
limit: int | None = None,
shard_key: str = ‘’,
) -> Iterable[tuple[bytes, bytes]]

Non-destructive scan of (key, value) in one namespace.

Returns an iterable of (key, value) ordered by key bytes (unsigned lexicographic / memcmp). reverse=True orders descending. The scan is bounded to the half-open key range [start, end) (either bound None is open) and capped at limit rows (None = unbounded). Large result sets may be streamed in pages by the backend (the cloudflare-do backend pages under the hood), so callers should iterate rather than assume a materialized list. Use when you need to enumerate an unknown key set (e.g. drainer-side discovery of which sink threads produced state).

method state_drain

source

state_drain(
scope_id: bytes,
ns: bytes,
*,
shard_key: str = ‘’,
) -> Iterable[tuple[bytes, bytes]]

Atomically scan-and-delete every (key, value) in one namespace.

Returns an iterable of (key, value) ordered by key. Remote backends (CfDo) tombstone the rows for HTTP replay-detection (a retried drain returns the same values without re-deleting) and stream the result in pages; local backends delete outright. The drain is atomic — beginning to iterate claims the whole namespace, so always consume it fully.

method state_delete

source

state_delete(
scope_id: bytes,
ns: bytes,
keys: list[bytes] | None = None,
*,
start: bytes | None = None,
end: bytes | None = None,
shard_key: str = ‘’,
) -> int

Delete by key list, by key range, or wipe the entire namespace.

keys=[...] deletes those keys. keys is None with a start and/or end deletes the half-open key range [start, end) (either bound None is open). keys is None with no range wipes the whole namespace. keys and the range are mutually exclusive.

Naturally idempotent — deleting an already-deleted key/range is a no-op. Returns the count of rows actually removed. Replaces today’s per-family *_clear methods.

method execution_clear

source

execution_clear(scope_id: bytes, *, shard_key: str = ‘’) -> int

Wipe ALL state, log, and counter rows for scope_id across every namespace.

Used as a safety-sweep at end-of-execution / on crash recovery. Naturally idempotent. Returns total row count deleted across the function_state, function_state_log, and function_counter tables.

Does NOT touch queue_* rows.

method state_append

source

state_append(
scope_id: bytes,
ns: bytes,
key: bytes,
item: bytes,
*,
shard_key: str = ‘’,
) -> int

Append item to the log keyed by (scope_id, ns, key); return ordinal.

Ordinals are globally monotonic across all (scope, ns, key) triples on a given backend (one IDENTITY/AUTOINCREMENT column for the table). Per-key order is recovered via the (scope_id, ns, key, id) index; state_log_scan yields rows in id order, which corresponds to append order. Concurrent appenders to the same key get distinct ordinals but interleaving across writers is undefined.

Idempotency scope. Remote backends carry an internal attempt_id covering transport-layer retries within a single backend call (an HTTP retry on CfDo replays correctly); local SQLite has no retry layer. Caller-level retries (re-invoking state_append for the same logical record after the call already returned) always produce duplicate rows. If you need caller-level idempotency, dedupe on the caller side — e.g., check state_log_scan before appending, or key your namespace on a stable content hash.

method state_log_scan

source

state_log_scan(
scope_id: bytes,
ns: bytes,
key: bytes,
*,
after_id: int = -1,
limit: int | None = None,
shard_key: str = ‘’,
) -> list[tuple[int, bytes]]

Yield (id, value) pairs for (scope_id, ns, key) with id > after_id.

Returns rows in ascending id order. after_id=-1 is the before-first sentinel (returns from the start). limit=None is unbounded; positive values cap the result at that many rows. Use the returned id of the last row as the next after_id for cursor-based scrolling.

Non-destructive. Repeat calls with the same parameters return identical results until execution_clear wipes the log rows.

method state_counter_get

source

state_counter_get(
scope_id: bytes,
ns: bytes,
key: bytes,
*,
shard_key: str = ‘’,
) -> int

Return the int64 counter at (scope_id, ns, key); 0 if absent.

method state_counter_add

source

state_counter_add(
scope_id: bytes,
ns: bytes,
key: bytes,
delta: int,
*,
shard_key: str = ‘’,
) -> int

Atomically add delta and return the new value (init 0 if absent).

Single-statement upsert — no read-modify-write race, no caller loop. Not idempotent: a retried add double-applies. Remote/cloud backends carry an internal attempt_id (as state_put_many does) so a transport retry replays the prior result instead of re-adding; the local SQLite tier has no retry layer.

method state_counter_set

source

state_counter_set(
scope_id: bytes,
ns: bytes,
key: bytes,
value: int,
*,
shard_key: str = ‘’,
) -> None

Overwrite the counter at (scope_id, ns, key) with value.

method state_counter_delete

source

state_counter_delete(
scope_id: bytes,
ns: bytes,
key: bytes,
*,
shard_key: str = ‘’,
) -> None

Delete the counter at (scope_id, ns, key) (no-op if absent).

class FunctionStorageSqlite

source

Description

SQLite-backed storage for VGI function state.

This implementation uses SQLite with WAL mode to allow multiple worker processes to share state. It manages the three unified tables (the same shape every backend uses):

• function_state: composite-key K/V over (scope_id, ns, key) — the single home for per-execution / per-transaction / per-group / per-pid state
• function_state_log: append-only log keyed by (scope_id, ns, key)
• work_queue: FIFO queue of work items per execution

Attributes

attribute db_path

Path to the SQLite database file. If None, uses a default location in the user’s state directory. Pass ":memory:" to use a process-local in-memory database; the storage uses a shared-cache URI plus an anchor connection so the per-op connections in _connect see the same DB. Suitable for single-process test fixtures where commit-fsync overhead dominates and persistence isn’t needed.

Methods

method close

source

close() -> None

Close the calling thread’s persistent connection, if any.

method queue_push

source

queue_push(
execution_id: bytes,
items: list[bytes],
*,
shard_key: str = ‘’,
) -> int

Append work items to the queue.

method queue_pop

source

queue_pop(execution_id: bytes, *, shard_key: str = ‘’) -> bytes | None

Atomically claim one work item from the queue.

Returns None when the queue is empty or the execution_id was never pushed — there is no registration, matching the Durable Object.

method queue_clear

source

queue_clear(execution_id: bytes, *, shard_key: str = ‘’) -> int

Clear all remaining work items for the execution.

method state_get_many

source

state_get_many(
scope_id: bytes,
ns: bytes,
keys: list[bytes],
*,
shard_key: str = ‘’,
) -> list[bytes | None]

Batched read by key list. Returns parallel list with None for misses.

method state_put_many

source

state_put_many(
scope_id: bytes,
ns: bytes,
items: list[tuple[bytes, bytes]],
*,
shard_key: str = ‘’,
) -> None

Atomic batched upsert by (scope_id, ns, key).

method state_scan

source

state_scan(
scope_id: bytes,
ns: bytes,
*,
start: bytes | None = None,
end: bytes | None = None,
reverse: bool = False,
limit: int | None = None,
shard_key: str = ‘’,
) -> list[tuple[bytes, bytes]]

Non-destructive scan of (key, value) in a namespace.

Ordered by key bytes (BLOB compares bytewise / memcmp), descending when reverse, bounded to [start, end) and capped at limit.

method state_drain

source

state_drain(
scope_id: bytes,
ns: bytes,
*,
shard_key: str = ‘’,
) -> list[tuple[bytes, bytes]]

Atomic destructive scan: read all (key, value) in a namespace and delete them.

method state_delete

source

state_delete(
scope_id: bytes,
ns: bytes,
keys: list[bytes] | None = None,
*,
start: bytes | None = None,
end: bytes | None = None,
shard_key: str = ‘’,
) -> int

Delete by key list, by [start, end) range, or whole namespace.

keys and the range are mutually exclusive. Returns count deleted.

method execution_clear

source

execution_clear(scope_id: bytes, *, shard_key: str = ‘’) -> int

Wipe all state, log, and counter rows for scope_id across every namespace.

method state_append

source

state_append(
scope_id: bytes,
ns: bytes,
key: bytes,
item: bytes,
*,
shard_key: str = ‘’,
) -> int

Append item to the (scope_id, ns, key) log; return its ordinal (the row id).

method state_log_scan

source

state_log_scan(
scope_id: bytes,
ns: bytes,
key: bytes,
*,
after_id: int = -1,
limit: int | None = None,
shard_key: str = ‘’,
) -> list[tuple[int, bytes]]

Yield (id, value) pairs for (scope_id, ns, key) with id > after_id.

method state_counter_get

source

state_counter_get(
scope_id: bytes,
ns: bytes,
key: bytes,
*,
shard_key: str = ‘’,
) -> int

Read the int64 counter; 0 if absent.

method state_counter_add

source

state_counter_add(
scope_id: bytes,
ns: bytes,
key: bytes,
delta: int,
*,
shard_key: str = ‘’,
) -> int

Atomically add delta and return the new value (init 0 if absent).

method state_counter_set

source

state_counter_set(
scope_id: bytes,
ns: bytes,
key: bytes,
value: int,
*,
shard_key: str = ‘’,
) -> None

Overwrite the counter with value.

method state_counter_delete

source

state_counter_delete(
scope_id: bytes,
ns: bytes,
key: bytes,
*,
shard_key: str = ‘’,
) -> None

Delete the counter (no-op if absent).

class ShardedSqliteStorage

source

Description

Debug-only SQLite backend that PARTITIONS storage by shard_key.

The normal SQLite backend ignores shard_key (one shared DB), masking shard-routing bugs that only bite cloudflare-do (which truly shards per Durable Object). This wrapper isolates shards by PREFIXING the scope_id / execution_id with the shard_key, so an op under shard A can’t see state written under shard B — reproducing cloudflare-do isolation locally — while using ONE inner store, so concurrency behaves exactly like the normal sqlite backend. (Per-shard databases instead exploded connections and deadlocked the shared-cache :memory: DB under load.) Enabled via VGI_SQLITE_SHARD=1 (see vgi/function.py:_resolve_storage). Not for production.

With VGI_SQLITE_SHARD_LOG=1 it logs every op’s (op, shard_key, scope) so a write and a read for one execution can be compared without a remote tail.

Methods

method queue_push

source

queue_push(
execution_id: bytes,
items: list[bytes],
*,
shard_key: str = ‘’,
) -> int

method queue_pop

source

queue_pop(execution_id: bytes, *, shard_key: str = ‘’) -> bytes | None

method queue_clear

source

queue_clear(execution_id: bytes, *, shard_key: str = ‘’) -> int

method state_get_many

source

state_get_many(
scope_id: bytes,
ns: bytes,
keys: list[bytes],
*,
shard_key: str = ‘’,
) -> list[bytes | None]

method state_put_many

source

state_put_many(
scope_id: bytes,
ns: bytes,
items: list[tuple[bytes, bytes]],
*,
shard_key: str = ‘’,
) -> None

method state_scan

source

state_scan(
scope_id: bytes,
ns: bytes,
*,
start: bytes | None = None,
end: bytes | None = None,
reverse: bool = False,
limit: int | None = None,
shard_key: str = ‘’,
) -> list[tuple[bytes, bytes]]

method state_drain

source

state_drain(
scope_id: bytes,
ns: bytes,
*,
shard_key: str = ‘’,
) -> list[tuple[bytes, bytes]]

method state_delete

source

state_delete(
scope_id: bytes,
ns: bytes,
keys: list[bytes] | None = None,
*,
start: bytes | None = None,
end: bytes | None = None,
shard_key: str = ‘’,
) -> int

method execution_clear

source

execution_clear(scope_id: bytes, *, shard_key: str = ‘’) -> int

method state_append

source

state_append(
scope_id: bytes,
ns: bytes,
key: bytes,
item: bytes,
*,
shard_key: str = ‘’,
) -> int

method state_log_scan

source

state_log_scan(
scope_id: bytes,
ns: bytes,
key: bytes,
*,
after_id: int = -1,
limit: int | None = None,
shard_key: str = ‘’,
) -> list[tuple[int, bytes]]

method state_counter_get

source

state_counter_get(
scope_id: bytes,
ns: bytes,
key: bytes,
*,
shard_key: str = ‘’,
) -> int

method state_counter_add

source

state_counter_add(
scope_id: bytes,
ns: bytes,
key: bytes,
delta: int,
*,
shard_key: str = ‘’,
) -> int

method state_counter_set

source

state_counter_set(
scope_id: bytes,
ns: bytes,
key: bytes,
value: int,
*,
shard_key: str = ‘’,
) -> None

method state_counter_delete

source

state_counter_delete(
scope_id: bytes,
ns: bytes,
key: bytes,
*,
shard_key: str = ‘’,
) -> None

method close

source

close() -> None

class TransactionBoundStorage

source

Description

Convenience wrapper bound to a single transaction_opaque_data.

Lets a function read/write transaction-scoped state without threading the transaction_opaque_data through every call site. Get one via BoundStorage.transaction(transaction_opaque_data).

Methods

method get

source

get(keys: list[bytes]) -> list[bytes | None]

Load values for a list of keys; parallel return list.

method get_one

source

get_one(key: bytes) -> bytes | None

Load a single value, or None if missing.

method put

source

put(items: list[tuple[bytes, bytes]]) -> None

Write a batch of (key, value) pairs.

method put_one

source

put_one(key: bytes, value: bytes) -> None

Write a single (key, value) pair.

method clear

source

clear() -> None

Drop every value for this transaction (every namespace).