krabka_remote_storage/worm/manifest/
body.rs

1//! The manifest document: the segment it describes, the objects it names,
2//! and the envelope that carries the signature.
3//!
4//! These are the `serde` types of the manifest file. `ManifestBody` is the
5//! signed content and `SegmentManifest` is the file as a whole.
6
7use {
8    super::{ChainStamp, ManifestSignature, Sha256Digest},
9    crate::metadata::RemoteLogSegmentMetadata,
10    serde::{Deserialize, Serialize},
11    std::collections::BTreeMap,
12    uuid::Uuid,
13};
14
15/// One object that a segment copy wrote to the archive.
16#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
17#[serde(deny_unknown_fields)]
18pub struct ObjectEntry {
19    /// File suffix this object carries, such as `.log` or `.index`.
20    pub suffix: String,
21    /// Full object-store key of the object.
22    pub key: String,
23    /// Size of the object body in bytes.
24    pub size_bytes: u64,
25    /// `SHA-256` of the object body. The only integrity claim in this entry.
26    pub sha256: Sha256Digest,
27    /// Object-store `ETag`, as the store reported it at write time.
28    ///
29    /// A **locator, not an integrity proof**. A multipart `ETag` is a checksum
30    /// of the part checksums and not of the object body, so it cannot confirm
31    /// the bytes. Use it to find and pin an object; use
32    /// [`Self::sha256`] to prove it.
33    pub e_tag: Option<String>,
34    /// Object-store version id, when the bucket has versioning on.
35    ///
36    /// A **locator, not an integrity proof**. It names which version of the
37    /// key to read back. [`Self::sha256`] is the only integrity claim.
38    pub version_id: Option<String>,
39    /// Whether this write used an atomic create precondition.
40    ///
41    /// `false` means the object relied on the bucket's versioning and default
42    /// retention policy, as multipart uploads do.
43    #[serde(default)]
44    pub create_precondition: bool,
45}
46
47/// The segment this manifest describes, flattened out of
48/// [`RemoteLogSegmentMetadata`].
49///
50/// The manifest keeps its own copy rather than a reference to the metadata
51/// store, because a verifier reads the archive alone.
52#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
53#[serde(deny_unknown_fields)]
54pub struct SegmentIdentity {
55    /// Topic name, for diagnostics.
56    pub topic: String,
57    /// Stable topic id.
58    pub topic_id: Uuid,
59    /// Partition index.
60    pub partition: i32,
61    /// Per-segment id.
62    pub segment_id: Uuid,
63    /// First offset (inclusive) in the segment.
64    pub start_offset: i64,
65    /// Last offset (inclusive) in the segment.
66    pub end_offset: i64,
67    /// Highest record timestamp in the segment.
68    pub max_timestamp_ms: i64,
69    /// Broker that copied the segment.
70    pub broker_id: i32,
71    /// Wall-clock time of the copy event.
72    pub event_timestamp_ms: i64,
73    /// Size of the `.log` data in bytes.
74    pub segment_size_bytes: i64,
75    /// Leader epoch to the first offset that epoch contributed.
76    pub leader_epochs: BTreeMap<i32, i64>,
77    /// `true` when the segment has no transaction index.
78    pub txn_index_empty: bool,
79}
80
81impl SegmentIdentity {
82    /// Flattens the fields a manifest records out of `md`.
83    #[must_use]
84    pub fn from_metadata(md: &RemoteLogSegmentMetadata) -> Self {
85        let id = md.remote_log_segment_id();
86        Self {
87            topic: id.topic_id_partition.topic.clone(),
88            topic_id: id.topic_id_partition.topic_id,
89            partition: id.topic_id_partition.partition,
90            segment_id: id.id,
91            start_offset: md.start_offset(),
92            end_offset: md.end_offset(),
93            max_timestamp_ms: md.max_timestamp_ms(),
94            broker_id: md.broker_id(),
95            event_timestamp_ms: md.event_timestamp_ms(),
96            segment_size_bytes: i64::from(md.segment_size_in_bytes()),
97            leader_epochs: md
98                .segment_leader_epochs()
99                .iter()
100                .map(|(epoch, offset)| (epoch.0, *offset))
101                .collect(),
102            txn_index_empty: md.txn_index_empty(),
103        }
104    }
105}
106
107/// Everything a manifest signature covers.
108#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
109#[serde(deny_unknown_fields)]
110pub struct ManifestBody {
111    /// Encoding version. See [`MANIFEST_FORMAT_VERSION`](super::MANIFEST_FORMAT_VERSION).
112    pub format_version: u32,
113    /// The segment this manifest describes.
114    pub segment: SegmentIdentity,
115    /// Every object the copy wrote, in the order the copy wrote them.
116    pub objects: Vec<ObjectEntry>,
117    /// Chain position of this manifest.
118    pub chain: ChainStamp,
119}
120
121/// One segment manifest, as it is written to and read back from the archive.
122///
123/// The signature is a **sibling** of the body and never a member of it. A
124/// signature can then not be included in what it signs, and the property holds
125/// by the shape of the type rather than by the care of the writer.
126#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
127#[serde(deny_unknown_fields)]
128pub struct SegmentManifest {
129    /// The signed content.
130    pub body: ManifestBody,
131    /// The signature, when the archive has a signing key configured.
132    pub signature: Option<ManifestSignature>,
133}
134
135#[cfg(test)]
136mod tests {
137    use {
138        super::*,
139        crate::{
140            metadata::{
141                RemoteLogSegmentDetails, RemoteLogSegmentId, RemoteLogSegmentState,
142                TopicIdPartition,
143            },
144            worm::manifest::{
145                HexBytes,
146                test_support::{KEY_ID, bare_object, located_object, sample_body, sample_segment},
147            },
148        },
149        assert2::check,
150        krabka_ids::LeaderEpoch,
151    };
152
153    #[yare::parameterized(
154        every_optional_populated = {
155            SegmentManifest {
156                body: ManifestBody {
157                    objects: vec![located_object()],
158                    ..sample_body()
159                },
160                signature: Some(ManifestSignature {
161                    key_id: KEY_ID.to_owned(),
162                    public_key: HexBytes(vec![0x01, 0x02, 0x03]),
163                    signature: HexBytes(vec![0xfe, 0xed]),
164                }),
165            },
166        },
167        every_optional_absent = {
168            SegmentManifest {
169                body: ManifestBody {
170                    objects: vec![bare_object()],
171                    ..sample_body()
172                },
173                signature: None,
174            },
175        },
176    )]
177    fn manifest_json_round_trips(manifest: SegmentManifest) {
178        let json = serde_json::to_string(&manifest).unwrap();
179        let parsed: SegmentManifest = serde_json::from_str(&json).unwrap();
180        check!(parsed == manifest);
181    }
182
183    #[test]
184    fn manifest_json_rejects_unknown_fields_and_reads_legacy_precondition() {
185        // An unknown field is a decode failure, not a silently dropped one.
186        let with_extra = serde_json::json!({
187            "body": {"format_version": 1, "surprise": true},
188            "signature": null,
189        });
190        check!(serde_json::from_value::<SegmentManifest>(with_extra).is_err());
191
192        let mut legacy = serde_json::to_value(SegmentManifest {
193            body: sample_body(),
194            signature: None,
195        })
196        .unwrap();
197        legacy["body"]["format_version"] = serde_json::Value::from(1);
198        legacy["body"]["objects"][0]
199            .as_object_mut()
200            .unwrap()
201            .remove("create_precondition");
202        let legacy: SegmentManifest = serde_json::from_value(legacy).unwrap();
203        check!(!legacy.body.objects[0].create_precondition);
204    }
205
206    #[test]
207    fn segment_identity_flattens_metadata() {
208        let md = RemoteLogSegmentMetadata::new(
209            RemoteLogSegmentId::new(
210                TopicIdPartition::new(Uuid::from_u128(0x11), "orders", 3),
211                Uuid::from_u128(0x22),
212            ),
213            100,
214            199,
215            1_713_000_000_000,
216            7,
217            1_713_000_001_000,
218            RemoteLogSegmentDetails::new(
219                4096,
220                RemoteLogSegmentState::CopySegmentFinished,
221                maplit::btreemap! {LeaderEpoch(0) => 100, LeaderEpoch(1) => 150},
222            ),
223        )
224        .unwrap()
225        .with_txn_index_empty(true);
226
227        check!(
228            SegmentIdentity::from_metadata(&md)
229                == SegmentIdentity {
230                    txn_index_empty: true,
231                    ..sample_segment()
232                }
233        );
234    }
235}
krabka_remote_storage/worm/manifest/body.rs

krabka_remote_storage/worm/manifest/
body.rs
