State storage

On this page

State that outlives one call, or crosses worker processes.

class `BoundStorage`

public final class BoundStorage

Description

The storage facade handed to function code: a FunctionStorage view bound to one execution_id and shard-pinned to one logical ATTACH. Mirrors vgi-python’s BoundStorage (params.storage).

Construction resolves the shard once: an attach plaintext (laid out uuid(16) || catalog_bytes) yields ShardKey#derive and pins the backend via FunctionStorage#forShard; with no attach, a backend that FunctionStorage#requiresShardKey() requires sharding (the Durable Object tier) is refused, while local sqlite tiers run unpinned.

Namespaces passed as raw bytes are rejected when they begin with the reserved _vgi/ prefix; framework code uses the FrameworkNs overloads, which bypass the check.

Members

BoundStorage(FunctionStorage backend, byte[] executionId, byte[] attachPlaintext)

Binds a backend to one execution, resolving the shard from the attach.

byte[] executionId()

The execution_id this view is scoped to.

BoundStorage rescope(byte[] scopeId)

A view over the same (already shard-pinned) backend bound to a different scope id — e.g. attach-scoped state that must persist across queries, mirroring vgi-python fixtures’ BoundStorage(storage, attach_bytes).

byte[] stateGet(byte[] ns, byte[] key)

Single-key get in a user namespace.

byte[] stateGet(FrameworkNs ns, byte[] key)

Single-key get in a framework namespace.

List<byte[]> stateGetMany(byte[] ns, List<byte[]> keys)

Batch get in a user namespace.

List<byte[]> stateGetMany(FrameworkNs ns, List<byte[]> keys)

Batch get in a framework namespace.

void statePut(byte[] ns, byte[] key, byte[] value)

Single-key upsert in a user namespace.

void statePut(FrameworkNs ns, byte[] key, byte[] value)

Single-key upsert in a framework namespace.

void statePutMany(byte[] ns, List<FunctionStorage.KV> items)

Batch upsert in a user namespace.

void statePutMany(FrameworkNs ns, List<FunctionStorage.KV> items)

Batch upsert in a framework namespace.

List<FunctionStorage.KV> stateScan(byte[] ns, byte[] start, byte[] end, boolean reverse, int limit)

Ordered scan of a user namespace; see FunctionStorage#stateScan for the range/order contract.

List<FunctionStorage.KV> stateScan(FrameworkNs ns, byte[] start, byte[] end, boolean reverse, int limit)

Ordered scan of a framework namespace.

List<FunctionStorage.KV> stateDrain(byte[] ns)

Atomic destructive scan of a user namespace: reads and deletes every row.

List<FunctionStorage.KV> stateDrain(FrameworkNs ns)

Atomic destructive scan of a framework namespace.

int stateDelete(byte[] ns, List<byte[]> keys)

Deletes the given keys in a user namespace (missing keys are ignored).

int stateDelete(FrameworkNs ns, List<byte[]> keys)

Deletes the given keys in a framework namespace.

int stateDeleteRange(byte[] ns, byte[] start, byte[] end)

Deletes the half-open key range [start, end) in a user namespace; both bounds null wipes the whole namespace.

int stateDeleteRange(FrameworkNs ns, byte[] start, byte[] end)

Deletes the half-open key range [start, end) in a framework namespace; both bounds null wipes the whole namespace.

long stateAppend(byte[] ns, byte[] key, byte[] value)

Appends a value to the (executionId, ns, key) log in a user namespace.

long stateAppend(FrameworkNs ns, byte[] key, byte[] value)

Appends a value to the log in a framework namespace.

List<FunctionStorage.LogEntry> stateLogScan(byte[] ns, byte[] key, long afterId, int limit)

Scans the (executionId, ns, key) log after a cursor in a user namespace.

List<FunctionStorage.LogEntry> stateLogScan(FrameworkNs ns, byte[] key, long afterId, int limit)

Scans the log after a cursor in a framework namespace.

long counterGet(byte[] ns, byte[] key)

Reads a counter in a user namespace.

long counterGet(FrameworkNs ns, byte[] key)

Reads a counter in a framework namespace.

long counterAdd(byte[] ns, byte[] key, long delta)

Atomically adds delta to a counter in a user namespace and returns the new value (an absent counter starts at 0).

long counterAdd(FrameworkNs ns, byte[] key, long delta)

Atomically adds delta to a counter in a framework namespace.

void counterSet(byte[] ns, byte[] key, long value)

Overwrites a counter in a user namespace.

void counterSet(FrameworkNs ns, byte[] key, long value)

Overwrites a counter in a framework namespace.

void counterDelete(byte[] ns, byte[] key)

Deletes a counter in a user namespace (no-op when absent).

void counterDelete(FrameworkNs ns, byte[] key)

Deletes a counter in a framework namespace (no-op when absent).

int executionClear()

Wipes all state, log, and counter rows for this execution across every namespace (queue items are untouched — use #queueClear).

int queuePush(List<byte[]> items)

Appends work items to this execution’s FIFO queue.

byte[] queuePop()

Atomically claims the oldest work item from this execution’s queue.

int queueClear()

Removes all remaining work items for this execution.

int queuePushBatches(List<VectorSchemaRoot> batches)

Serializes each batch via Arrow IPC and pushes them as work items.

VectorSchemaRoot queuePopBatch(BufferAllocator allocator)

Pops one work item and deserializes it as an Arrow batch.

TransactionStorage transaction(byte[] transactionOpaqueData)

A transaction-scoped key/value view sharing this facade’s shard pinning.

byte[] packIntKey(long i)

Canonical integer key encoding for state_* keys: 8-byte little-endian signed, byte-identical to vgi-python’s pack_int_key.

byte[] serializeRecordBatch(VectorSchemaRoot root)

Serializes a batch as a single-batch Arrow IPC stream (the queue and buffering wire encoding, identical to the Python helpers).

class `CfdoException`

source

public static final class CfdoException extends RuntimeException

Description

Thrown on a non-2xx response from the Worker, or a transport error.

Members

CfdoException(String message)

Creates the exception with the Worker’s error body or the transport failure message.

CfdoStorage(String baseUrl, String token)

Creates an unpinned client; call #forShard before issuing requests.

CfdoStorage fromEnv()

Builds from VGI_CF_DO_URL / VGI_CF_DO_TOKEN.

CfdoStorage forShard(String shard)

Returns a view pinned to one shard key, routing per logical ATTACH.

boolean requiresShardKey()

The Durable Object routes on shard_key (idFromName) — unsharded use is refused, so BoundStorage demands an attach identity.

long stateAppend(byte[] scopeId, byte[] ns, byte[] key, byte[] value)

List<LogEntry> stateLogScan(byte[] scopeId, byte[] ns, byte[] key, long afterId, int limit)

int executionClear(byte[] scopeId)

Wipe all state + log rows for a scope across every namespace.

List<byte[]> stateGetMany(byte[] scopeId, byte[] ns, List<byte[]> keys)

void statePutMany(byte[] scopeId, byte[] ns, List<KV> items)

int stateDelete(byte[] scopeId, byte[] ns, List<byte[]> keys)

int stateDeleteRange(byte[] scopeId, byte[] ns, byte[] start, byte[] end)

List<KV> stateScan(byte[] scopeId, byte[] ns, byte[] start, byte[] end, boolean reverse, int limit)

List<KV> stateDrain(byte[] scopeId, byte[] ns)

long stateCounterGet(byte[] scopeId, byte[] ns, byte[] key)

long stateCounterAdd(byte[] scopeId, byte[] ns, byte[] key, long delta)

void stateCounterSet(byte[] scopeId, byte[] ns, byte[] key, long value)

void stateCounterDelete(byte[] scopeId, byte[] ns, byte[] key)

int queuePush(byte[] executionId, List<byte[]> items)

byte[] queuePop(byte[] executionId)

int queueClear(byte[] executionId)

class `CfdoStorage`

source

public final class CfdoStorage implements FunctionStorage

Description

Cloudflare Durable Object storage client, speaking the Worker’s unified state_* JSON+base64 protocol (vgi-cloudflare-durable-object-storage/src/index.ts). Every request carries the per-attach shard_key (set via #forShard); destructive ops carry a fresh 32-hex attempt_id. Mirrors vgi-python / vgi-go.

Implements the full unified surface — composite-key K/V with ranged scan/drain/delete, the append-only log, atomic int64 counters, and the FIFO work queue. state_scan / state_drain page under a server-side byte budget via an opaque after_key/next_after continuation cursor; a drain mints ONE attempt_id and reuses it on every page so the server’s snapshot-then-page semantics stay atomic and replay-safe. It is the distributed tier of FunctionStorage.

enum `FrameworkNs`

source

public enum FrameworkNs

Description

Framework-reserved storage namespaces, all under the _vgi/ prefix that BoundStorage refuses for user-supplied namespaces. Byte-for-byte identical to vgi-python’s FrameworkNS so storage rows stay portable across SDKs. The names are persisted in storage rows — never rename a member value.

Members

byte[] bytes()

The namespace’s wire bytes (UTF-8 of the _vgi/... name).

interface `FunctionStorage`

source

public interface FunctionStorage extends AutoCloseable

Description

The worker’s shared-state surface, unified across the three deployment tiers (mirrors vgi-python / vgi-go / vgi-typescript and the Cloudflare DO protocol):

in-process — SqliteFunctionStorage at :memory:; process-local, no cross-process coordination. Single-process only.
local cross-process — SqliteFunctionStorage at a file; WAL coordinates worker subprocesses / multi-worker HTTP on one box.
distributed — CfdoStorage, a Cloudflare Durable Object over HTTP, for multi-replica / edge deployments.

Selected at startup by StorageResolver#fromEnv() via VGI_WORKER_SHARED_STORAGE. Every state row is addressed by (scope_id, ns, key), where scope_id is the execution_id (buffering / aggregate) or transaction_opaque_data (transactions), and ns namespaces the kind of state. Work-queue rows are addressed by execution_id alone. Functions normally reach this surface through the per-execution BoundStorage facade rather than directly.

class `ShardKey`

source

public final class ShardKey

Description

Derives the Cloudflare Durable Object routing key for an attach.

The attach id is laid out uuid(16) || ... (catalog_attach mints a fresh 16-byte UUID at the head). The shard key is "att-" + hex(uuid) — one DO per logical ATTACH, stable across re-seals and globally unique (unlike the random-nonce ciphertext or the catalog options bytes). Mirrors vgi-python’s _derive_shard_key and the Go/TypeScript deriveShardKey.

Members

int ATTACH_UUID_LEN = 16

Width of the framework UUID at the head of every attach id.

String derive(byte[] attachId)

Derives the shard key for an attach id.

class `SqliteFunctionStorage`

source

public final class SqliteFunctionStorage implements FunctionStorage

Description

SQLite-backed FunctionStorage for the in-process (:memory:) and local cross-process (file) tiers.

Schema and pragmas mirror vgi-go / vgi-python / vgi-typescript: a function_state key/value table, a function_state_log append log whose AUTOINCREMENT id is the resumable scan cursor (same as vgi-go’s state_log), a function_counter table of atomic int64 counters, and a work_queue FIFO. The state tables are addressed by (scope_id, ns, key). The constructor self-heals an older on-disk DB: tables carrying the pre-unification HTTP-idempotency columns (or obsolete tables eliminated by the unified schema) are dropped and recreated — all of this is ephemeral in-progress worker state, so dropping is safe.

Concurrency follows vgi-go’s model: a single long-lived connection per process serializes operations (this class synchronizes every method), and WAL

busy_timeout let multiple worker processes sharing a file see each other’s rows. A single held connection is also what keeps a :memory: database alive for the worker’s lifetime (an in-memory DB exists only as long as a connection to it is open).

Members

String SQLITE_PARAMS = “journal_mode=WAL&busy_timeout=30000&synchronous=NORMAL&temp_store=MEMORY&cache_size=-65536”

Per-connection pragmas, identical to AggregateStateStore.SQLITE_PARAMS and the other SDKs: WAL for cross-process concurrency, synchronous=NORMAL, a 30s busy_timeout, temp_store=MEMORY, and a 64 MiB page cache. (WAL is a no-op for :memory:, which SQLite keeps in “memory” journal mode.)

SqliteFunctionStorage(String path)

Opens (and initializes the schema of) a SQLite database.

long stateAppend(byte[] scopeId, byte[] ns, byte[] key, byte[] value)

List<LogEntry> stateLogScan(byte[] scopeId, byte[] ns, byte[] key, long afterId, int limit)

List<byte[]> stateGetMany(byte[] scopeId, byte[] ns, List<byte[]> keys)

void statePutMany(byte[] scopeId, byte[] ns, List<KV> items)

int stateDelete(byte[] scopeId, byte[] ns, List<byte[]> keys)

int stateDeleteRange(byte[] scopeId, byte[] ns, byte[] start, byte[] end)

List<KV> stateScan(byte[] scopeId, byte[] ns, byte[] start, byte[] end, boolean reverse, int limit)

List<KV> stateDrain(byte[] scopeId, byte[] ns)

long stateCounterGet(byte[] scopeId, byte[] ns, byte[] key)

long stateCounterAdd(byte[] scopeId, byte[] ns, byte[] key, long delta)

void stateCounterSet(byte[] scopeId, byte[] ns, byte[] key, long value)

void stateCounterDelete(byte[] scopeId, byte[] ns, byte[] key)

int queuePush(byte[] executionId, List<byte[]> items)

byte[] queuePop(byte[] executionId)

int queueClear(byte[] executionId)

int executionClear(byte[] scopeId)

void close()

class `StorageResolver`

source

public final class StorageResolver

Description

Picks the FunctionStorage tier at worker startup from VGI_WORKER_SHARED_STORAGE, mirroring vgi-python’s vgi.function._resolve_storage and vgi-go’s resolve.FromEnv:

memory — in-process SqliteFunctionStorage at :memory:. Process-local, no cross-process coordination; single-process deployments only.
sqlite (default, or unset) — local cross-process SqliteFunctionStorage at a file (honors VGI_WORKER_SQLITE_PATH).
cloudflare-do — distributed CfdoStorage (requires VGI_CF_DO_URL, optionally VGI_CF_DO_TOKEN).

Members

String ENV_VAR = “VGI_WORKER_SHARED_STORAGE”

The environment variable selecting the storage tier.

FunctionStorage fromEnv()

Resolves the configured backend.

class `TransactionBoundStorage`

source

public final class TransactionBoundStorage implements TransactionStorage

Description

A TransactionStorage backed by the unified state surface: scope_id = transaction_opaque_data, namespace txn (byte-identical to vgi-python’s TransactionBoundStorage). Obtain one via BoundStorage#transaction (which shares the parent’s shard pinning) or framework-side from TransactionStore.

Members

TransactionBoundStorage(FunctionStorage store, byte[] transactionOpaqueData)

Binds a (possibly shard-pinned) backend to one transaction.

byte[] getOne(byte[] key)

void putOne(byte[] key, byte[] value)

List<byte[]> getMany(List<byte[]> keys)

void putMany(List<FunctionStorage.KV> items)

int clear()