Stores¶
A ChunkStore is the ordered log behind a
node and an extension point for custom storage. Readers and writers are its cursors.
A11 supplies node semantics over established storage infrastructure instead of requiring a separate durable-stream database. Choose the backend by deployment:
LocalChunkStorekeeps short-lived data inside one process;SQLiteChunkStoreuses an embedded SQLite database for one-machine persistence;RedisChunkStorelets independent programs share named streams through an existing Redis deployment.
Persistence and transactions retain the behavior and operational constraints of SQLite or Redis. Redis deployments also retain their existing replication, monitoring, and scaling facilities. A11 adds chunk ordering, reader and writer cursors, node finality, and structured status.
ChunkStore¶
The main store operations are intentionally low-level:
put appends a fragment,
get waits at a sequence position,
and
close_writes_with_status
publishes the terminal state to every reader.
Construct the abstract base. Subclass this in Python to back an agent with a custom asynchronous chunk store; every data method returns an awaitable so callers never block the event loop.
clear_data(seq: SupportsInt) -> Future[NodeFragment]
Erase the payload of the fragment at a sequence number while keeping its slot, and await the resulting fragment.
close_writes_with_status(status: Status, return_status_if_already_closed: bool = False) -> Future[Status]
Seal the store against further writes with a terminal status and await completion. Waiting readers are released. With return_status_if_already_closed, a repeated close returns the status recorded by the first.
Examples:
Publish clean producer completion to all readers:
get(seq: SupportsInt, deadline: Time | None = None) -> Future[NodeFragment]
get_by_arrival_order(arrival_order: SupportsInt, deadline: Time | None = None) -> Future[NodeFragment]
Await the fragment identified by the order in which it arrived rather than its sequence number. The future resolves when the fragment is present or the optional deadline passes.
Await the explicitly marked final sequence, or None if no fragment has declared finality. Finality is independent of write closure: closing the store does not create a final sequence.
Return the store's node identifier. Raises if a Python subclass does not override get_id.
Await the sequence number that corresponds to a given arrival order.
next(deadline: Time | None = None, limit: SupportsInt = 1) -> Future[list[NodeFragment | None]]
Await up to limit of the next available fragments as a stream. This is the primary way an agent reads chunks as they are produced: the future resolves with whatever is ready before the optional deadline, and slots may be None when a fragment is missing. Loop over successive calls to follow a growing store.
put(fragment: NodeFragment) -> Future[int]
put_many(fragments: Sequence[NodeFragment]) -> Future[list[int]]
Append several fragments in one batch and await their assigned sequence numbers. Prefer this over repeated put calls when an agent emits many chunks at once, to reduce round-trips.
module-attribute
¶
ChunkStoreFactory = Callable[[NameString], ChunkStore]
native_chunk_store(store: ChunkStore) -> ChunkStore
The native store behind store, or store itself.
A facade that only forwards -- an unsubclassed
LocalChunkStore -- is
indistinguishable from the store it wraps except in cost, and the cost is
large: a native reader or writer calling such a store re-enters Python for
every fragment, so each read and each write becomes an event-loop turn and
none of the inline fast paths in ChunkStoreReader or ChunkStoreWriter
can fire. Handing native code the store underneath removes the round trip.
A facade only advertises itself when nothing is overriding it, so a subclass with Python behaviour still goes through the trampoline and stays virtual.
LocalChunkStore¶
Bases: ChunkStore
An in-memory store whose state and synchronization live in C++.
The small forwarding layer keeps Python overrides virtual when a subclass is passed back into native readers, writers, nodes, or sessions.
An instance of this class exposes its underlying native store directly to native readers and writers. Subclasses retain the forwarding layer so their Python overrides remain active.
async
¶
get(seq: int, deadline: Time | None = infinite_future()) -> NodeFragment
Wait for and return the fragment at sequence number seq.
Use a ChunkStoreReader for ordinary sequential reads; direct lookup is useful for replay, inspection, and custom retention logic.
async
¶
get_by_arrival_order(arrival_order: int, deadline: Time | None = infinite_future()) -> NodeFragment
Wait for a fragment by its zero-based ingestion order.
Arrival order can differ from sequence order when fragments reach a store out of order.
async
¶
next(deadline: Time | None = infinite_future(), limit: int = 1) -> list[NodeFragment | None]
Read from the store's shared logical-sequence cursor.
The cursor advances through sequence numbers 0, 1, 2, and so on,
waiting at gaps. None is the clean end sentinel, not a placeholder
for a missing fragment. Use get_by_arrival_order for ingestion order;
most agent code should let ChunkStoreReader manage this cursor,
buffering, ordering, and end-of-stream handling.
async
¶
put(fragment: NodeFragment) -> int
Append one fragment and return its accepted sequence number.
async
¶
put_many(fragments: Sequence[NodeFragment]) -> list[int]
Append a batch and return sequence numbers in matching order.
async
¶
clear_data(seq: int) -> NodeFragment
Discard one stored payload while retaining its sequence slot.
This supports pop_chunks readers and retention policies without
changing the ordering metadata seen by other readers.
async
¶
Translate a zero-based ingestion position to its sequence number.
async
¶
Return the logical final sequence, if a final fragment was written.
Finality and closure are independent: closing writes does not create a final sequence, and writing a final fragment does not close the store.
async
¶
Seal writes with status and wake blocked readers.
This records whether production completed or failed, but it does not
mark any fragment as the final data value. Producers that need semantic
finality should write a fragment with continued=False first (most
commonly through AsyncNode.finalize, which does both).
RedisChunkStore¶
RedisChunkStore(id: str, client: RedisClient | None = None, options: RedisChunkStoreOptions | dict[str, Any] | None = None)
Bases: ChunkStore
A persistent, multi-process ChunkStore backed by Redis Streams.
Open one node's persistent stream with an injected Redis client.
Stores for the same id address the same Redis state. Pass a shared
RedisClient in production so many nodes reuse one connection pool;
call initialize when metadata must exist before the first write.
staticmethod
¶
create(id: str, client: RedisClient | None = None, options: RedisChunkStoreOptions | dict[str, Any] | None = None) -> RedisChunkStore
Create a Redis-backed fragment log for one node id.
async
¶
clear_data(seq: int) -> NodeFragment
Tombstone one payload while retaining ordering metadata.
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
¶
get(seq: int, deadline: Time | None = None) -> NodeFragment
Wait for and return a fragment by sequence number.
The deadline bounds both Redis work and the wait for a future fragment.
async
¶
get_by_arrival_order(arrival_order: int, deadline: Time | None = None) -> NodeFragment
Wait for a fragment by its zero-based Redis ingestion order.
async
¶
Return the logical final sequence, if one has been written.
The final marker is independent of Redis write closure. Closing a store does not synthesize it, and a final fragment does not close the store.
async
¶
get_metadata() -> RedisChunkStoreMetadata
Read size, finality, and closure state without scanning fragments.
async
¶
Translate a zero-based ingestion position to its sequence number.
async
¶
Ensure node metadata exists without writing a fragment.
This is useful during provisioning or health checks; ordinary writes initialize the store lazily.
async
¶
next(deadline: Time | None = None, limit: int = 1) -> list[NodeFragment | None]
Read from the persistent shared logical-sequence cursor.
The cursor advances through sequence numbers and waits at gaps;
None marks clean end-of-stream. Use get_by_arrival_order for
ingestion order. Prefer ChunkStoreReader for normal node reads;
it adds buffering, offsets, and final-sequence handling above this
primitive.
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.
RedisChunkStoreOptions(key_prefix: str = 'a11:', inline_data_threshold: SupportsInt | None = 262144)
Key layout and inline-payload policy for RedisChunkStore.
Construct validated Redis chunk-store options.
property
writable
¶
Chunk data larger than this many bytes uses the blob hash.
staticmethod
¶
from_environment() -> RedisChunkStoreOptions
Read the A11_REDIS_CHUNK_STORE_* environment variables.
Node-level Redis state read without iterating over chunk entries.
property
¶
Number of chunks appended over the store lifetime.
The sharding-safe Redis keys owned by one node stream.
Return keys in the stable order used by the Lua state machine.
SQLiteChunkStore¶
Durable storage without a server: fragments live in one SQLite database per
storage root, with payloads above 128 KiB moved into adjacent blob files. It is
also the only backend that accepts NodeRef payloads, storing the reference
target as indexed columns so traversal between nodes is a query.
The SQLite page covers the on-disk layout, durability, and
configuration, and is where
SQLiteChunkStore,
SQLiteChunkStoreFactory,
SQLiteChunkStoreOptions,
and
SQLiteChunkStoreMetadata
are documented.
ChunkStoreReader¶
Create a reader at an offset and use
next for one record or
async iteration to drain its configured range.
ChunkStoreReader(store: ChunkStore, options: ChunkStoreReaderOptions | dict[str, Any] | None = None)
Open a reader over store.
options (a ChunkStoreReaderOptions or plain dict) tunes
ordering, buffering, starting offset, sticky mimetypes, and whether
chunks are popped as they are read.
property
¶
Number of prefetched fragments currently held in the reader's buffer.
property
¶
options: ChunkStoreReaderOptions
The ChunkStoreReaderOptions this reader was created with.
Stop the background read pump. Pending next awaitables are resolved and no further chunks are fetched.
Start the background read pump if it is not already running. Reading normally starts it lazily; call this to begin buffering before the first next.
get_status() -> Status
Return the reader's current status. An agent can inspect this to distinguish a healthy stream from one that has failed or ended.
next(timeout: Duration = ...)
Return an awaitable for the next fragment in this reader's view.
It resolves to None after the configured range or final sequence is
exhausted. timeout bounds this wait only; a timed-out read does not
close the store or prevent a later call from continuing the stream.
Raises:
| Type | Description |
|---|---|
StatusException
|
If the timeout expires, the store closes with an error, or the reader encounters invalid stream state. |
Examples:
Resume a replay after an application checkpoint:
Await completion of the background read pump. The returned future resolves once the reader has drained the store or been cancelled.
ChunkStoreReaderOptions(ordered: bool = True, pop_chunks: bool = False, num_chunks_to_buffer: SupportsInt | None = 32, offset: SupportsInt | None = 0, max_chunks_to_read: SupportsInt | None = None, sticky_mimetype: bool = False)
Construct validated options for a ChunkStoreReader.
property
writable
¶
Optional cap on the total number of chunks to read.
property
writable
¶
Maximum number of chunks to prefetch into the buffer.
property
writable
¶
Whether chunks are removed from the store as they are read.
property
writable
¶
Whether ordered chunks inherit the last explicitly set mimetype.
ChunkStoreWriter¶
put_chunk returns
a confirmation future after bounded admission. Await that future when the
application must checkpoint only after storage accepts the chunk, then call
drain_and_close
at producer shutdown. This is the storage-level cursor; application code reaches
it through a node, where finalize
marks the logical end of the data and closes in one call. Closing also tees a
closure marker to every attached stream, so a peer mirroring this node closes
its own write half — see
the node lifecycle.
ChunkStoreWriter(chunk_store: ChunkStore, options: ChunkStoreWriterOptions | dict[str, Any] | None = None)
Open a writer over chunk_store.
options (a ChunkStoreWriterOptions or plain dict) tunes the
starting offset, sticky mimetypes, and how much is buffered/flushed at
once.
property
¶
options: ChunkStoreWriterOptions
The ChunkStoreWriterOptions this writer was created with.
property
¶
Number of chunks currently waiting in the writer's flush queue.
abort_with_status(status: Status) -> Future[None]
Abort the writer with an error status and await teardown. Readers then observe the error rather than a clean end-of-stream.
attach_stream(stream: WireStream) -> None
Tee stored fragments to an additional wire stream. After the store accepts a batch, the writer calls send on attached streams; a successful send confirms local transport admission, not peer delivery. A transport failure stops later writes but cannot revoke the current batch's store confirmations. The writer keeps the stream alive while attached.
Stop the writer immediately and await teardown, discarding any chunks still queued.
detach_stream(stream: WireStream) -> None
Stop mirroring fragments to a previously attached wire stream. Raises if the stream was not attached.
Flush every queued chunk, close the writer, and await completion. This does not append a final fragment: mark the last chunk final before draining when readers need a final sequence number.
enqueue_chunk(chunk: Chunk, seq: SupportsInt | None = None, final: bool = False) -> tuple
Enqueue a chunk and get back a (confirmation, admission) pair of awaitables. Unlike put_chunk, this exposes backpressure explicitly: admission resolves when the chunk is accepted into the bounded queue (None if it fit immediately) and confirmation resolves with the sequence assigned by the backing store. An agent awaits admission to pace production and confirmation to know the store accepted the write.
Start the background flush loop if it is not already running. Writing normally starts it lazily; call this to begin flushing before the first chunk is enqueued.
Run the flush loop now, on this thread, if it is idle. A store that accepts the batch without waiting -- an in-memory one does -- has confirmed those writes by the time this returns, so their confirmations resolve without an event-loop turn. A store that cannot leaves the batch in flight and this does nothing, so chunks enqueued behind it still go out together. Awaiting a confirmation calls this for you; call it directly only to push a queue out without awaiting anything.
get_abort_status() -> Status | None
Return the status the writer was aborted with, or None if it was not aborted. Use this to distinguish a clean close from an error-driven abort.
get_status() -> Status | None
Return the writer's terminal status, or None while it is still open. An agent can poll this to detect that the stream has closed or failed.
Return whether the writer still accepts chunks. False once the stream has been drained, closed, or aborted.
async
¶
async
¶
put_chunk(chunk: Chunk, seq: int | None = None, final: bool = False) -> Future[int]
Enqueue a native chunk and return its store-confirmation future.
Set final=True on the last chunk when readers must know the logical
end of the sequence. Calling drain_and_close later only flushes and
closes the writer; it does not add that final marker for you. A node
does both at once -- see AsyncNode.finalize.
Examples:
Checkpoint only after the store accepts the final event:
confirmation = await writer.put_chunk(
a11.to_chunk(event), final=True
)
stored_seq = await confirmation
await checkpoints.save(stored_seq)
Awaiting the confirmation flushes the writer on this thread, so a store that accepts the write without waiting resolves it without an event-loop turn. Awaiting only this coroutine does not flush: the write goes out on the writer's own pump, which is what lets a producer run ahead of a slow store.
Await until the in-flight write buffer empties. An agent can use this as a backpressure checkpoint before enqueuing more chunks.
ChunkStoreWriterOptions(offset: SupportsInt | None = 0, max_chunks_to_write_at_once: SupportsInt | None = 8, num_chunks_to_buffer: SupportsInt | None = None, sticky_mimetype: bool = False)
Construct validated options for a ChunkStoreWriter.