a11::stores::RedisChunkStore Class Reference

A persistent, multi-process ChunkStore backed by Redis Streams. More...

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

Inheritance diagram for a11::stores::RedisChunkStore:

Public Member Functions
	~RedisChunkStore () 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 a single 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::Task	Initialize ()
	Ensure that the metadata hash exists, without writing chunk data.
a11::Future< RedisChunkStoreMetadata >	GetMetadata ()
	Read all node-level metadata without iterating over stream entries.
std::shared_ptr< redis::Client >	client () const
	Return the shared Redis client used for commands and subscriptions.
const RedisChunkStoreOptions &	options () const
	Return the validated storage policy captured at construction.
const RedisChunkStoreKeys &	keys () const
	Return the sharding-safe key set owned by this node.
	RedisChunkStore (ConstructorToken, std::string node_id, std::shared_ptr< redis::Client > client, RedisChunkStoreOptions options, RedisChunkStoreKeys keys)
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.
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< RedisChunkStore > >	Create (std::string node_id, std::shared_ptr< redis::Client > client, RedisChunkStoreOptions options)
	Create a store with an injected client and explicit storage policy.
static absl::StatusOr< std::shared_ptr< RedisChunkStore > >	Create (std::string node_id, std::shared_ptr< redis::Client > client)
	Create a store with an injected client and environment/default policy.
static absl::StatusOr< std::shared_ptr< RedisChunkStore > >	Create (std::string node_id)
	Create with the process-global, environment-configured Redis client.

Detailed Description

A persistent, multi-process ChunkStore backed by Redis Streams.

Every key for a node contains the same Redis Cluster hash tag, so each Lua operation is valid on both standalone and sharded Redis deployments. The metadata hash exposes node identity, closure, final sequence, counts and cursors without walking the stream. Chunks up to the configured threshold are inline stream fields; larger encoded chunks live in a separate blob hash and the stream records storage=redis plus a reference. The storage-kind field reserves s3 for a future S3-compatible blob implementation.

Writes, closure and tombstoning are validated and committed by one Lua invocation. A getter first performs an optimistic atomic lookup. Only when it must wait does it subscribe and recheck state before waiting on a broadcast generation, eliminating the lookup/notification race without a blocking Redis command or a fiber parked on a foreign lock.

Constructor & Destructor Documentation

~RedisChunkStore()

a11::stores::RedisChunkStore::~RedisChunkStore ( )

overridedefault

RedisChunkStore()

a11::stores::RedisChunkStore::RedisChunkStore ( ConstructorToken  , std::string  node_id, std::shared_ptr< redis::Client >  client, RedisChunkStoreOptions  options, RedisChunkStoreKeys  keys  )

inline

Member Function Documentation

ClearData()

a11::Future< data::NodeFragment > a11::stores::RedisChunkStore::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

seq	The sequence number whose payload should be cleared.

Returns

An awaitable that resolves with the (now payload-cleared) fragment.

Implements a11::stores::ChunkStore.

client()

std::shared_ptr< redis::Client > a11::stores::RedisChunkStore::client ( ) const

inline

Return the shared Redis client used for commands and subscriptions.

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

status

The 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::RedisChunkStore::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

status	The terminal status to record.
return_status_if_already_closed	When 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

status	The terminal status to record.
return_status_if_already_closed	When 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< RedisChunkStore > > a11::stores::RedisChunkStore::Create ( std::string  node_id )

static

Create with the process-global, environment-configured Redis client.

Create() [2/3]

absl::StatusOr< std::shared_ptr< RedisChunkStore > > a11::stores::RedisChunkStore::Create ( std::string  node_id, std::shared_ptr< redis::Client >  client  )

static

Create a store with an injected client and environment/default policy.

Create() [3/3]

absl::StatusOr< std::shared_ptr< RedisChunkStore > > a11::stores::RedisChunkStore::Create ( std::string  node_id, std::shared_ptr< redis::Client >  client, RedisChunkStoreOptions  options  )

static

Create a store with an injected client and explicit storage policy.

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

seq	The 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::RedisChunkStore::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

seq	The sequence number of the fragment to retrieve.
deadline	The 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

seq	The sequence number of the fragment to retrieve.
deadline	The 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_order

The 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::RedisChunkStore::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_order	The zero-based rank in which the fragment arrived in the store.
deadline	The 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_order	The zero-based rank in which the fragment arrived in the store.
deadline	The 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::RedisChunkStore::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::RedisChunkStore::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< RedisChunkStoreMetadata > a11::stores::RedisChunkStore::GetMetadata

(

)

Read all node-level metadata without iterating over stream entries.

GetSeqForArrivalOrder()

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

overridevirtual

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

Parameters

arrival_order

The 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.

Initialize()

a11::Task a11::stores::RedisChunkStore::Initialize

(

)

Ensure that the metadata hash exists, without writing chunk data.

keys()

const RedisChunkStoreKeys & a11::stores::RedisChunkStore::keys ( ) const

inline

Return the sharding-safe key set owned by this node.

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

deadline

The 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::RedisChunkStore::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

deadline	The absolute time after which the wait gives up.
limit	The 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

deadline	The absolute time after which the wait gives up.
limit	The 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 RedisChunkStoreOptions & a11::stores::RedisChunkStore::options ( ) const

inline

Return the validated storage policy captured at construction.

Put()

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

overridevirtual

Append a single fragment to the log.

Parameters

fragment

The fragment to append.

Returns

An awaitable that resolves with the sequence number assigned to the fragment once the write is accepted.

Implements a11::stores::ChunkStore.

PutMany()

a11::Future< std::vector< std::uint32_t > > a11::stores::RedisChunkStore::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

fragments

The 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.

Size()

a11::Future< size_t > a11::stores::RedisChunkStore::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.

---

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

• cpp/a11/stores/redis_chunk_store.h
• cpp/a11/stores/redis_chunk_store.cc