Stores¶
A ChunkStore is the ordered log behind a
node, and a deliberate
extension point for custom
storage. Readers and writers are its cursors.
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.
Erase the payload of the fragment at a sequence number while keeping its slot, and await the resulting fragment. Use this to reclaim memory for chunks an agent has already consumed.
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_by_arrival_order(arrival_order: SupportsInt | SupportsIndex, deadline: Any | None = None) -> Any
Await the fragment identified by the order in which it arrived rather than its sequence number. Use this when an agent needs to replay chunks in ingestion order; 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. Use this to translate ingestion-order references into the sequence numbers the rest of the API expects.
Await up to limit of the next available fragments as a stream. This is the primary way an agent consumes 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) -> Any
put_many(fragments: Sequence[NodeFragment]) -> Any
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.
Await the number of fragments currently in the store. Useful for an agent to gauge backlog or progress without reading chunks.
module-attribute
¶
ChunkStoreFactory = Callable[[NameString], ChunkStore]
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.
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 consumption; 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.put_final or put_null_final).
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 consumption;
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.
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.
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.
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: Any = 32, offset: Any = 0, max_chunks_to_read: Any | 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.
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 the writer with an error status and await teardown. Use this to propagate a failure downstream so readers observe the error instead of 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. Use this to abandon a stream an agent no longer needs.
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: Any | None = None, final: bool = False) -> tuple[Any, Any]
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.
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.
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.
Examples:
Checkpoint only after the store accepts the final event:
Await until the in-flight write buffer empties. An agent can use this as a backpressure checkpoint before enqueuing more chunks.
ChunkStoreWriterOptions(offset: Any = 0, max_chunks_to_write_at_once: Any = 8, num_chunks_to_buffer: Any | None = None, sticky_mimetype: bool = False)
Construct validated options for a ChunkStoreWriter.