Skip to main content

Crate krabka_remote_storage

Crate krabka_remote_storage 

Source
Expand description

KIP-405 tiered-storage SPI and reference implementations for Krabka.

This crate is the foundation layer for Krabka’s tiered storage. It defines the two plugin SPIs and the data model that they exchange, and it supplies the two reference implementations that the rest of the tiered-storage stack builds and tests against. It mirrors the shapes of Apache Kafka’s storage-api module (org.apache.kafka.server.log.remote.storage).

It also holds the write-once (WORM) archive layer, which turns an object-store tier into a compliance archive: single-PUT objects are conditional creates, multipart objects are guarded by verified bucket retention, every segment copy seals a signed and hash-chained manifest, and the backend refuses every delete. See Write-once archive mode.

§What this crate provides

§Boundary with the broker

This crate is the SPI and reference-implementation layer. Broker-specific behavior such as segment-copy scheduling, Fetch remote reads, local-vs-remote retention policy, and topic config parsing lives in the broker.

The SPIs are intentionally synchronous. They mirror Kafka’s blocking RemoteStorageManager / RemoteLogMetadataManager, which the broker drives from a thread pool. The broker wraps the calls in spawn_blocking. Because the SPIs stay synchronous, the segment-copy and segment-fetch paths here need no async runtime of their own. The archive verifier is the exception: verify_archive is an async fn over the object store, and the krabka-worm-verify binary starts a Tokio runtime to drive it.

§Write-once (WORM) archive mode

S3RemoteStorage::with_worm puts a verified S3 backend into archive mode. Startup confirms versioning and locked default retention (S3 Object Lock in compliance mode or GCS Bucket Lock). Each copy writes small objects with PutMode::Create; multipart objects atomically reserve the key, then read back and verify the completed digest and version. The manifest records which protection each object used, joins the partition’s hash chain, and is signed with Ed25519. The backend refuses every delete, and a WormConfig::write_only archive refuses every remote fetch as well.

The bucket enforces the retention, not this crate. The archive layer refuses to start unless the bucket already has the required policy.

verify_archive reads a finished archive back with nothing but the objects. It recomputes every chain head, checks every signature against a TrustedManifestKeys set, and confirms that every object a manifest names is present with the recorded size. Tail truncation is the one attack the archive cannot reveal on its own, so a caller that holds an independently recorded tip passes it as VerifyRequest::expect_head. The krabka-worm-verify binary is the same check with a command line and graded exit codes.

§Filesystem-backed remote tier

use {
    bytes::Bytes,
    krabka_ids::LeaderEpoch,
    krabka_remote_storage::{
        IndexType, LocalTieredStorage, LogSegmentData, RemoteLogSegmentDetails,
        RemoteLogSegmentId, RemoteLogSegmentMetadata, RemoteLogSegmentState,
        RemoteStorageManager, TopicIdPartition,
    },
    std::{collections::BTreeMap, path::PathBuf},
    uuid::Uuid,
};

let storage = LocalTieredStorage::new(PathBuf::from("/var/lib/krabka-remote"));
let topic_partition = TopicIdPartition::new(Uuid::new_v4(), "orders", 0);
let segment_id = RemoteLogSegmentId::new(topic_partition, Uuid::new_v4());
let mut leader_epochs = BTreeMap::new();
leader_epochs.insert(LeaderEpoch(0), 0);
let metadata = RemoteLogSegmentMetadata::new(
    segment_id,
    0,
    999,
    1_713_000_000_000,
    1,
    1_713_000_000_000,
    RemoteLogSegmentDetails::new(
        1_048_576,
        RemoteLogSegmentState::CopySegmentStarted,
        leader_epochs,
    ),
)?;

// The broker fills these paths from a closed local log segment.
let segment = LogSegmentData {
    log_segment: PathBuf::from("/var/lib/krabka/orders-0/00000000000000000000.log"),
    offset_index: PathBuf::from("/var/lib/krabka/orders-0/00000000000000000000.index"),
    time_index: PathBuf::from("/var/lib/krabka/orders-0/00000000000000000000.timeindex"),
    transaction_index: None,
    producer_snapshot_index: None,
    leader_epoch_index: Bytes::new(),
};
let _custom_metadata = storage.copy_log_segment_data(&metadata, &segment)?;
let bytes = storage.fetch_index(&metadata, IndexType::Offset)?;

Re-exports§

pub use dump::PartitionDump;
pub use dump::RlmmCacheDump;
pub use index::AbortedTxnIndexEntry;
pub use index::BytePosition;
pub use index::LogOffset;
pub use index::OffsetIndexEntry;
pub use index::RelativeOffset;
pub use index::TimeIndexEntry;
pub use index::TimestampMs;
pub use index::corrupt_log;
pub use index::end_position_for;
pub use index::first_batch_at_or_after;
pub use index::first_record_at_or_after_timestamp;
pub use index::parse_offset_index;
pub use index::parse_time_index;
pub use index::parse_txn_index;
pub use index::position_for_relative_offset;
pub use index::relative_offset_floor_for_timestamp;
pub use index::txn_overlaps;

Modules§

diskless
Shared diskless-WAL range metadata and independently versioned formats.
dump
RlmmCacheDump is a flat, owned snapshot of an InmemoryRemoteLogMetadataManager’s cache. The topic-backed manager’s on-disk snapshot uses it.
index
Decoders for the Kafka-format index and log bytes that a remote segment carries.

Structs§

ArchiveVerifyReport
Verification result for a whole archive.
AuthenticatedArchive
A deep WORM verification and the signed object claims it authenticated.
ChainHead
Head of a partition’s manifest hash chain.
ChainStamp
Where this manifest sits in its partition’s hash chain.
CustomMetadata
Opaque bytes that an RemoteStorageManager can return from copy_log_segment_data, for example an object-store key or a version id. The caller sends the bytes back on every later call for that segment.
EpochId
Identifier of one unbroken run of a partition’s chain.
EpochSpan
One unbroken run of a partition’s chain, as the archive holds it.
GcsConfig
Connection and bucket parameters for native Google Cloud Storage.
HexBytes
Variable-length bytes that serialise as a lowercase hex string.
IndexCacheStats
A point-in-time reading of the cache’s counters, for the broker’s metrics.
InmemoryRemoteLogMetadataManager
In-memory RemoteLogMetadataManager: one RemoteLogMetadataCache per partition behind a single mutex.
LocalTieredStorage
A RemoteStorageManager that keeps offloaded segments on a local filesystem under root.
LogSegmentData
The local files, and the in-memory leader-epoch bytes, that make up one log segment for copy to the remote tier.
ManifestBody
Everything a manifest signature covers.
ManifestSeq
Position of one manifest in its partition’s hash chain.
ManifestSignature
Ed25519 signature over one manifest’s chain head.
ObjectEntry
One object that a segment copy wrote to the archive.
ObjectProtectionReport
Count and bounded key sample for one object-protection mechanism.
OffsetGap
A hole between two consecutive archived segments.
PartitionDirName
The partition that a remote-tier directory name identifies.
PartitionVerifyReport
Verification result for one partition directory.
RemoteIndexCache
A bounded on-disk cache of remote segment index objects.
RemoteLogSegmentDetails
Size, lifecycle state, and leader-epoch offsets for a remote log segment.
RemoteLogSegmentId
Globally-unique identifier for one remote log segment: the owning partition plus a random per-segment UUID.
RemoteLogSegmentMetadata
Metadata for one segment that is stored, or is being stored, in the remote tier.
RemoteLogSegmentMetadataUpdate
An update to an existing RemoteLogSegmentMetadata’s lifecycle state.
RemotePartitionDeleteMetadata
Metadata describing the deletion lifecycle of a partition’s remote data.
S3Config
Connection and bucket parameters for an S3-compatible backend.
S3RemoteStorage
A RemoteStorageManager backed by any S3-compatible object store.
SealedManifest
One sealed manifest: the value, the bytes to write, and the receipt.
SegmentFileName
The segment artifact that a remote-tier filename identifies.
SegmentIdentity
The segment this manifest describes, flattened out of RemoteLogSegmentMetadata.
SegmentManifest
One segment manifest, as it is written to and read back from the archive.
Sha256Digest
SHA-256 digest of one archived object’s body.
TopicIdPartition
A partition addressed by its stable topic UUID, with the topic name for diagnostics.
TrustedManifestKeys
Public keys the verifier accepts, keyed by key_id.
VerifyBreak
The first detected break in one partition.
VerifyRequest
What to verify, and against what expectation.
WormArchiver
Builds and signs the per-segment integrity manifest that backs WORM mode.
WormChainRecord
The chain receipt a copy leaves on a segment’s custom metadata.
WormConfig
Settings that make an object-store backend a write-once archive.

Enums§

IndexCacheOutcome
What a lookup did, so the caller can count hits and misses without reaching into the cache’s internals.
IndexType
The kinds of index a segment carries alongside its .log data.
ObjectStoreConfig
Selects and parameterises the object-store backend to construct.
RemoteLogSegmentState
Lifecycle state of a remote log segment.
RemotePartitionDeleteState
Lifecycle state of a remote partition deletion.
RemoteStorageError
Errors raised by RemoteStorageManager and RemoteLogMetadataManager implementations.
VerifyDepth
How hard the verifier works.
WormError
A failure in the WORM archive layer.

Constants§

DEFAULT_INDEX_CACHE_TOTAL_SIZE_BYTES
Kafka’s remote.log.index.file.cache.total.size.bytes default: 1 GiB.
DEFAULT_MULTIPART_CHUNK_SIZE
Default per-part size for multipart uploads. The size is 16 MiB. AWS requires >= 5 MiB per non-final part and caps the parts at 10 000, so 16 MiB scales past any real segment.
DEFAULT_MULTIPART_THRESHOLD
Default threshold above which a segment upload changes from a single PUT to a streaming multipart upload. The threshold is 100 MiB, well below AWS’s 5 GiB single-PUT cap.
LOG_FILE_SUFFIX
Filename suffix of a segment’s data, as distinct from one of its indexes.
MANIFEST_BODY_DOMAIN
Domain separation for the chain preimage, so a manifest body can never collide with any other chained value.
MANIFEST_DOMAIN
Domain separation for the manifest signature. Distinct from [krabka_audit::signing::CHECKPOINT_DOMAIN] — never share a domain across signature purposes.
MANIFEST_FORMAT_VERSION
Version of the manifest encoding, both the JSON shape and the canonical byte layout. A change to either is a change to this number.
MANIFEST_SUFFIX
Object-store key suffix of a segment manifest.
MAX_MANIFEST_BYTES
Byte cap on one manifest object.
REMOTE_INDEX_CACHE_DIR
The directory, relative to the broker’s log directory, that holds the cached index files. The name is Kafka’s.
WORM_CHAIN_RECORD_VERSION
Version of the WormChainRecord JSON in a segment’s CustomMetadata.

Traits§

RemoteLogMetadataManager
SPI for the store that holds metadata about remote log segments.
RemoteStorageManager
SPI for the remote object store that holds offloaded segment data.

Functions§

authenticate_archive
Deeply verifies an archive and returns the signed claims for every object accepted by the chain walk.
canonical_manifest_bytes
Deterministic byte encoding of a manifest body.
decode_kafka_uuid
Reads back a UUID that kafka_uuid rendered.
kafka_uuid
Kafka renders UUIDs in URL-safe, unpadded Base64 for remote-tier paths.
manifest_head
Chain head after body.
manifest_signing_bytes
Canonical signed payload for a manifest.
next_chain_stamp
Next chain position for a partition, given every segment the metadata manager knows about.
parse_partition_dir_name
Reads back the partition that partition_dir_name rendered.
parse_segment_file_name
Reads back the segment artifact that segment_file_name rendered.
partition_dir_name
Directory name used by Kafka’s LocalTieredStorage for a partition.
segment_file_name
Filename used by Kafka’s LocalTieredStorage for one segment artifact.
verify_archive
Verifies every partition the request selects.
verify_manifest_signature
Recomputes the head from the body, then checks the signature against public_key.