a11::stores::ChunkStoreWriter Class Reference

A buffered, backpressured write cursor over a ChunkStore. More...

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

Classes
struct	State

Public Member Functions
	~ChunkStoreWriter ()=default
void	EnsureStarted ()
	Start the background flush loop if it is not already running.
void	Flush ()
	Run the flush loop now, on the calling thread, if it is idle.
ChunkStoreWrite	EnqueueChunk (data::Chunk chunk, std::optional< std::uint32_t > seq=std::nullopt, bool final=false, bool ensure_started=true)
	Enqueue a chunk, exposing backpressure and confirmation separately.
a11::Future< std::uint32_t >	PutChunk (data::Chunk chunk, std::optional< std::uint32_t > seq=std::nullopt, bool final=false)
	Write a chunk and await backing-store confirmation.
std::optional< absl::Status >	GetStatus () const
std::optional< absl::Status >	GetAbortStatus () const
bool	IsWritable () const
a11::Task	Cancel ()
	Stop the writer immediately, discarding any queued chunks.
a11::Task	DrainAndClose ()
	Flush every queued chunk, then close the writer.
a11::Task	AbortWithStatus (absl::Status status)
	Abort the writer with an error status.
a11::Task	WaitForBufferToDrain ()
	Wait until the in-flight write buffer empties.
absl::Status	AttachStream (std::shared_ptr< net::WireStream > stream)
	Mirror persisted fragments to an additional wire stream.
absl::Status	DetachStream (const std::shared_ptr< net::WireStream > &stream)
	Stop mirroring fragments to a previously attached wire stream.
std::shared_ptr< ChunkStore >	store () const
ChunkStoreWriterOptions	options () const
size_t	queue_size () const

Static Public Member Functions
static absl::StatusOr< std::shared_ptr< ChunkStoreWriter > >	Create (std::shared_ptr< ChunkStore > store, ChunkStoreWriterOptions options={})
	Create a writer over store.

Detailed Description

A buffered, backpressured write cursor over a ChunkStore.

A writer admits data::Chunk values into a store in sequence, pacing the producer through a bounded buffer sized by its ChunkStoreWriterOptions. A background flush loop drains the queue into the store; persisted fragments can additionally be mirrored to attached wire streams. It is usable directly though most code reaches one through a node. State is reference-counted and held via shared_ptr.

Constructor & Destructor Documentation

~ChunkStoreWriter()

a11::stores::ChunkStoreWriter::~ChunkStoreWriter ( )

default

Member Function Documentation

AbortWithStatus()

a11::Task a11::stores::ChunkStoreWriter::AbortWithStatus

(

absl::Status

status

)

Abort the writer with an error status.

Propagates a failure downstream so readers observe the error instead of a clean end-of-stream.

Parameters

status

The error status to record.

Returns

An awaitable that resolves once teardown completes.

AttachStream()

absl::Status a11::stores::ChunkStoreWriter::AttachStream

(

std::shared_ptr< net::WireStream >

stream

)

Mirror persisted fragments to an additional wire stream.

After the store accepts a batch, the writer calls WireStream::Send() on each attached stream, and DrainAndClose() follows the last batch with a closure marker. A successful send means local transport admission, not remote delivery. A transport failure stops subsequent writes but cannot revoke store confirmations returned for the current batch. The writer keeps the stream alive while attached.

Parameters

stream

The wire stream to fan output out to.

Returns

OK if the stream was attached, otherwise an error status.

Cancel()

a11::Task a11::stores::ChunkStoreWriter::Cancel

(

)

Stop the writer immediately, discarding any queued chunks.

Returns

An awaitable that resolves once teardown completes.

Create()

absl::StatusOr< std::shared_ptr< ChunkStoreWriter > > a11::stores::ChunkStoreWriter::Create ( std::shared_ptr< ChunkStore >  store, ChunkStoreWriterOptions  options = {}  )

static

Create a writer over store.

Parameters

store	The store to persist chunks to.
options	Tuning for offset, sticky-mimetype compression, and how much is buffered/flushed at once.

Returns

A shared, ready-to-use writer, or an error status if the options are invalid.

DetachStream()

absl::Status a11::stores::ChunkStoreWriter::DetachStream

(

const std::shared_ptr< net::WireStream > &

stream

)

Stop mirroring fragments to a previously attached wire stream.

Parameters

stream

The stream to detach.

Returns

OK if the stream was detached, or an error if it was not attached.

DrainAndClose()

a11::Task a11::stores::ChunkStoreWriter::DrainAndClose

(

)

Flush every queued chunk, then close the writer.

This closes the backing store to further writes, but it does not append a final fragment. The producer must mark its last chunk final=true (or write a null final chunk through AsyncNode) before draining when readers need a final sequence number to identify the logical end of the stream.

Attached streams are told: after the last batch is flushed and teed, the writer sends one closure marker – a status chunk carrying data::kCloseAttribute and the OK close status – so a mirror of this node on the far side closes its own write half. A peer that cannot be reached does not keep the store open; the send error surfaces as this writer's terminal status and through the returned awaitable.

Returns

An awaitable that resolves once the flush and storage close complete.

EnqueueChunk()

ChunkStoreWrite a11::stores::ChunkStoreWriter::EnqueueChunk	(	data::Chunk	chunk,
		std::optional< std::uint32_t >	seq = std::nullopt,
		bool	final = false,
		bool	ensure_started = true
	)

Enqueue a chunk, exposing backpressure and confirmation separately.

Unlike PutChunk(), this returns both awaitables: admitted resolves once the chunk is accepted into the bounded queue and confirmation resolves with the sequence assigned by the backing store. Await admission to pace production and confirmation to know the store accepted the write.

Parameters

chunk	The chunk to enqueue.
seq	Optional explicit sequence number; assigned automatically if unset.
final	Whether this chunk establishes the logical final sequence. This does not close the writer or backing store.
ensure_started	Whether to start the flush loop as part of enqueuing.

Returns

A ChunkStoreWrite holding the admission and confirmation awaitables.

EnsureStarted()

void a11::stores::ChunkStoreWriter::EnsureStarted

(

)

Start the background flush loop if it is not already running.

Writing normally starts the loop lazily; call this to begin flushing before the first chunk is enqueued.

Flush()

void a11::stores::ChunkStoreWriter::Flush

(

)

Run the flush loop now, on the calling thread, if it is idle.

Where EnsureStarted() hands the flush to a worker – a scheduler hop, and for a caller who then awaits the confirmation an event-loop turn before the store is even asked – this does the work here. A store that accepts the batch inline, as LocalChunkStore does, has confirmed the write by the time this returns, so the confirmation is already resolved and its awaiter never suspends.

Call it before waiting for a write, not when enqueuing one. Enqueue does not flush because producers pacing against the admission buffer may run ahead of the store. The Python binding flushes on the confirmation's first await.

A store that cannot answer inline leaves the operation in flight and this is a no-op, which is what keeps batching where batching is worth something: chunks enqueued while a slow store is busy accumulate and go out in one PutMany.

GetAbortStatus()

std::optional< absl::Status > a11::stores::ChunkStoreWriter::GetAbortStatus

(

)

const

Returns

The status the writer was aborted with, or empty if not aborted.

GetStatus()

std::optional< absl::Status > a11::stores::ChunkStoreWriter::GetStatus

(

)

const

Returns

The writer's terminal status, or empty while still open.

IsWritable()

bool a11::stores::ChunkStoreWriter::IsWritable

(

)

const

Returns

Whether the writer still accepts chunks (false once drained, closed, or aborted).

options()

ChunkStoreWriterOptions a11::stores::ChunkStoreWriter::options

(

)

const

Returns

The options this writer was created with.

PutChunk()

a11::Future< std::uint32_t > a11::stores::ChunkStoreWriter::PutChunk	(	data::Chunk	chunk,
		std::optional< std::uint32_t >	seq = std::nullopt,
		bool	final = false
	)

Write a chunk and await backing-store confirmation.

Convenience path for callers that only need confirmation, without observing queue admission separately.

Parameters

chunk	The chunk to write.
seq	Optional explicit sequence number; assigned automatically if unset.
final	Whether this chunk establishes the logical final sequence. This does not close the writer or backing store.

Returns

An awaitable that resolves with the stored sequence number.

queue_size()

size_t a11::stores::ChunkStoreWriter::queue_size

(

)

const

Returns

The number of chunks currently waiting in the flush queue.

store()

std::shared_ptr< ChunkStore > a11::stores::ChunkStoreWriter::store

(

)

const

Returns

The store this writer persists chunks to.

WaitForBufferToDrain()

a11::Task a11::stores::ChunkStoreWriter::WaitForBufferToDrain

(

)

Wait until the in-flight write buffer empties.

A backpressure checkpoint before enqueuing more chunks. Covers both halves of a write: the store confirmation and the mirror handed to any attached WireStream. A chunk's own confirmation resolves on the store alone, so this is the barrier for a caller that needs the mirror to have been handed over.

Returns

An awaitable that resolves once the buffer has drained.

---

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

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