krabka_format/format/
output.rs

1//! The files a format leaves in a log directory.
2//!
3//! Every formatted directory holds Kafka's `meta.properties`. The metadata log
4//! directory also holds the bootstrap manifest with its binary record stream,
5//! and for a dynamic KIP-853 format the offset-zero metadata checkpoint. Each
6//! writer serializes records the run has already resolved, so the encoding and
7//! the I/O sit together here, apart from the flag handling that decides what
8//! goes in them.
9
10use {
11    crate::{ids::ClusterId, meta_properties::MetaProperties},
12    base64::{Engine as _, engine::general_purpose::STANDARD},
13    krabka_metadata::MetadataRecord,
14    serde::Serialize,
15    serde_wincode::SerdeCompat,
16    std::{env, ffi::OsString, fs, io, path::Path},
17    wincode::Serialize as _,
18};
19
20pub(super) const ZERO_CHECKPOINT_NAME: &str = "00000000000000000000-0000000000.checkpoint";
21
22/// The environment variable that makes a run fail after it writes the file of
23/// the given name. It is a test seam: it is how the tests interrupt a run
24/// partway through, to show that the next run recovers.
25pub const FAIL_AFTER_ENV: &str = "KRABKA_FORMAT_FAIL_AFTER";
26
27/// A failure injected through [`FAIL_AFTER_ENV`], or `None` in normal use.
28#[derive(Debug, Clone, Default)]
29pub(super) struct Fault(Option<OsString>);
30
31impl Fault {
32    /// Reads [`FAIL_AFTER_ENV`].
33    pub(super) fn from_env() -> Self { Self(env::var_os(FAIL_AFTER_ENV)) }
34
35    /// Fails when `path` is the file the fault names.
36    fn after(&self, path: &Path) -> Result<(), String> {
37        match (&self.0, path.file_name()) {
38            (Some(name), Some(written)) if name == written => Err(format!(
39                "injected failure after {} ({FAIL_AFTER_ENV})",
40                path.display()
41            )),
42            _ => Ok(()),
43        }
44    }
45}
46
47/// Writes Kafka's `meta.properties`: the marker of a formatted directory, and
48/// the file the broker reads its cluster, node, and directory ids from on
49/// every boot.
50///
51/// The file is written under a temporary name, synced, and renamed into
52/// place, as Kafka's `PropertiesUtils.writePropertiesFile` does. A run that
53/// stops partway therefore never leaves a truncated marker. A failure gets
54/// the message of the `FormatterException` that `Formatter.doFormat` throws.
55#[tracing::instrument(
56    level = "debug",
57    name = "cli.write_meta_properties",
58    skip_all,
59    fields(log_dir = %log_dir.display(), cluster_id = %meta.cluster_id, node_id = meta.node_id),
60    err
61)]
62pub(super) fn write_meta_properties(
63    log_dir: &Path,
64    meta: &MetaProperties,
65    fault: &Fault,
66) -> Result<(), String> {
67    meta.write_observed(log_dir, |path| fault.after(path).map_err(io::Error::other))
68        .map_err(|e| {
69            format!(
70                "Error while writing meta.properties file {}: {e}",
71                log_dir.display()
72            )
73        })
74}
75
76/// The `bootstrap.records.bin` format version this build writes: a
77/// big-endian `i16` at the front of the file, before the first record.
78///
79/// It is part of the 1.x on-disk contract. The broker's reader,
80/// `krabka_broker::bootstrap::load_bootstrap_records`, refuses a missing or
81/// unknown version, and its `BOOTSTRAP_RECORDS_VERSION` is the same number.
82pub(super) const BOOTSTRAP_RECORDS_VERSION: i16 = 0;
83
84/// The `bootstrap.json` format version this build writes: the required
85/// top-level `version` field.
86///
87/// It is part of the 1.x on-disk contract. No reader exists: the broker never
88/// reads `bootstrap.json`, which only mirrors `bootstrap.records.bin` for an
89/// operator. A reader added later refuses a missing or unknown version. The
90/// number is 1 because the field carried 1 before it was named `version`.
91pub(super) const BOOTSTRAP_MANIFEST_VERSION: u32 = 1;
92
93/// Human-readable manifest written to `<log_dir>/bootstrap.json`.
94#[derive(Debug, Serialize)]
95struct BootstrapManifest {
96    /// Always [`BOOTSTRAP_MANIFEST_VERSION`].
97    version: u32,
98    /// Kafka's 22-character base64 form.
99    cluster_id: ClusterId,
100    record_count: usize,
101    /// Base64-encoded `SerdeCompat<MetadataRecord>` payloads, one per
102    /// seed record. Mirrors the contents of `bootstrap.records.bin` so
103    /// operators can inspect the file without a hex editor.
104    records_b64: Vec<String>,
105}
106
107/// Write the KIP-630/KIP-853 offset-zero bootstrap checkpoint for a
108/// dynamically formatted controller into
109/// `<metadata_log_dir>/__cluster_metadata-0/`, where Kafka writes it.
110///
111/// The checkpoint holds the control state of `control_records`, then
112/// `metadata_records` in their own order, as Kafka's
113/// `Formatter.writeBoostrapSnapshot` writes them. The active controller
114/// writes those records to the metadata log.
115pub(super) fn write_dynamic_checkpoint(
116    metadata_log_dir: &Path,
117    cluster_id: ClusterId,
118    control_records: &[MetadataRecord],
119    metadata_records: &[MetadataRecord],
120    fault: &Fault,
121) -> Result<(), String> {
122    let mut controls = krabka_metadata::MetadataImage::new(cluster_id.into());
123    for record in control_records {
124        controls.apply(record);
125    }
126    let bytes = krabka_raft::serialize_bootstrap_snapshot(
127        controls.kraft_version(),
128        controls.voters(),
129        metadata_records,
130        0,
131    )
132    .map_err(|e| format!("serialize offset-zero checkpoint: {e}"))?;
133    let checkpoint_dir = krabka_raft::metadata_partition_dir(metadata_log_dir);
134    fs::create_dir_all(&checkpoint_dir).map_err(|e| format!("create checkpoint directory: {e}"))?;
135    let path = checkpoint_dir.join(ZERO_CHECKPOINT_NAME);
136    fs::write(&path, bytes).map_err(|e| format!("write offset-zero checkpoint: {e}"))?;
137    fault.after(&path)
138}
139
140/// Serialize the manifest + records to disk under `log_dir`. Returns the
141/// first I/O or encoding error encountered.
142#[tracing::instrument(
143    level = "debug",
144    name = "cli.write_bootstrap_files",
145    skip_all,
146    fields(log_dir = %log_dir.display(), record_count = records.len()),
147    err
148)]
149pub(super) fn write_bootstrap_files(
150    log_dir: &Path,
151    cluster_id: ClusterId,
152    records: &[MetadataRecord],
153    fault: &Fault,
154) -> Result<(), String> {
155    // 1. Per-record `SerdeCompat<MetadataRecord>` payloads.
156    let mut record_blobs: Vec<Vec<u8>> = Vec::with_capacity(records.len());
157    for rec in records {
158        let bytes = <SerdeCompat<MetadataRecord>>::serialize(rec)
159            .map_err(|e| format!("serialize record: {e}"))?;
160        record_blobs.push(bytes);
161    }
162
163    // 2. Binary stream: the version header, then length-prefixed (u32 LE)
164    //    blobs, concatenated.
165    let mut bin = BOOTSTRAP_RECORDS_VERSION.to_be_bytes().to_vec();
166    for blob in &record_blobs {
167        let len: u32 = u32::try_from(blob.len())
168            .map_err(|_| format!("record too large: {} bytes", blob.len()))?;
169        bin.extend_from_slice(&len.to_le_bytes());
170        bin.extend_from_slice(blob);
171    }
172    let bin_path = log_dir.join("bootstrap.records.bin");
173    fs::write(&bin_path, &bin).map_err(|e| format!("write bootstrap.records.bin: {e}"))?;
174    fault.after(&bin_path)?;
175
176    // 3. Manifest JSON (cluster id + base64 mirrors of each blob).
177    let records_b64: Vec<String> = record_blobs.iter().map(|b| STANDARD.encode(b)).collect();
178    let manifest = BootstrapManifest {
179        version: BOOTSTRAP_MANIFEST_VERSION,
180        cluster_id,
181        record_count: records.len(),
182        records_b64,
183    };
184    let json =
185        serde_json::to_string_pretty(&manifest).map_err(|e| format!("serialize manifest: {e}"))?;
186    let json_path = log_dir.join("bootstrap.json");
187    fs::write(&json_path, json).map_err(|e| format!("write bootstrap.json: {e}"))?;
188    fault.after(&json_path)
189}
190
191#[cfg(test)]
192mod tests {
193    use super::*;
194
195    /// The records the golden files hold.
196    fn golden_records() -> Vec<MetadataRecord> {
197        vec![MetadataRecord::V1FeatureLevel(
198            krabka_metadata::FeatureLevelRecord {
199                name: "metadata.version".into(),
200                level: 30,
201            },
202        )]
203    }
204
205    /// `golden_records` in `bootstrap.records.bin`: the version header, then
206    /// one `u32` little-endian length and the wincode record. The broker's
207    /// reader pins the same bytes.
208    const GOLDEN_RECORDS_BIN: &[u8] = include_bytes!("../../tests/fixtures/bootstrap.records.bin");
209
210    /// `golden_records` in `bootstrap.json`, for the cluster id below.
211    const GOLDEN_MANIFEST: &str = r#"{
212  "version": 1,
213  "cluster_id": "AQIDBAUGBwgJCgsMDQ4PEA",
214  "record_count": 1,
215  "records_b64": [
216    "EQAAABAAAAAAAAAAbWV0YWRhdGEudmVyc2lvbh4A"
217  ]
218}"#;
219
220    /// Both bootstrap files are laid out byte for byte as the 1.x contract
221    /// fixes them.
222    #[test]
223    fn the_bootstrap_files_match_their_golden_bytes() {
224        let dir = tempfile::tempdir().unwrap();
225        let cluster_id = ClusterId(uuid::Uuid::from_u128(
226            0x0102_0304_0506_0708_090a_0b0c_0d0e_0f10,
227        ));
228        write_bootstrap_files(dir.path(), cluster_id, &golden_records(), &Fault::default())
229            .unwrap();
230        assert2::assert!(
231            fs::read(dir.path().join("bootstrap.records.bin")).unwrap() == GOLDEN_RECORDS_BIN
232        );
233        assert2::assert!(
234            fs::read_to_string(dir.path().join("bootstrap.json")).unwrap() == GOLDEN_MANIFEST
235        );
236    }
237}
krabka_format/format/output.rs

krabka_format/format/
output.rs
