SQLite¶
SQLiteChunkStore is a
durable ChunkStore that needs no server.
It maps node records and lifecycle state onto the established SQLite
transaction and write-ahead-log facilities, with large payloads in adjacent
blob files. A single-machine service gains durable streams without operating a
custom A11 storage server. LocalChunkStore remains the process-local choice;
RedisChunkStore supports programs sharing streams through a Redis deployment.
from a11.stores.sqlite_chunk_store import SQLiteChunkStoreFactory
factory = SQLiteChunkStoreFactory("/var/lib/agent/chunks")
store = factory.open("agent-output")
await store.put_many(fragments)
Pass the factory itself wherever a chunk_store_factory is expected — it is
callable — to make SQLite the backing store for a whole node map or session:
Omitting the root uses
default_root:
$A11_SQLITE_CHUNK_STORE_ROOT when set, otherwise $XDG_CACHE_HOME/a11/chunks,
otherwise ~/.cache/a11/chunks.
Storage layout¶
A root holds one database and one blob directory:
./store.sqlite
./blobs/939f2184-db19-4dd0-b949-bb31c5eadcf8
./blobs/7ee4a05e-f439-4e5f-bb97-8d1388960f29
Every store opened under the same root shares one database, connection set, and worker pool. Use one factory per root to share those resources across nodes.
The nodes table holds a row per represented node: the shared producer and
consumer cursors, closure state and terminal status, the declared final
sequence, owner_id, created_at/updated_at, and cached counters such as
size so
size and
get_final_seq
are single-row reads. A row is created implicitly by the first accepted write.
The fragments table stores fragment fields in separate columns, allowing SQL
filters by owner, timestamp, or reference target. Indexes cover
(node_id, seq) for sequenced reading,
(node_id, arrival_order) for ingestion order, owner_id on the node table,
and partial indexes on node_ref_id and on the blob reference.
A fragment's timestamp column is always populated — from
ChunkMetadata.timestamp when present, otherwise the current UTC time — so it
is a usable index key; a companion flag records whether the original metadata
actually carried one, so round-tripping stays exact.
continued is derived, never stored as truth
The flag returned to callers is recomputed from the node's current final sequence on every read. A later batch can declare finality, and a stored flag would then disagree with the store.
Payloads and blob files¶
Chunk.data at or below inline_data_threshold (128 KiB by default) stays in
the row. Anything larger is written to blobs/ under a UUID name recorded in
the row, which keeps the page cache useful and the write-ahead log small.
Reading is transparent either way.
A blob is written to a temporary name, fsynced, renamed into place, and then
its directory is fsynced — all before the transaction referencing it commits,
so a committed row can never point at a payload that never reached disk.
Removals run in the other order, after the commit, so a rollback cannot take
the data with it. A crash in either window leaves an unreferenced file, which
sweep_orphan_blobs
reclaims; its grace period keeps it from deleting a blob whose transaction is
still in flight.
Node references¶
This is the only backend that accepts NodeRef payloads — LocalChunkStore
and RedisChunkStore reject them as UNIMPLEMENTED. The target, offset, and
length become indexed columns, making referrer lookup a query:
Because a tombstone is chunk-shaped,
clear_data
rejects node-reference fragments to avoid returning a different payload type.
Transactions and waiting¶
Every mutation runs as one BEGIN IMMEDIATE transaction, so a batch either
lands whole or not at all, leaving no partial rows and no stray blob files.
Contention uses bounded retries because sqlite3_busy_timeout would block the
calling thread inside SQLite.
Readers never poll. A getter snapshots a per-node change event, runs an
optimistic read, and parks on that event only if the fragment it wants has not
arrived; a committing writer fires the event once COMMIT has returned. Taking
the snapshot before the read, and firing strictly after the commit, is what
closes the lost-wakeup window in both directions.
SQLite calls run on a small dedicated thread pool. Running them on A11's fiber
workers could block deadline timers while sqlite3_step waits.
One writing process per root, by default
SQLite's change hooks are per-connection, so a writer in another process
cannot wake a reader parked in this one. Multi-process readers would wait
out their deadlines. Set cross_process_poll_interval to enable a
PRAGMA data_version watcher when more than one process writes a root; it
is disabled by default so the common case never polls.
Configuration¶
An explicit SQLiteChunkStoreOptions takes the place of the environment.
Per-root settings — durability, the poll interval, the grace period — apply on
the first open of that root, since the database behind it is shared.
| Variable | Default | Meaning |
|---|---|---|
A11_SQLITE_CHUNK_STORE_ROOT |
unset | Default storage root, overriding the cache-directory convention. |
A11_SQLITE_CHUNK_STORE_INLINE_DATA_THRESHOLD_BYTES |
131072 |
Chunk.data size above which the payload moves to a blob file. |
A11_SQLITE_CHUNK_STORE_OWNER_ID |
empty | Owner recorded on node rows. |
A11_SQLITE_CHUNK_STORE_SYNCHRONOUS |
normal |
off, normal, or full; applied as PRAGMA synchronous. |
A11_SQLITE_CHUNK_STORE_CROSS_PROCESS_POLL_MS |
0 |
Interval for noticing other processes' commits; 0 disables it. |
A11_SQLITE_CHUNK_STORE_BLOB_GRACE_MS |
3600000 |
How long an unreferenced blob survives before a sweep may remove it. |
normal in WAL mode survives an application crash but may lose the newest
commits on power loss. Use full when that matters.
Ownership has no enforcement semantics yet; owner_id exists so nodes can be
attributed and filtered.
SQLite is compiled into the extension from the upstream amalgamation with
hidden visibility, so it cannot collide with the system libsqlite3 that
CPython's own _sqlite3 module loads into the same process. The build option
is A11_BUILD_SQLITE.
SQLiteChunkStore(id: str, root: str | Any | None = None, options: SQLiteChunkStoreOptions | dict[str, Any] | None = None)
Bases: ChunkStore
A durable, embedded ChunkStore backed by SQLite and blob files.
Open one node's durable stream under a storage root.
Stores for the same id and root address the same rows, so reopening
after a restart resumes the same fragment log. Omit root to use
SQLiteChunkStoreFactory.default_root(). Opening many nodes under one
root is cheap; they share a single database.
staticmethod
¶
create(id: str, root: str | Any | None = None, options: SQLiteChunkStoreOptions | dict[str, Any] | None = None) -> SQLiteChunkStore
Create a SQLite-backed fragment log for one node id.
async
¶
clear_data(seq: int) -> NodeFragment
Tombstone one payload while retaining ordering metadata.
Returns the fragment as it was. Any blob file backing it is unlinked once the transaction commits. Node-reference fragments cannot be cleared, because a tombstone is chunk-shaped.
async
¶
Atomically seal writes with a terminal status and wake readers.
This closes the producer side but does not mark data final. Write a
final fragment first when consumers use whole-value semantics such as
AsyncNode.consume.
async
¶
find_referrers(limit: int = 100) -> list[NodeFragment]
Find fragments elsewhere whose NodeRef points at this node.
This is the traversal the relational layout exists for: the answer comes from an index on the reference target, so the cost tracks the number of referrers rather than the size of the database.
async
¶
get(seq: int, deadline: Time | None = None) -> NodeFragment
Wait for and return a fragment by sequence number.
The wait parks on a per-node event rather than polling the database, and resolves early with an error once the fragment can no longer arrive.
async
¶
get_by_arrival_order(arrival_order: int, deadline: Time | None = None) -> NodeFragment
Wait for a fragment by its zero-based ingestion order.
async
¶
Return the logical final sequence, if one has been written.
The final marker is independent of write closure. Closing a store does not synthesize it, and a final fragment does not close the store.
async
¶
get_metadata() -> SQLiteChunkStoreMetadata
Read cursors, finality, and closure state in one row read.
async
¶
Translate a zero-based ingestion position to its sequence number.
async
¶
next(deadline: Time | None = None, limit: int = 1) -> list[NodeFragment | None]
Read from the persistent shared logical-sequence cursor.
The cursor lives in the database, so it survives a restart and is
shared by every store open on this node. It advances through sequence
numbers and waits at gaps; None marks clean end-of-stream. Prefer
ChunkStoreReader for ordinary consumption.
async
¶
put(fragment: NodeFragment) -> int
Atomically append one fragment and return its sequence number.
async
¶
put_many(fragments: Sequence[NodeFragment]) -> list[int]
Atomically append a batch and return its assigned sequences.
The batch commits in one transaction: either every fragment is stored, or none is and no blob file is left behind.
async
¶
Delete unreferenced blob files older than the grace period.
A crash between writing a blob and committing its row leaves a file nothing points at. This reclaims those; the grace period keeps it from deleting a blob whose transaction is still in flight elsewhere.
SQLiteChunkStoreFactory(root: str | Any | None = None, options: SQLiteChunkStoreOptions | dict[str, Any] | None = None)
Creates SQLiteChunkStores that share one database per storage root.
Create a factory rooted at a directory, created when absent.
Pass the factory itself wherever a chunk_store_factory callable is
expected to make SQLite the backing store for a NodeMap, AsyncNode,
or Session.
property
¶
options: SQLiteChunkStoreOptions
A copy of the storage policy applied to every store created here.
staticmethod
¶
The process-wide default storage root: $A11_SQLITE_CHUNK_STORE_ROOT, else $XDG_CACHE_HOME/a11/chunks, else ~/.cache/a11/chunks.
async
¶
Delete unreferenced blob files older than the grace period.
SQLiteChunkStoreOptions(inline_data_threshold: SupportsInt | None = 131072, owner_id: str = '', synchronous: SQLiteSynchronous = ..., cross_process_poll_interval: Any | None = None, blob_grace_period: Any | None = None)
Payload, ownership and durability policy for SQLiteChunkStore.
Construct validated SQLite chunk-store options.
property
writable
¶
blob_grace_period: Duration
How long an unreferenced blob survives before a sweep removes it.
property
writable
¶
cross_process_poll_interval: Duration
How often to notice other processes' commits; zero disables it.
property
writable
¶
Payloads larger than this many bytes move into a blob file.
property
writable
¶
Durability level applied with PRAGMA synchronous.
staticmethod
¶
from_environment() -> SQLiteChunkStoreOptions
Read the A11_SQLITE_CHUNK_STORE_* environment variables.
Node-level SQLite state read without listing fragments.