Bedrock.ObjectStorage.ChunkWriter (bedrock v0.5.3)

View Source

Writes transaction chunks to object storage with hybrid sizing.

ChunkWriter accumulates transactions and flushes them to object storage when either condition is met:

  • Size threshold reached (e.g., 64MB)
  • Time gap exceeded (e.g., 5 minutes since last transaction)

Hybrid Sizing Strategy

Since transaction IDs are microsecond timestamps, time-based cutting is equivalent to version-based. When no new transactions arrive for the configured time gap, the current buffer is flushed even if small.

Background compaction (wxf-6) will later merge small chunks into larger ones.

Usage

{:ok, writer} = ChunkWriter.new(backend, "a",
  size_threshold: 64 * 1024 * 1024,  # 64MB
  time_gap_ms: 5 * 60 * 1000         # 5 minutes
)

{:ok, writer} = ChunkWriter.add_transaction(writer, version, data)
{:ok, writer} = ChunkWriter.add_transaction(writer, version2, data2)

# Check if flush is needed (call periodically or after adds)
case ChunkWriter.maybe_flush(writer) do
  {:ok, writer, :flushed} -> # chunk was written
  {:ok, writer, :not_needed} -> # nothing to flush
end

# Force flush remaining transactions
{:ok, writer} = ChunkWriter.flush(writer)

Summary

Types

t()
transaction_data()
version()

Functions

add_transaction(writer, version, data)

Adds a transaction to the writer's buffer.

buffer_count(chunk_writer)

Returns the number of transactions in the buffer.

buffer_size(chunk_writer)

Returns the current buffer size in bytes.

chunks_written(chunk_writer)

Returns the number of chunks written so far.

empty?(chunk_writer)

Returns true if the buffer is empty.

flush(writer)

Forces a flush of the current buffer, even if thresholds haven't been reached.

maybe_flush(writer)

Checks if a flush is needed and performs it if so.

new(backend, shard_tag, opts \\ [])

Creates a new chunk writer.

Types

t()

@type t() :: %Bedrock.ObjectStorage.ChunkWriter{
  backend: Bedrock.ObjectStorage.backend(),
  buffer: [{version(), transaction_data()}],
  buffer_size: non_neg_integer(),
  chunks_written: non_neg_integer(),
  last_add_time: integer() | nil,
  shard_tag: String.t(),
  size_threshold: pos_integer(),
  time_gap_ms: pos_integer()
}

transaction_data()

@type transaction_data() :: binary()

version()

@type version() :: non_neg_integer()

Functions

add_transaction(writer, version, data)

@spec add_transaction(t(), version(), transaction_data()) :: {:ok, t()}

Adds a transaction to the writer's buffer.

Transactions should be added in version order. After adding, call maybe_flush/1 to check if a flush is needed.

buffer_count(chunk_writer)

@spec buffer_count(t()) :: non_neg_integer()

Returns the number of transactions in the buffer.

buffer_size(chunk_writer)

@spec buffer_size(t()) :: non_neg_integer()

Returns the current buffer size in bytes.

chunks_written(chunk_writer)

@spec chunks_written(t()) :: non_neg_integer()

Returns the number of chunks written so far.

empty?(chunk_writer)

@spec empty?(t()) :: boolean()

Returns true if the buffer is empty.

flush(writer)

@spec flush(t()) :: {:ok, t()} | {:error, term()}

Forces a flush of the current buffer, even if thresholds haven't been reached.

Does nothing if buffer is empty.

maybe_flush(writer)

@spec maybe_flush(t()) :: {:ok, t(), :flushed | :not_needed} | {:error, term()}

Checks if a flush is needed and performs it if so.

Flush triggers:

  • Buffer size >= size_threshold
  • Time since last add >= time_gap_ms (and buffer not empty)

Returns

  • {:ok, writer, :flushed} - Chunk was written
  • {:ok, writer, :not_needed} - No flush needed
  • {:error, reason} - Flush failed

new(backend, shard_tag, opts \\ [])

@spec new(Bedrock.ObjectStorage.backend(), String.t(), keyword()) :: {:ok, t()}

Creates a new chunk writer.

Options

  • :size_threshold - Flush when buffer exceeds this size (default: 64MB)
  • :time_gap_ms - Flush when this many ms pass without new transactions (default: 5 min)