krabka_log/
lib.rs

1//! Byte-compatible reader/writer for Apache Kafka's on-disk log format.
2//!
3//! This crate is the append-only storage layer that the Krabka broker uses.
4//! It reads and writes Kafka 4.x's on-disk log format byte-for-byte:
5//! 20-digit zero-padded segment filenames, sparse `.index` and
6//! `.timeindex` files, and append-only `.log` files that hold
7//! [`krabka_protocol::records::RecordBatch`] v2 streams.
8//!
9//! ## What this crate does
10//!
11//! - Open and recover existing log directories.
12//! - Append `RecordBatch`es to the active segment.
13//! - Read sequentially from an absolute offset.
14//! - Truncate the log to an offset, for replication and leader election.
15//! - Time-based and size-based retention.
16//! - Deliver-at-time visibility, where a batch becomes readable only once its
17//!   activation time has passed.
18//!
19//! ## Scope and boundaries
20//!
21//! This crate is the byte-compatible segment and index layer. It exposes
22//! append, read, truncation, leader-epoch checkpoint, transaction-index,
23//! retention, and compaction primitives over a single log directory.
24//! `krabka-broker` applies broker-level policy above this storage layer. That
25//! policy covers topic configuration, leader and follower ownership,
26//! tiered-storage scheduling, transaction visibility rules, and write
27//! serialization.
28//!
29//! ## Quick start
30//!
31//! ```no_run
32//! use {
33//!     krabka_ids::Offset,
34//!     krabka_log::{Log, LogConfig},
35//!     krabka_protocol::records::RecordBatch,
36//!     krabka_units::prelude::mebibytes,
37//! };
38//!
39//! let mut log = Log::open("/var/kafka/my-topic-0", LogConfig::default()).unwrap();
40//! let mut batch = RecordBatch::default();
41//! // ... fill the batch ...
42//! let assigned_offset = log.append(&mut batch).unwrap();
43//!
44//! let out = log.read(Offset(0), mebibytes(1)).unwrap();
45//! # let _ = (assigned_offset, out);
46//! ```
47//! ## Exporting a segment
48//!
49//! ```no_run
50//! use krabka_log::{Log, LogConfig};
51//!
52//! # fn run() -> Result<(), Box<dyn std::error::Error>> {
53//! let mut log = Log::open("/var/kafka/my-topic-0", LogConfig::default())?;
54//! for segment in log.tierable_segments() {
55//!     println!(
56//!         "segment {} starts at offset {}",
57//!         segment.log_path.display(),
58//!         segment.base_offset
59//!     );
60//! }
61//! # Ok(())
62//! # }
63//! ```
64
65#![doc(html_root_url = "https://docs.rs/krabka-log/1.1.0")]
66
67mod barrier_marker;
68mod compact;
69#[cfg(any(test, feature = "test-helpers"))]
70mod compaction_pause;
71mod config;
72mod delivery;
73mod error;
74mod filter;
75mod index;
76mod io;
77mod leader_epoch_checkpoint;
78mod log;
79mod log_start_offset_checkpoint;
80/// Shared breadth-first runner for the exhaustive Stateright models.
81#[cfg(test)]
82mod model_check;
83pub mod name;
84mod producer_snapshot;
85mod record_limit;
86mod recovery;
87mod retention;
88mod segment;
89mod stamp_index;
90mod stamp_source;
91#[cfg(test)]
92mod test_support;
93pub mod topic_name;
94mod txn_index;
95
96#[cfg(any(test, feature = "test-helpers"))]
97pub use compaction_pause::{CompactionPause, CompactionPauseControl, compaction_pause};
98#[cfg(any(test, feature = "test-helpers"))]
99pub use io::{DirectFile, IoTarget, LogIo};
100#[krabka_macros::sendfile_cfg]
101pub use log::RawReadDesc;
102#[krabka_macros::sendfile_cfg]
103pub use segment::RawSegmentDesc;
104// Re-export the zero-copy fetch descriptor so broker code can name
105// `krabka_log::FileRegion` without depending on the protocol crate's path.
106#[cfg(any(test, feature = "test-helpers"))]
107pub use stamp_source::MonotonicStampSource;
108pub use {
109    barrier_marker::{BarrierMarker, parse_barrier_marker},
110    config::{
111        CleanupPolicy, DEFAULT_MAX_MESSAGE_SIZE, DEFAULT_READ_AHEAD_MAX, DEFAULT_TAIL_CACHE_SIZE,
112        DEFAULT_WRITEBACK_WINDOW, DeliveryPolicy, LogConfig, RemoteTierFlags, ScheduleOrder,
113        SegmentAllocation,
114    },
115    delivery::{DeliveryAdvance, batch_is_deliverable},
116    error::LogError,
117    filter::{FilteredBatch, filter_batch},
118    io::write_file_atomic,
119    krabka_ids::{LeaderEpoch, Offset, ProducerId},
120    krabka_protocol::records::FileRegion,
121    leader_epoch_checkpoint::{EpochEntry, LeaderEpochCheckpoint, epoch_and_offset_for_entries},
122    log::{
123        ActiveProducer, ActiveSegmentExport, BARRIER_CONTROL_TYPE, BuiltCompaction,
124        CompactionContext, CompactionJob, CompactionOutcome, DueDeletions, FailedDeletion, Log,
125        RawRead, RawReadPlan, ReadOutput, SegmentExport, TransactionAppendRefusal,
126        TransactionalBatch, VerbatimBatch, VerificationGuard,
127    },
128    producer_snapshot::{
129        NUM_BATCHES_TO_RETAIN, ProducerBatchMetadata, ProducerSnapshotEntry, RecoveredProducer,
130        SnapshotDecodeError, decode as decode_producer_snapshot,
131    },
132    retention::now_ms as epoch_ms,
133    segment::{RawSegmentRead, Segment},
134    stamp_index::{StampEntry, StampIndex},
135    stamp_source::StampSource,
136    txn_index::{AbortedTxn, TxnIndex},
137};
krabka_log/lib.rs

krabka_log/
lib.rs
