Skip to content
Query.Farm
Talk with Us

State storage

On this page

Cross-process state: the store, its backends, and state codecs.

struct AggregateConstArgs

Section titled “struct AggregateConstArgs”
source
type AggregateConstArgs struct {
FunctionName  string
Args    []byte
}

Description

AggregateConstArgs is a (function_name, args) pair stashed at bind time for an aggregate, recovered at finalize time. Unlike the Python protocol (where const args ride the wire on every phase), Go’s aggregate runtime stashes them in storage so all worker processes see the same bind-time arguments. Backends can implement this against any (execution_id, function_name) → bytes K/V table.

struct AggregateStateEntry

Section titled “struct AggregateStateEntry”
source
type AggregateStateEntry struct {
GroupID  int64
State  []byte
}

Description

AggregateStateEntry is one (group_id, state) pair.

struct AttachScanOptions

Section titled “struct AttachScanOptions”
source
type AttachScanOptions struct {
Start  []byte
End  []byte
Reverse  bool
Limit  int
}

Description

AttachScanOptions bounds an AttachStateScan. Start/End describe a half-open key range [Start, End) compared bytewise (a nil bound is open on that side); Reverse flips the result order to descending; Limit caps the number of rows returned, with Limit <= 0 meaning no limit (matching StateLogScan). The zero value scans the whole namespace ascending. Mirrors vgi-python’s state_scan keyword arguments. The range predicate is independent of Reverse: Start is always the >= bound and End always the < bound regardless of direction.

struct AttachStateKV

Section titled “struct AttachStateKV”
source
type AttachStateKV struct {
Key  []byte
Value  []byte
}

Description

AttachStateKV is one (key, value) pair returned by an attach-state scan, ordered by key.

interface AttachStateStorage

Section titled “interface AttachStateStorage”
source
type AttachStateStorage interface {
// AttachStatePut stores or replaces value under (scope, ns, key).
AttachStatePut(scope, ns, key, value []byte) error
// AttachStateGet returns the value under (scope, ns, key), or (nil, nil) if absent.
AttachStateGet(scope, ns, key []byte) ([]byte, error)
// AttachStateScan returns the (key, value) pairs in (scope, ns) ordered by
// key, bounded and limited per opts. See AttachScanOptions.
AttachStateScan(scope, ns []byte, opts AttachScanOptions) ([]AttachStateKV, error)
// AttachStateDeleteKey removes one key. No-op if absent.
AttachStateDeleteKey(scope, ns, key []byte) error
// AttachStateDeleteNS removes every key in (scope, ns).
AttachStateDeleteNS(scope, ns []byte) error
// AttachStateDeleteRange removes every key in the half-open range
// [start, end) of (scope, ns) (a nil bound is open on that side) and returns
// the number of keys removed.
AttachStateDeleteRange(scope, ns, start, end []byte) (int, error)
// AttachStateDrain atomically reads and removes every (key, value) in
// (scope, ns), returning them ordered by key.
AttachStateDrain(scope, ns []byte) ([]AttachStateKV, error)


// AttachCounterGet returns the int64 counter under (scope, ns, key), or 0 if
// absent.
AttachCounterGet(scope, ns, key []byte) (int64, error)
// AttachCounterAdd atomically adds delta to the counter under (scope, ns,
// key) (initializing absent counters to 0) and returns the new value.
AttachCounterAdd(scope, ns, key []byte, delta int64) (int64, error)
// AttachCounterSet overwrites the counter under (scope, ns, key) with value.
AttachCounterSet(scope, ns, key []byte, value int64) error
// AttachCounterDelete removes the counter under (scope, ns, key). No-op if
// absent.
AttachCounterDelete(scope, ns, key []byte) error
}

Description

AttachStateStorage is an optional capability: a scoped, namespaced, ordered key/value store that persists for the life of the shared backend rather than a single execution — i.e. across queries (and, for the SQLite backend, across worker processes). It is keyed by (scope, ns, key); Scan returns the entries in a (scope, ns) ordered by key. Attach-scoped fixtures such as the accumulate example use it to keep per-ATTACH row collections alive between the fresh worker processes a subprocess-transport query spawns. Mirrors the subset of vgi-python’s attach-scoped BoundStorage those fixtures rely on.

Implemented by the SQLite backend (over the same function_state table that backs worker/transaction state); backends that don’t implement it cause AttachStore() to return an error at runtime.

struct AttachStore

Section titled “struct AttachStore”
source
type AttachStore struct {
back  AttachStateStorage
scope  []byte
}

Description

AttachStore is a namespaced, ordered key/value view bound to one ATTACH scope. Unlike ExecutionStorage’s execution-scoped state, it persists for the life of the shared backend, so it survives the fresh worker process a subprocess-transport query spawns. Used by attach-scoped fixtures such as the accumulate example.

Methods

method CounterAdd

Section titled “method CounterAdd”
source
func (a *AttachStore) CounterAdd(ns, key []byte, delta int64) (int64, error)

CounterAdd atomically adds delta to the counter under (ns, key) and returns the new value.

method CounterDelete

Section titled “method CounterDelete”
source
func (a *AttachStore) CounterDelete(ns, key []byte) error

CounterDelete removes the counter under (ns, key). No-op if absent.

method CounterGet

Section titled “method CounterGet”
source
func (a *AttachStore) CounterGet(ns, key []byte) (int64, error)

CounterGet returns the int64 counter under (ns, key), or 0 if absent.

method CounterSet

Section titled “method CounterSet”
source
func (a *AttachStore) CounterSet(ns, key []byte, value int64) error

CounterSet overwrites the counter under (ns, key) with value.

method DeleteKey

Section titled “method DeleteKey”
source
func (a *AttachStore) DeleteKey(ns, key []byte) error

DeleteKey removes one key under ns. No-op if absent.

method DeleteNS

Section titled “method DeleteNS”
source
func (a *AttachStore) DeleteNS(ns []byte) error

DeleteNS removes every key under ns.

method DeleteRange

Section titled “method DeleteRange”
source
func (a *AttachStore) DeleteRange(ns, start, end []byte) (int, error)

DeleteRange removes every key in the half-open range [start, end) under ns (a nil bound is open on that side) and returns the number removed.

method Drain

Section titled “method Drain”
source
func (a *AttachStore) Drain(ns []byte) ([]AttachStateKV, error)

Drain atomically reads and removes every (key, value) under ns, returning them ordered by key.

method Get

Section titled “method Get”
source
func (a *AttachStore) Get(ns, key []byte) ([]byte, error)

Get returns the value under (ns, key), or (nil, nil) if absent.

method Put

Section titled “method Put”
source
func (a *AttachStore) Put(ns, key, value []byte) error

Put stores or replaces value under (ns, key) in this attach scope.

method Scan

Section titled “method Scan”
source
func (a *AttachStore) Scan(ns []byte, opts …ScanOption) ([]AttachStateKV, error)

Scan returns the (key, value) pairs under ns ordered by key. With no options it returns the whole namespace ascending; pass WithRange/WithReverse/WithLimit to bound it.

function newAttachStore

Section titled “function newAttachStore”
source
func newAttachStore(back FunctionStorage, scope []byte) (*AttachStore, error)

newAttachStore binds an AttachStateStorage-capable backend to one scope.

struct ExecutionStorage

Section titled “struct ExecutionStorage”
source
type ExecutionStorage struct {
mu    sync.Mutex
back    FunctionStorage
executionID  []byte
// shardKey routes per logical ATTACH for the CfDo backend (att-<hex uuid>);
// "" for non-attach / non-sharding paths. The framework sets it from the
// unwrapped attach UUID when the execution's storage is created.
shardKey  string
}

Description

ExecutionStorage binds a FunctionStorage to one execution_id.

Methods

method AttachStore

Section titled “method AttachStore”
source
func (s *ExecutionStorage) AttachStore(scope []byte) (*AttachStore, error)

AttachStore returns an attach-scoped key/value store bound to scope, using this execution’s underlying backend. The scope is typically ProcessParams.AttachScope. Errors if the backend lacks AttachStateStorage.

method Cleanup

Section titled “method Cleanup”
source
func (s *ExecutionStorage) Cleanup()

Cleanup drops every record under this execution_id: all scope-keyed state (worker, scan-worker, aggregate, attach, log, counters — via ExecutionClear) plus the separately-keyed work queue. The underlying FunctionStorage is owned by the Worker and is shared across executions; it is NOT closed here.

method Collect

Section titled “method Collect”
source
func (s *ExecutionStorage) Collect() ([][]byte, error)

Collect returns all stored worker values and removes them.

method ExecutionID

Section titled “method ExecutionID”
source
func (s *ExecutionStorage) ExecutionID() []byte

ExecutionID returns the bound execution_id, or nil if unset.

function NewExecutionStorage

Section titled “function NewExecutionStorage”
source
func NewExecutionStorage() *ExecutionStorage

NewExecutionStorage creates a new unbound ExecutionStorage. SetBackend and SetExecutionID must be called before use; the Worker does this for you.

method Put

Section titled “method Put”
source
func (s *ExecutionStorage) Put(data []byte) error

Put stores a value keyed by the current worker PID. Upsert semantics.

method QueuePop

Section titled “method QueuePop”
source
func (s *ExecutionStorage) QueuePop() ([]byte, error)

QueuePop atomically claims one item. Returns (nil, nil) when the queue is empty or the execution_id was never pushed (no registration).

method QueuePopBatch

Section titled “method QueuePopBatch”
source
func (s *ExecutionStorage) QueuePopBatch() (arrow.RecordBatch, error)

QueuePopBatch claims and deserializes the next batch, or (nil, nil) if empty.

method QueuePush

Section titled “method QueuePush”
source
func (s *ExecutionStorage) QueuePush(items [][]byte) error

QueuePush appends items to the per-execution work queue.

method QueuePushBatches

Section titled “method QueuePushBatches”
source
func (s *ExecutionStorage) QueuePushBatches(batches []arrow.RecordBatch) error

QueuePushBatches serializes record batches and appends them.

method SetBackend

Section titled “method SetBackend”
source
func (s *ExecutionStorage) SetBackend(back FunctionStorage)

SetBackend wires a FunctionStorage into this binding wrapper. Called once by the framework before SetExecutionID.

method SetExecutionID

Section titled “method SetExecutionID”
source
func (s *ExecutionStorage) SetExecutionID(execID []byte) error

SetExecutionID binds this wrapper to one execution_id.

method SetShardKey

Section titled “method SetShardKey”
source
func (s *ExecutionStorage) SetShardKey(shardKey string)

SetShardKey pins the per-attach routing key (att-<hex uuid>). A no-op for backends that ignore sharding; used by the CfDo backend to route to the right Durable Object. Empty means “no attach” (CfDo would reject it).

method Snapshot

Section titled “method Snapshot”
source
func (s *ExecutionStorage) Snapshot() ([][]byte, error)

Snapshot returns all stored worker values without removing them.

method StateAppend

Section titled “method StateAppend”
source
func (s *ExecutionStorage) StateAppend(key, value []byte) (int64, error)

StateAppend appends value under key in the execution-scoped log, returning the new monotonic log id.

method StateLogClear

Section titled “method StateLogClear”
source
func (s *ExecutionStorage) StateLogClear() error

StateLogClear removes all state-log rows for this execution.

method StateLogScan

Section titled “method StateLogScan”
source
func (s *ExecutionStorage) StateLogScan(key []byte, afterID int64, limit int) ([]StateLogEntry, error)

StateLogScan returns entries under key with id > afterID (use -1 from the start), ordered by id. limit <= 0 means no limit.

method resolve

Section titled “method resolve”
source
func (s *ExecutionStorage) resolve() (FunctionStorage, []byte, error)

method stateLog

Section titled “method stateLog”
source
func (s *ExecutionStorage) stateLog() (StateLogStorage, []byte, error)

interface FunctionStorage

Section titled “interface FunctionStorage”
source
type FunctionStorage interface {


// WorkerPut stores or replaces the state for one worker process under
// the given execution_id.
WorkerPut(executionID []byte, workerID int64, state []byte) error


// WorkerCollect atomically reads and deletes all worker states for an
// execution_id. Typically called by the primary worker at finalize time.
WorkerCollect(executionID []byte) ([][]byte, error)


// WorkerScan reads all worker states without deleting them. Order is
// implementation-defined. Used by best-effort end-of-stream consumers
// like dynamic_to_string where multiple readers see the same state.
WorkerScan(executionID []byte) ([]WorkerStateEntry, error)


// ScanWorkerPut stores or replaces state for one scan worker.
ScanWorkerPut(executionID, streamID, state []byte) error


// ScanWorkerScan reads all per-stream-worker states without deleting.
ScanWorkerScan(executionID []byte) ([]ScanWorkerStateEntry, error)


// QueuePush appends items to the queue for the given execution_id. There
// is no registration step (matching the Cloudflare DO).
QueuePush(executionID []byte, items [][]byte) (int, error)


// QueuePop atomically claims one item from the queue. Returns:
//   - (item, nil) when an item was claimed.
//   - (nil, nil) when the queue is empty or the execution_id was never
//     pushed (the two are indistinguishable — no registration).
QueuePop(executionID []byte) ([]byte, error)


// QueueClear removes all remaining items for an execution_id. Returns the
// number of items dropped.
QueueClear(executionID []byte) (int, error)


// AggregateStateGet loads states for the given group_ids. Returns a
// list parallel to group_ids: each entry is the state for that group
// or nil if no state has been stored. DuckDB's thread-local hash tables
// guarantee no two callers race on the same group_id during update.
AggregateStateGet(executionID []byte, groupIDs []int64) ([]AggregateStateEntry, error)


// AggregateStatePut writes states for the given group_ids using
// INSERT OR REPLACE semantics.
AggregateStatePut(executionID []byte, entries []AggregateStateEntry) error


// AggregateStateClear drops all aggregate state for an execution_id.
AggregateStateClear(executionID []byte) error


// AggregateConstArgsPut stashes serialized bind-time arguments for an
// aggregate execution.
AggregateConstArgsPut(executionID []byte, functionName string, args []byte) error


// AggregateConstArgsGet loads previously stashed arguments. Returns
// (nil, nil) if no args have been stashed (the aggregate had no const args).
AggregateConstArgsGet(executionID []byte, functionName string) ([]byte, error)


// AggregateWindowPartitionPut writes the cached payload for a single
// window-aggregate partition. INSERT OR REPLACE.
AggregateWindowPartitionPut(executionID []byte, partitionID int64, data []byte) error


// AggregateWindowPartitionGet loads the cached payload for a window
// partition, or (nil, nil) if absent.
AggregateWindowPartitionGet(executionID []byte, partitionID int64) ([]byte, error)


// AggregateWindowPartitionDelete removes one partition. No-op if absent.
AggregateWindowPartitionDelete(executionID []byte, partitionID int64) error


// AggregateWindowPartitionClear drops every cached partition for an
// execution_id (safety sweep for dropped destructor RPCs).
AggregateWindowPartitionClear(executionID []byte) error


// TransactionStateGet loads values for the given keys under one
// transaction_opaque_data. Returns a list parallel to keys: nil entries for
// keys with no stored value.
TransactionStateGet(transactionOpaqueData []byte, keys [][]byte) ([][]byte, error)


// TransactionStatePut writes (key, value) pairs for a transaction_opaque_data
// using INSERT OR REPLACE semantics.
TransactionStatePut(transactionOpaqueData []byte, items []TransactionStateItem) error


// TransactionStateClear removes all keys for a transaction_opaque_data. Called
// when the catalog observes commit/rollback; implementations should
// also TTL-sweep to handle leaked transaction_opaque_data values.
TransactionStateClear(transactionOpaqueData []byte) error


// ExecutionClear wipes every K/V state row, append-log row and counter under
// the given scope (an execution_id or transaction_opaque_data), across all
// namespaces, in one shot. It does NOT touch the work queue (keyed
// separately). Returns the total number of rows removed and is idempotent.
// Consolidates the per-family Clear/Collect calls a teardown would otherwise
// issue. Mirrors vgi-python's execution_clear.
ExecutionClear(scope []byte) (int, error)


// Close releases any underlying resources (DB handles, HTTP clients).
// Safe to call multiple times.
Close() error
}

Description

FunctionStorage is the cross-process shared-state interface backing distributed VGI execution. One implementation per backend (SQLite, Cloudflare Durable Object, …); selected at worker startup.

Methods

function NewSQLiteStorage

Section titled “function NewSQLiteStorage”
source
func NewSQLiteStorage(opts SQLiteStorageOptions) (FunctionStorage, error)

NewSQLiteStorage opens (or creates) a SQLite-backed FunctionStorage. Safe for concurrent use across goroutines and across processes (WAL + busy_timeout): when DuckDB spawns subprocess workers for one execution, every subprocess opens the same database file and sees the others’ rows.

struct SQLiteStorageOptions

Section titled “struct SQLiteStorageOptions”
source
type SQLiteStorageOptions struct {
// Path is the SQLite database file path. Empty defaults to a per-user,
// per-machine state path. Use ":memory:" for the in-process tier.
Path string
}

Description

SQLiteStorageOptions tunes a SQLite-backed FunctionStorage.

func type ScanOption

Section titled “func type ScanOption”
source
type ScanOption func(*AttachScanOptions)

Description

ScanOption configures an AttachStore.Scan. See WithRange / WithStart / WithEnd / WithReverse / WithLimit.

Methods

function WithEnd

Section titled “function WithEnd”
source
func WithEnd(end []byte) ScanOption

WithEnd bounds a scan to keys < end (exclusive).

function WithLimit

Section titled “function WithLimit”
source
func WithLimit(n int) ScanOption

WithLimit caps the scan at n rows (n <= 0 means no limit).

function WithRange

Section titled “function WithRange”
source
func WithRange(start, end []byte) ScanOption

WithRange bounds a scan to the half-open key range [start, end) (a nil bound is open on that side).

function WithReverse

Section titled “function WithReverse”
source
func WithReverse() ScanOption

WithReverse returns the scan in descending key order.

function WithStart

Section titled “function WithStart”
source
func WithStart(start []byte) ScanOption

WithStart bounds a scan to keys >= start (inclusive).

struct ScanWorkerStateEntry

Section titled “struct ScanWorkerStateEntry”
source
type ScanWorkerStateEntry struct {
StreamID  []byte
State    []byte
}

Description

ScanWorkerStateEntry is one (stream_id, state) pair returned by ScanWorkerScan.

interface ShardedBackend

Section titled “interface ShardedBackend”
source
type ShardedBackend interface {
// ForShard returns a FunctionStorage view of the backend pinned to shardKey.
ForShard(shardKey string) FunctionStorage
}

Description

ShardedBackend is implemented by FunctionStorage backends that route remotely on a per-attach shard key (the Cloudflare Durable Object backend). ForShard returns a view of the backend pinned to one shard key; backends that ignore sharding (SQLite) don’t implement it. Lets ExecutionStorage attach the shard key without threading it through all ~22 FunctionStorage methods.

struct StateLogEntry

Section titled “struct StateLogEntry”
source
type StateLogEntry struct {
ID  int64
Value  []byte
}

Description

StateLogEntry is one (id, value) row from an execution-scoped state log.

interface StateLogStorage

Section titled “interface StateLogStorage”
source
type StateLogStorage interface {
// StateAppend appends value to the (executionID, key) log; returns the new
// monotonic log id.
StateAppend(executionID, key, value []byte) (int64, error)
// StateLogScan returns entries with id > afterID (use -1 from the start),
// ordered by id. limit <= 0 means no limit.
StateLogScan(executionID, key []byte, afterID int64, limit int) ([]StateLogEntry, error)
// StateLogClear removes all log rows for an execution_id.
StateLogClear(executionID []byte) error
}

Description

StateLogStorage is an optional capability for an execution-scoped, keyed, append-only log with a monotonic cursor. Table-buffering functions use it to stash batches between the sink (process) and source (finalize) phases across worker processes. Implemented by the SQLite backend; backends that don’t implement it cause buffering functions to error at runtime. Mirrors vgi-python’s BoundStorage.state_append / state_log_scan.

struct TransactionStateItem

Section titled “struct TransactionStateItem”
source
type TransactionStateItem struct {
Key  []byte
Value  []byte
}

Description

TransactionStateItem is one (key, value) pair for transaction-scoped K/V.

struct WorkerStateEntry

Section titled “struct WorkerStateEntry”
source
type WorkerStateEntry struct {
WorkerID  int64
State    []byte
}

Description

WorkerStateEntry is one (worker_id, state) pair returned by WorkerScan.

struct sqliteStorage

Section titled “struct sqliteStorage”
source
type sqliteStorage struct {
db *sql.DB
}

Description

sqliteStorage implements FunctionStorage against a single SQLite database. Concurrency is handled entirely by database/sql + SQLite WAL:

  • Within-process: MaxOpenConns(1) serializes operations through one
connection. database/sql queues callers transparently.
  • Cross-process: WAL mode + busy_timeout=30000 lets multiple worker
subprocesses share the file.

Methods

method AggregateConstArgsGet

Section titled “method AggregateConstArgsGet”
source
func (s *sqliteStorage) AggregateConstArgsGet(executionID []byte, functionName string) ([]byte, error)

AggregateConstArgsGet returns the constant arguments stored for an aggregate function, or nil if none are stored.

method AggregateConstArgsPut

Section titled “method AggregateConstArgsPut”
source
func (s *sqliteStorage) AggregateConstArgsPut(executionID []byte, functionName string, args []byte) error

AggregateConstArgsPut stores the constant arguments for an aggregate function, keyed by function name.

method AggregateStateClear

Section titled “method AggregateStateClear”
source
func (s *sqliteStorage) AggregateStateClear(executionID []byte) error

AggregateStateClear removes all aggregate state for the execution.

method AggregateStateGet

Section titled “method AggregateStateGet”
source
func (s *sqliteStorage) AggregateStateGet(executionID []byte, groupIDs []int64) ([]AggregateStateEntry, error)

AggregateStateGet returns the aggregate state for each requested group ID, omitting groups with no stored state.

method AggregateStatePut

Section titled “method AggregateStatePut”
source
func (s *sqliteStorage) AggregateStatePut(executionID []byte, entries []AggregateStateEntry) error

AggregateStatePut stores the aggregate state for each entry, keyed by group ID, in a single transaction.

method AggregateWindowPartitionClear

Section titled “method AggregateWindowPartitionClear”
source
func (s *sqliteStorage) AggregateWindowPartitionClear(executionID []byte) error

AggregateWindowPartitionClear removes all window-partition data for the execution.

method AggregateWindowPartitionDelete

Section titled “method AggregateWindowPartitionDelete”
source
func (s *sqliteStorage) AggregateWindowPartitionDelete(executionID []byte, partitionID int64) error

AggregateWindowPartitionDelete removes the window-partition data for the given partition ID.

method AggregateWindowPartitionGet

Section titled “method AggregateWindowPartitionGet”
source
func (s *sqliteStorage) AggregateWindowPartitionGet(executionID []byte, partitionID int64) ([]byte, error)

AggregateWindowPartitionGet returns the window-partition data for the given partition ID, or nil if none is stored.

method AggregateWindowPartitionPut

Section titled “method AggregateWindowPartitionPut”
source
func (s *sqliteStorage) AggregateWindowPartitionPut(executionID []byte, partitionID int64, data []byte) error

AggregateWindowPartitionPut stores window-partition data, keyed by partition ID.

method AttachCounterAdd

Section titled “method AttachCounterAdd”
source
func (s *sqliteStorage) AttachCounterAdd(scope, ns, key []byte, delta int64) (int64, error)

AttachCounterAdd atomically adds delta to the counter under (scope, ns, key), initializing an absent counter to 0, and returns the new value.

method AttachCounterDelete

Section titled “method AttachCounterDelete”
source
func (s *sqliteStorage) AttachCounterDelete(scope, ns, key []byte) error

AttachCounterDelete removes the counter under (scope, ns, key). No-op if absent.

method AttachCounterGet

Section titled “method AttachCounterGet”
source
func (s *sqliteStorage) AttachCounterGet(scope, ns, key []byte) (int64, error)

AttachCounterGet returns the int64 counter under (scope, ns, key), or 0 if absent.

method AttachCounterSet

Section titled “method AttachCounterSet”
source
func (s *sqliteStorage) AttachCounterSet(scope, ns, key []byte, value int64) error

AttachCounterSet overwrites the counter under (scope, ns, key) with value.

method AttachStateDeleteKey

Section titled “method AttachStateDeleteKey”
source
func (s *sqliteStorage) AttachStateDeleteKey(scope, ns, key []byte) error

AttachStateDeleteKey removes a single attach-state entry by scope, namespace, and key.

method AttachStateDeleteNS

Section titled “method AttachStateDeleteNS”
source
func (s *sqliteStorage) AttachStateDeleteNS(scope, ns []byte) error

AttachStateDeleteNS removes all attach-state entries under the given scope and namespace.

method AttachStateDeleteRange

Section titled “method AttachStateDeleteRange”
source
func (s *sqliteStorage) AttachStateDeleteRange(scope, ns, start, end []byte) (int, error)

AttachStateDeleteRange removes attach-state entries in the half-open key range [start, end) of (scope, ns) and returns the number removed.

method AttachStateDrain

Section titled “method AttachStateDrain”
source
func (s *sqliteStorage) AttachStateDrain(scope, ns []byte) ([]AttachStateKV, error)

AttachStateDrain atomically reads and removes every attach-state entry under (scope, ns), returning them ordered by key.

method AttachStateGet

Section titled “method AttachStateGet”
source
func (s *sqliteStorage) AttachStateGet(scope, ns, key []byte) ([]byte, error)

AttachStateGet returns the attach-state value for the given scope, namespace, and key, or nil if none is stored.

method AttachStatePut

Section titled “method AttachStatePut”
source
func (s *sqliteStorage) AttachStatePut(scope, ns, key, value []byte) error

AttachStatePut stores a value in attach state under the given scope, namespace, and key.

method AttachStateScan

Section titled “method AttachStateScan”
source
func (s *sqliteStorage) AttachStateScan(scope, ns []byte, opts AttachScanOptions) ([]AttachStateKV, error)

AttachStateScan returns the attach-state key/value pairs under the given scope and namespace, ordered by key and bounded per opts.

method Close

Section titled “method Close”
source
func (s *sqliteStorage) Close() error

Close closes the underlying SQLite database. Safe to call multiple times.

method ExecutionClear

Section titled “method ExecutionClear”
source
func (s *sqliteStorage) ExecutionClear(scope []byte) (int, error)

ExecutionClear wipes function_state, function_state_log and function_counter rows for the scope across every namespace in one transaction, returning the total rows removed. Does not touch work_queue. Idempotent.

method QueueClear

Section titled “method QueueClear”
source
func (s *sqliteStorage) QueueClear(executionID []byte) (int, error)

QueueClear removes all items from the execution’s work queue and returns the number removed.

method QueuePop

Section titled “method QueuePop”
source
func (s *sqliteStorage) QueuePop(executionID []byte) ([]byte, error)

QueuePop atomically removes and returns the next item from the execution’s work queue, or nil if the queue is empty.

method QueuePush

Section titled “method QueuePush”
source
func (s *sqliteStorage) QueuePush(executionID []byte, items [][]byte) (int, error)

QueuePush appends the given items to the execution’s work queue and returns the number pushed.

method ScanWorkerPut

Section titled “method ScanWorkerPut”
source
func (s *sqliteStorage) ScanWorkerPut(executionID, streamID, state []byte) error

ScanWorkerPut stores a scan worker’s state under the execution’s scan-worker namespace, keyed by stream ID.

method ScanWorkerScan

Section titled “method ScanWorkerScan”
source
func (s *sqliteStorage) ScanWorkerScan(executionID []byte) ([]ScanWorkerStateEntry, error)

ScanWorkerScan returns all scan-worker states for the execution without removing them, ordered by stream ID.

method StateAppend

Section titled “method StateAppend”
source
func (s *sqliteStorage) StateAppend(executionID, key, value []byte) (int64, error)

StateAppend appends a value to the (executionID, key) log and returns the new monotonic log id.

method StateLogClear

Section titled “method StateLogClear”
source
func (s *sqliteStorage) StateLogClear(executionID []byte) error

StateLogClear removes all state-log rows for an execution_id.

method StateLogScan

Section titled “method StateLogScan”
source
func (s *sqliteStorage) StateLogScan(executionID, key []byte, afterID int64, limit int) ([]StateLogEntry, error)

StateLogScan returns log entries for (executionID, key) with id > afterID, ordered by id. afterID = -1 reads from the start. limit <= 0 means no limit.

method TransactionStateClear

Section titled “method TransactionStateClear”
source
func (s *sqliteStorage) TransactionStateClear(transactionOpaqueData []byte) error

TransactionStateClear removes all transaction state for the given scope.

method TransactionStateGet

Section titled “method TransactionStateGet”
source
func (s *sqliteStorage) TransactionStateGet(transactionOpaqueData []byte, keys [][]byte) ([][]byte, error)

TransactionStateGet returns the transaction-state value for each requested key, with nil entries for keys that have no stored value.

method TransactionStatePut

Section titled “method TransactionStatePut”
source
func (s *sqliteStorage) TransactionStatePut(transactionOpaqueData []byte, items []TransactionStateItem) error

TransactionStatePut stores each key/value item under the transaction scope in a single transaction.

method WorkerCollect

Section titled “method WorkerCollect”
source
func (s *sqliteStorage) WorkerCollect(executionID []byte) ([][]byte, error)

WorkerCollect drains and returns all worker states for the execution, removing them from the store.

method WorkerPut

Section titled “method WorkerPut”
source
func (s *sqliteStorage) WorkerPut(executionID []byte, workerID int64, state []byte) error

WorkerPut stores a worker’s state under the execution’s worker namespace, keyed by worker ID.

method WorkerScan

Section titled “method WorkerScan”
source
func (s *sqliteStorage) WorkerScan(executionID []byte) ([]WorkerStateEntry, error)

WorkerScan returns all worker states for the execution without removing them, ordered by worker ID.

method stateDeleteKey

Section titled “method stateDeleteKey”
source
func (s *sqliteStorage) stateDeleteKey(scope, ns, key []byte) error

method stateDeleteNS

Section titled “method stateDeleteNS”
source
func (s *sqliteStorage) stateDeleteNS(scope, ns []byte) error

method stateDeleteRange

Section titled “method stateDeleteRange”
source
func (s *sqliteStorage) stateDeleteRange(scope, ns, start, end []byte) (int, error)

stateDeleteRange removes keys in the half-open range [start, end) of (scope, ns) (nil bound = open) and returns the number removed.

method stateDrain

Section titled “method stateDrain”
source
func (s *sqliteStorage) stateDrain(scope, ns []byte) ([][2][]byte, error)

stateDrain reads and deletes every (key, value) in (scope, ns), ordered by key, in one transaction.

method stateGetOne

Section titled “method stateGetOne”
source
func (s *sqliteStorage) stateGetOne(scope, ns, key []byte) ([]byte, error)

method statePut

Section titled “method statePut”
source
func (s *sqliteStorage) statePut(scope, ns, key, value []byte) error

method stateScan

Section titled “method stateScan”
source
func (s *sqliteStorage) stateScan(scope, ns []byte) ([][2][]byte, error)

stateScan returns every (key, value) in (scope, ns), ordered by key.

method stateScanRange

Section titled “method stateScanRange”
source
func (s *sqliteStorage) stateScanRange(scope, ns []byte, opts AttachScanOptions) ([][2][]byte, error)

stateScanRange returns the (key, value) pairs in (scope, ns) within the half-open range [opts.Start, opts.End) (nil bound = open), ordered by key (descending when opts.Reverse), capped at opts.Limit (<= 0 = no limit).

function columnExists

Section titled “function columnExists”
source
func columnExists(db *sql.DB, table, col string) bool

function defaultSQLitePath

Section titled “function defaultSQLitePath”
source
func defaultSQLitePath() string

defaultSQLitePath returns a per-user, per-machine stable path for the FunctionStorage SQLite database. Honors XDG_STATE_HOME, falling back to ~/.local/state/vgi/storage.db on Unix or %LOCALAPPDATA%/vgi/storage.db on Windows. The path is created if absent.

function deriveShardKey

Section titled “function deriveShardKey”
source
func deriveShardKey(attachUUID []byte) (string, error)

deriveShardKey returns the Cloudflare-DO routing key for an attach: the 16-byte framework UUID at the head of the unwrapped attach plaintext, as “att-” + hex(uuid). One DO per logical ATTACH — stable across re-seals and globally unique (unlike the random-nonce ciphertext or possibly-non-unique catalog bytes). Mirrors vgi-python’s _derive_shard_key.

The UUID must be exactly 16 bytes: the storage path is always bound to a logical ATTACH, so a missing/short value is a programming error.

function initSQLiteSchema

Section titled “function initSQLiteSchema”
source
func initSQLiteSchema(db *sql.DB) error

initSQLiteSchema creates the unified tables. Idempotent. Drops any legacy split-schema or idempotency-column tables left over from an older on-disk DB (all this state is ephemeral in-progress worker state).

function int64FromKey

Section titled “function int64FromKey”
source
func int64FromKey(b []byte) int64

function int64Key

Section titled “function int64Key”
source
func int64Key(v int64) []byte

int64Key encodes an int64 worker/group/partition id as an 8-byte big-endian state key (matching the DO client).