Persist state across workers

How functions that span multiple worker processes — notably distributed aggregates and buffering functions — share and persist state across them.

Prerequisites

You’ve built an aggregate or multi-worker function (see Function patterns → Aggregate).
For the Azure backend: pip install "vgi-python[azure]". SQLite and Cloudflare Durable Objects need no extra.

`vgi-serve`

The commands below use vgi-serve, the CLI installed with vgi-python that runs a worker module as a long-lived process (the production counterpart to the tutorial’s uv run). The --http flag serves it over HTTP instead of stdin/stdout.

What shared state is for

A function normally keeps its state in memory for the life of a single call. Shared storage exists for the cases where that isn’t enough — state that has to be reached from another process or another phase:

Combining partial results — when a function runs across parallel workers, each worker writes its partial output to shared storage and the primary reads them all back to produce the final answer (a distributed aggregate, or a buffering function’s combine).
Buffering across phases — a buffering function stashes every input batch in its process (sink) phase and reads them back in finalize (source); those phases can run in different processes, so the batches can’t live in memory.
Coordinating work across processes — handing data or work items between separate worker processes (for example, a shared work queue).
Outliving a single call — any state that must survive beyond one RPC call.

Because that state crosses process boundaries it is serialized to bytes and kept in a pluggable store — by default a local SQLite database, with cloud backends for when workers run on separate hosts.

Two kinds of “state” — don’t confuse them

Generator cursor state — the small state a table generator keeps within one scan (see streaming with state). It lives in the worker for the duration of the call.
Shared storage (this page) — state that must outlive a single call or be shared across separate worker processes, e.g. combining partial aggregate results. This is backed by a pluggable store.

State doesn’t have to be Arrow

Whichever kind of state you’re holding, the framework needs exactly two things of it: turn the state into bytes, and turn those bytes back into state. That requirement is a structural protocol, StreamStateCodec — not a base class — for table generators and table-in-out functions since 0.24.0, and for aggregates too as of the next release.

ArrowSerializableDataclass satisfies it and remains the default: declare a dataclass and the encoding is written for you. Implement the two methods yourself when you want to own the bytes:

import struct
from dataclasses import dataclass

@dataclass(slots=True)
class CountdownState:
  remaining: int
  emitted: int = 0

  def serialize_to_bytes(self) -> bytes:
      return struct.pack("<qq", self.remaining, self.emitted)

  @classmethod
  def deserialize_from_bytes(cls, data: bytes) -> "CountdownState":
      return cls(*struct.unpack("<qq", data))

Two reasons this matters. Small states pay heavily for Arrow IPC framing: a one-row Arrow stream pays for a schema message, a batch message and an end-of-stream marker no matter the payload, which measured 416 bytes and 36µs for a two-integer state against 16 bytes and 0.21µs packed directly. That is per serialization — and an aggregate serializes once per group, per batch, so the framing cost scales with cardinality. And a Python worker may need to match the state encoding a sibling VGI implementation in another language already uses. Neither was expressible before.

A codec must round-trip exactly

T.deserialize_from_bytes(s.serialize_to_bytes()) must equal s for every state your function can produce, including the initial one. The framework never inspects the bytes, so it cannot check this — a lossy codec surfaces as a stream that silently restarts or skips rows, not as an error.

A plain dataclass that implements neither still raises at class-definition time, as before.

The `BoundStorage` interface

Inside a function you reach shared storage through params.storage, a BoundStorage handle scoped to the current call (its execution_id) and sharded per attached catalog. It offers a few storage primitives — pick whichever fits how your data is shaped:

Key–value store — state_put / state_get (plus _many batch variants), state_scan, state_drain, state_delete. The general-purpose option: stash each worker’s partial result under a key, then scan them all back in finalize.
Append-log — state_append adds an item under a key and returns its index; state_log_scan reads the log back in order. Good for accumulating an ordered stream of partial outputs.
Counters — counter_add / counter_get / counter_set / counter_delete: atomic integer counters shared across workers (row counts, running totals).
Work queue — queue_push / queue_pop (plus batch variants): hand work items between worker processes.
Transaction view — storage.transaction(...) returns a view whose state stays stable across the statements of one SQL transaction (e.g. snapshot-isolation watermarks).

Values are bytes, so serialize first — serialize_record_batch / deserialize_record_batch helpers are provided for Arrow payloads. The full method list is in the State storage API reference.

The default: SQLite (zero config)

Under the subprocess transport, shared storage “just works” — all workers share a local SQLite database (WAL mode) at the platform state directory. Nothing to configure:

vgi-serve my_worker.py

Choosing a backend

Select a backend with the VGI_WORKER_SHARED_STORAGE environment variable:

# Local / subprocess (default)
VGI_WORKER_SHARED_STORAGE=sqlite vgi-serve my_worker.py

# Azure cloud (requires vgi-python[azure])
VGI_WORKER_SHARED_STORAGE=azure-sql vgi-serve my_worker.py --http

# Cloudflare Durable Objects (edge / multi-cloud)
VGI_WORKER_SHARED_STORAGE=cloudflare-do vgi-serve my_worker.py --http

Backend	Value	Use case	Dependencies
SQLite	`sqlite` (default)	local / subprocess	none (stdlib)
Azure SQL	`azure-sql`	Azure deployments	`vgi-python[azure]`
Cloudflare Durable Objects	`cloudflare-do`	edge / multi-cloud	none extra — uses `httpx`, which ships with `vgi-python`; needs a Worker endpoint + token

The per-backend setup (connection strings, credentials, table provisioning) is documented in the Shared Storage reference.

Next steps

Aggregates that use this → Function patterns → Aggregate.
Full backend setup → Shared Storage reference.
Exact API → API Reference: State storage.