A11 (C++ runtime)
Native C++ implementation of the A11 streaming action runtime
Loading...
Searching...
No Matches
a11::stores::SQLiteChunkStore Class Referencefinal

A persistent ChunkStore backed by one SQLite database and a blob directory. More...

#include <cpp/a11/stores/sqlite_chunk_store.h>

Inheritance diagram for a11::stores::SQLiteChunkStore:
[legend]

Public Member Functions

 ~SQLiteChunkStore () override=default
 
a11::Future< data::NodeFragment > Get (std::uint32_t seq, absl::Time deadline) override
 Get the fragment stored at a sequence number.
 
a11::Future< data::NodeFragment > GetByArrivalOrder (std::uint64_t arrival_order, absl::Time deadline) override
 Get the fragment identified by the order in which it arrived, rather than by its sequence number.
 
a11::Future< std::vector< std::optional< data::NodeFragment > > > Next (absl::Time deadline, size_t limit) override
 Get up to limit fragments from the shared logical-sequence cursor.
 
a11::Future< std::uint32_t > Put (data::NodeFragment fragment) override
 Append one fragment to the log.
 
a11::Future< std::vector< std::uint32_t > > PutMany (std::vector< data::NodeFragment > fragments) override
 Append several fragments in one batch.
 
a11::Future< data::NodeFragment > ClearData (std::uint32_t seq) override
 Erase the payload of the fragment at a sequence number while keeping its slot.
 
a11::Future< std::uint32_t > GetSeqForArrivalOrder (std::uint64_t arrival_order) override
 Translate an arrival order into the sequence number of that fragment.
 
a11::Future< std::optional< std::uint32_t > > GetFinalSeq () override
 Get the sequence number explicitly marked as the final fragment.
 
a11::Future< absl::Status > CloseWritesWithStatus (absl::Status status, bool return_status_if_already_closed) override
 Seal the store against further writes with a terminal status.
 
a11::Future< size_t > Size () override
 Get the number of fragments currently in the store.
 
absl::StatusOr< std::string > GetId () const override
 Get the store's node identifier.
 
a11::Future< SQLiteChunkStoreMetadata > GetMetadata ()
 Read all node-level state in one row read, without listing fragments.
 
a11::Future< std::vector< data::NodeFragment > > FindReferrers (size_t limit)
 Find fragments elsewhere in the database whose NodeRef points at this node, newest sequence last.
 
a11::Future< size_t > SweepOrphanBlobs ()
 Delete unreferenced blob files older than the configured grace period.
 
const SQLiteChunkStoreOptions & options () const
 The validated storage policy captured at construction.
 
std::string root () const
 The storage root this store reads and writes under.
 
 SQLiteChunkStore (ConstructorToken, std::string node_id, std::shared_ptr< internal::SqliteDatabase > database, SQLiteChunkStoreOptions options)
 
a11::Future< absl::Status > CloseWritesWithStatus (absl::Status status)
 Seal the store against further writes with a terminal status.
 
virtual a11::Future< absl::Status > CloseWritesWithStatus (absl::Status status, bool return_status_if_already_closed)=0
 Seal the store against further writes with a terminal status.
 
a11::Future< data::NodeFragment > Get (std::uint32_t seq)
 Get the fragment at a sequence number, waiting indefinitely.
 
virtual a11::Future< data::NodeFragment > Get (std::uint32_t seq, absl::Time deadline)=0
 Get the fragment stored at a sequence number.
 
a11::Future< data::NodeFragment > GetByArrivalOrder (std::uint64_t arrival_order)
 Get a fragment by arrival order, waiting indefinitely.
 
virtual a11::Future< data::NodeFragment > GetByArrivalOrder (std::uint64_t arrival_order, absl::Time deadline)=0
 Get the fragment identified by the order in which it arrived, rather than by its sequence number.
 
a11::Future< std::vector< std::optional< data::NodeFragment > > > Next ()
 Get the next logical-sequence fragment, waiting indefinitely.
 
a11::Future< std::vector< std::optional< data::NodeFragment > > > Next (absl::Time deadline)
 Get the next logical-sequence fragment before a deadline.
 
virtual a11::Future< std::vector< std::optional< data::NodeFragment > > > Next (absl::Time deadline, size_t limit)=0
 Get up to limit fragments from the shared logical-sequence cursor.
 
- Public Member Functions inherited from a11::stores::ChunkStore
virtual ~ChunkStore ()=default
 
a11::Future< data::NodeFragment > Get (std::uint32_t seq)
 Get the fragment at a sequence number, waiting indefinitely.
 
a11::Future< data::NodeFragment > GetByArrivalOrder (std::uint64_t arrival_order)
 Get a fragment by arrival order, waiting indefinitely.
 
a11::Future< std::vector< std::optional< data::NodeFragment > > > Next ()
 Get the next logical-sequence fragment, waiting indefinitely.
 
a11::Future< std::vector< std::optional< data::NodeFragment > > > Next (absl::Time deadline)
 Get the next logical-sequence fragment before a deadline.
 
virtual bool HoldsObjects () const
 Whether this store can hold a chunk that carries a value.
 
a11::Future< absl::Status > CloseWritesWithStatus (absl::Status status)
 Seal the store against further writes with a terminal status.
 

Static Public Member Functions

static absl::StatusOr< std::shared_ptr< SQLiteChunkStore > > Create (std::string node_id)
 Create a store for node_id under the process-default root.
 
static absl::StatusOr< std::shared_ptr< SQLiteChunkStore > > Create (std::string node_id, std::string root)
 Create a store for node_id under an explicit root.
 
static absl::StatusOr< std::shared_ptr< SQLiteChunkStore > > Create (std::string node_id, const std::string &root, SQLiteChunkStoreOptions options)
 Create a store for node_id with an explicit root and policy.
 

Friends

class SQLiteChunkStoreFactory
 

Detailed Description

A persistent ChunkStore backed by one SQLite database and a blob directory.

The layout under the storage root is:

./store.sqlite
./blobs/939f2184-db19-4dd0-b949-bb31c5eadcf8
./blobs/7ee4a05e-f439-4e5f-bb97-8d1388960f29

Node rows carry the shared producer and consumer cursors, closure state and cached counters, so Size() and GetFinalSeq() are single-row reads. Each fragment is decomposed into columns rather than stored as an opaque blob, so ordinary SQL can filter by owner, timestamp or node reference. Payloads above the inline threshold live in blobs/ under a UUID name recorded in the row.

Unlike LocalChunkStore and RedisChunkStore, this backend accepts data::NodeRef payloads: their target, offset and length become indexed columns, which is what makes traversal between nodes a query rather than an application-level walk.

Every mutation is one BEGIN IMMEDIATE transaction, so a batch either lands whole or not at all. Readers never poll: they snapshot a per-node generation event, run an optimistic read, and park on that event when the fragment they want has not arrived, which a committing writer then fires.

Constructor & Destructor Documentation

◆ ~SQLiteChunkStore()

a11::stores::SQLiteChunkStore::~SQLiteChunkStore ( )
overridedefault

◆ SQLiteChunkStore()

a11::stores::SQLiteChunkStore::SQLiteChunkStore ( ConstructorToken  ,
std::string  node_id,
std::shared_ptr< internal::SqliteDatabase >  database,
SQLiteChunkStoreOptions  options 
)
inline

Member Function Documentation

◆ ClearData()

a11::Future< data::NodeFragment > a11::stores::SQLiteChunkStore::ClearData ( std::uint32_t  seq)
overridevirtual

Erase the payload of the fragment at a sequence number while keeping its slot.

Used to reclaim memory for fragments that have already been consumed without disturbing the sequence numbering.

Parameters
seqThe sequence number whose payload should be cleared.
Returns
An awaitable that resolves with the (now payload-cleared) fragment.

Implements a11::stores::ChunkStore.

◆ CloseWritesWithStatus() [1/3]

a11::Future< absl::Status > a11::stores::ChunkStore::CloseWritesWithStatus ( absl::Status  status)
inline

Seal the store against further writes with a terminal status.

Convenience overload that does not return the previously recorded status if the store is already closed.

Parameters
statusThe terminal status to record; readers awaiting Next() are released.
Returns
An awaitable that resolves when the close completes.

◆ CloseWritesWithStatus() [2/3]

a11::Future< absl::Status > a11::stores::SQLiteChunkStore::CloseWritesWithStatus ( absl::Status  status,
bool  return_status_if_already_closed 
)
overridevirtual

Seal the store against further writes with a terminal status.

Once closed, readers blocked on Next() are released. Call this when a producer has finished (or failed).

Parameters
statusThe terminal status to record.
return_status_if_already_closedWhen true, a second close resolves with the status recorded by the first close instead of overwriting it.
Returns
An awaitable that resolves when the close completes.

Implements a11::stores::ChunkStore.

◆ CloseWritesWithStatus() [3/3]

virtual a11::Future< absl::Status > a11::stores::ChunkStore::CloseWritesWithStatus ( absl::Status  status,
bool  return_status_if_already_closed 
)
virtual

Seal the store against further writes with a terminal status.

Once closed, readers blocked on Next() are released. Call this when a producer has finished (or failed).

Parameters
statusThe terminal status to record.
return_status_if_already_closedWhen true, a second close resolves with the status recorded by the first close instead of overwriting it.
Returns
An awaitable that resolves when the close completes.

Implements a11::stores::ChunkStore.

◆ Create() [1/3]

absl::StatusOr< std::shared_ptr< SQLiteChunkStore > > a11::stores::SQLiteChunkStore::Create ( std::string  node_id)
static

Create a store for node_id under the process-default root.

Parameters
node_idThe node whose fragment log this store backs.
Returns
The store, or an error if the id or storage root is unusable.

◆ Create() [2/3]

absl::StatusOr< std::shared_ptr< SQLiteChunkStore > > a11::stores::SQLiteChunkStore::Create ( std::string  node_id,
const std::string &  root,
SQLiteChunkStoreOptions  options 
)
static

Create a store for node_id with an explicit root and policy.

Parameters
node_idThe node whose fragment log this store backs.
rootDirectory holding store.sqlite and blobs/; created when absent.
optionsStorage policy; per-root settings apply only on first open of that root.
Returns
The store, or an error if the id, root or options are unusable.

◆ Create() [3/3]

absl::StatusOr< std::shared_ptr< SQLiteChunkStore > > a11::stores::SQLiteChunkStore::Create ( std::string  node_id,
std::string  root 
)
static

Create a store for node_id under an explicit root.

Parameters
node_idThe node whose fragment log this store backs.
rootDirectory holding store.sqlite and blobs/; created when absent.
Returns
The store, or an error if the id or storage root is unusable.

◆ FindReferrers()

a11::Future< std::vector< data::NodeFragment > > a11::stores::SQLiteChunkStore::FindReferrers ( size_t  limit)

Find fragments elsewhere in the database whose NodeRef points at this node, newest sequence last.

This is the traversal the relational layout exists for: the answer comes from an index on node_ref_id rather than from scanning every node.

Parameters
limitMaximum number of referring fragments to return.
Returns
The referring fragments, resolved and ordered by (node id, seq).

◆ Get() [1/3]

a11::Future< data::NodeFragment > a11::stores::ChunkStore::Get ( std::uint32_t  seq)
inline

Get the fragment at a sequence number, waiting indefinitely.

Convenience overload equivalent to Get() with an infinite deadline.

Parameters
seqThe sequence number of the fragment to retrieve.
Returns
An awaitable that resolves with the requested fragment once it is available.

◆ Get() [2/3]

a11::Future< data::NodeFragment > a11::stores::SQLiteChunkStore::Get ( std::uint32_t  seq,
absl::Time  deadline 
)
overridevirtual

Get the fragment stored at a sequence number.

As storage may be non-local, errors are surfaced through the resolved fragment/status. Correct implementations resolve with an error rather than blocking forever once the fragment can no longer arrive (e.g. writes were closed with a smaller final sequence number).

Parameters
seqThe sequence number of the fragment to retrieve.
deadlineThe absolute time after which the wait gives up.
Returns
An awaitable that resolves with the fragment once available, or with an error if the deadline elapses or the fragment can never arrive.

Implements a11::stores::ChunkStore.

◆ Get() [3/3]

virtual a11::Future< data::NodeFragment > a11::stores::ChunkStore::Get ( std::uint32_t  seq,
absl::Time  deadline 
)
virtual

Get the fragment stored at a sequence number.

As storage may be non-local, errors are surfaced through the resolved fragment/status. Correct implementations resolve with an error rather than blocking forever once the fragment can no longer arrive (e.g. writes were closed with a smaller final sequence number).

Parameters
seqThe sequence number of the fragment to retrieve.
deadlineThe absolute time after which the wait gives up.
Returns
An awaitable that resolves with the fragment once available, or with an error if the deadline elapses or the fragment can never arrive.

Implements a11::stores::ChunkStore.

◆ GetByArrivalOrder() [1/3]

a11::Future< data::NodeFragment > a11::stores::ChunkStore::GetByArrivalOrder ( std::uint64_t  arrival_order)
inline

Get a fragment by arrival order, waiting indefinitely.

Convenience overload equivalent to GetByArrivalOrder() with an infinite deadline.

Parameters
arrival_orderThe zero-based rank in which the fragment arrived in the store.
Returns
An awaitable that resolves with the requested fragment once available.

◆ GetByArrivalOrder() [2/3]

a11::Future< data::NodeFragment > a11::stores::SQLiteChunkStore::GetByArrivalOrder ( std::uint64_t  arrival_order,
absl::Time  deadline 
)
overridevirtual

Get the fragment identified by the order in which it arrived, rather than by its sequence number.

Useful for replaying fragments in ingestion order regardless of the sequence numbers assigned to them.

Parameters
arrival_orderThe zero-based rank in which the fragment arrived in the store.
deadlineThe absolute time after which the wait gives up.
Returns
An awaitable that resolves with the fragment once available, or with an error if the deadline elapses.

Implements a11::stores::ChunkStore.

◆ GetByArrivalOrder() [3/3]

virtual a11::Future< data::NodeFragment > a11::stores::ChunkStore::GetByArrivalOrder ( std::uint64_t  arrival_order,
absl::Time  deadline 
)
virtual

Get the fragment identified by the order in which it arrived, rather than by its sequence number.

Useful for replaying fragments in ingestion order regardless of the sequence numbers assigned to them.

Parameters
arrival_orderThe zero-based rank in which the fragment arrived in the store.
deadlineThe absolute time after which the wait gives up.
Returns
An awaitable that resolves with the fragment once available, or with an error if the deadline elapses.

Implements a11::stores::ChunkStore.

◆ GetFinalSeq()

a11::Future< std::optional< std::uint32_t > > a11::stores::SQLiteChunkStore::GetFinalSeq ( )
overridevirtual

Get the sequence number explicitly marked as the final fragment.

Returns
An awaitable that resolves with the logical final sequence number, or an empty optional when no fragment has declared finality.

Finality and write closure are separate state transitions. A producer may mark a fragment final before closing the store, and CloseWritesWithStatus() does not invent a final fragment.

Implements a11::stores::ChunkStore.

◆ GetId()

absl::StatusOr< std::string > a11::stores::SQLiteChunkStore::GetId ( ) const
overridevirtual

Get the store's node identifier.

Returns
The node id, or an error status if the identifier is unavailable.

Implements a11::stores::ChunkStore.

◆ GetMetadata()

a11::Future< SQLiteChunkStoreMetadata > a11::stores::SQLiteChunkStore::GetMetadata ( )

Read all node-level state in one row read, without listing fragments.

◆ GetSeqForArrivalOrder()

a11::Future< std::uint32_t > a11::stores::SQLiteChunkStore::GetSeqForArrivalOrder ( std::uint64_t  arrival_order)
overridevirtual

Translate an arrival order into the sequence number of that fragment.

Parameters
arrival_orderThe zero-based rank in which the fragment arrived in the store.
Returns
An awaitable that resolves with the corresponding sequence number.

Implements a11::stores::ChunkStore.

◆ Next() [1/4]

a11::Future< std::vector< std::optional< data::NodeFragment > > > a11::stores::ChunkStore::Next ( )
inline

Get the next logical-sequence fragment, waiting indefinitely.

Convenience overload equivalent to Next() with an infinite deadline and a limit of one.

Returns
An awaitable that resolves with at most one fragment, or a nullopt sentinel once the store reaches a clean logical end.

◆ Next() [2/4]

a11::Future< std::vector< std::optional< data::NodeFragment > > > a11::stores::ChunkStore::Next ( absl::Time  deadline)
inline

Get the next logical-sequence fragment before a deadline.

Convenience overload equivalent to Next() with a limit of one.

Parameters
deadlineThe absolute time after which the wait gives up.
Returns
An awaitable that resolves with at most one fragment, or a nullopt sentinel at a clean logical end.

◆ Next() [3/4]

a11::Future< std::vector< std::optional< data::NodeFragment > > > a11::stores::SQLiteChunkStore::Next ( absl::Time  deadline,
size_t  limit 
)
overridevirtual

Get up to limit fragments from the shared logical-sequence cursor.

Next() advances through sequence numbers 0, 1, 2, and so on. It waits at a gap rather than switching to ingestion order; use GetByArrivalOrder() for that view. After the final sequence or a clean write closure, the result includes a nullopt end sentinel. Call repeatedly to follow the store as it grows.

Parameters
deadlineThe absolute time after which the wait gives up.
limitThe maximum number of fragments to return in one call.
Returns
An awaitable that resolves with up to limit data fragments; a clean end may append a nullopt sentinel.

Implements a11::stores::ChunkStore.

◆ Next() [4/4]

virtual a11::Future< std::vector< std::optional< data::NodeFragment > > > a11::stores::ChunkStore::Next ( absl::Time  deadline,
size_t  limit 
)
virtual

Get up to limit fragments from the shared logical-sequence cursor.

Next() advances through sequence numbers 0, 1, 2, and so on. It waits at a gap rather than switching to ingestion order; use GetByArrivalOrder() for that view. After the final sequence or a clean write closure, the result includes a nullopt end sentinel. Call repeatedly to follow the store as it grows.

Parameters
deadlineThe absolute time after which the wait gives up.
limitThe maximum number of fragments to return in one call.
Returns
An awaitable that resolves with up to limit data fragments; a clean end may append a nullopt sentinel.

Implements a11::stores::ChunkStore.

◆ options()

const SQLiteChunkStoreOptions & a11::stores::SQLiteChunkStore::options ( ) const
inline

The validated storage policy captured at construction.

◆ Put()

a11::Future< std::uint32_t > a11::stores::SQLiteChunkStore::Put ( data::NodeFragment  fragment)
overridevirtual

Append one fragment to the log.

Parameters
fragmentFragment to append.
Returns
An awaitable resolving to its assigned sequence number.

Implements a11::stores::ChunkStore.

◆ PutMany()

a11::Future< std::vector< std::uint32_t > > a11::stores::SQLiteChunkStore::PutMany ( std::vector< data::NodeFragment >  fragments)
overridevirtual

Append several fragments in one batch.

Preferred over repeated Put() calls when many fragments are emitted at once, to reduce round-trips.

Parameters
fragmentsThe fragments to append, in order.
Returns
An awaitable that resolves with the sequence numbers assigned to the fragments, in the same order.

Implements a11::stores::ChunkStore.

◆ root()

std::string a11::stores::SQLiteChunkStore::root ( ) const

The storage root this store reads and writes under.

◆ Size()

a11::Future< size_t > a11::stores::SQLiteChunkStore::Size ( )
overridevirtual

Get the number of fragments currently in the store.

Returns
An awaitable that resolves with the current fragment count.

Implements a11::stores::ChunkStore.

◆ SweepOrphanBlobs()

a11::Future< size_t > a11::stores::SQLiteChunkStore::SweepOrphanBlobs ( )

Delete unreferenced blob files older than the configured grace period.

Friends And Related Symbol Documentation

◆ SQLiteChunkStoreFactory

friend class SQLiteChunkStoreFactory
friend

The documentation for this class was generated from the following files: