krabka_remote_storage/worm/
chain.rs

1//! The per-partition manifest chain, and where the next manifest joins it.
2//!
3//! Each copy stamps a [`WormChainRecord`] onto the segment's
4//! [`CustomMetadata`]. The record is the receipt: it says which chain run the
5//! manifest belongs to, where in the run it sits, and what head it produced.
6//! [`next_chain_stamp`] reads those receipts back and works out where the next
7//! manifest goes.
8
9use {
10    crate::{
11        metadata::{CustomMetadata, RemoteLogSegmentMetadata, RemoteLogSegmentState},
12        worm::{
13            error::WormError,
14            manifest::{ChainHead, ChainStamp, EpochId, ManifestSeq},
15        },
16    },
17    krabka_verified::{ChainStep, chain::select_chain_tip, chain_step},
18    serde::{Deserialize, Serialize},
19};
20
21/// Version of the [`WormChainRecord`] JSON in a segment's
22/// [`CustomMetadata`].
23///
24/// Part of the 1.x on-disk contract: a 1.x broker reads every chain record
25/// that any earlier 1.x broker wrote to `__remote_log_metadata`.
26pub const WORM_CHAIN_RECORD_VERSION: i16 = 0;
27
28/// The JSON form of a [`WormChainRecord`], version first.
29#[derive(Serialize, Deserialize)]
30#[serde(deny_unknown_fields)]
31struct WormChainRecordJson {
32    version: i16,
33    epoch_id: EpochId,
34    seq: ManifestSeq,
35    prev_head: ChainHead,
36    head: Option<ChainHead>,
37    manifest_version_id: Option<String>,
38}
39
40/// Just the version of a chain record, read before the rest of it.
41#[derive(Deserialize)]
42struct WormChainRecordVersion {
43    version: Option<serde_json::Value>,
44}
45
46/// The chain receipt a copy leaves on a segment's custom metadata.
47///
48/// The record has two forms. The **request** form, from
49/// [`WormChainRecord::request`], carries no head: the broker builds it before
50/// the copy, when the manifest bytes do not exist yet. The **receipt** form,
51/// from [`WormChainRecord::with_head`], carries the head the manifest produced
52/// and is the only form [`next_chain_stamp`] continues from.
53///
54/// On the wire the record is a JSON object with a required top-level
55/// `"version"`, [`WORM_CHAIN_RECORD_VERSION`]. The object is closed
56/// (`deny_unknown_fields`): a later field always comes with a new version, so
57/// a reader never drops a part of a receipt it does not understand.
58#[derive(Debug, Clone, PartialEq, Eq)]
59pub struct WormChainRecord {
60    /// Chain run this manifest belongs to.
61    pub epoch_id: EpochId,
62    /// Position within the run.
63    pub seq: ManifestSeq,
64    /// Chain head before this manifest.
65    pub prev_head: ChainHead,
66    /// The head *after* this manifest. `None` in the pre-copy request form.
67    pub head: Option<ChainHead>,
68    /// Object-store version id of the manifest object, when the bucket has
69    /// versioning on.
70    pub manifest_version_id: Option<String>,
71}
72
73impl WormChainRecord {
74    /// The pre-copy request form of a record at `stamp`, with no head yet.
75    #[must_use]
76    pub fn request(stamp: ChainStamp) -> Self {
77        Self {
78            epoch_id: stamp.epoch_id,
79            seq: stamp.seq,
80            prev_head: stamp.prev_head,
81            head: None,
82            manifest_version_id: None,
83        }
84    }
85
86    /// Turns a request form into a receipt by recording the head the manifest
87    /// produced.
88    #[must_use]
89    pub fn with_head(mut self, head: ChainHead) -> Self {
90        self.head = Some(head);
91        self
92    }
93
94    /// Records the object-store version id of the manifest object.
95    #[must_use]
96    pub fn with_manifest_version(mut self, version_id: Option<String>) -> Self {
97        self.manifest_version_id = version_id;
98        self
99    }
100
101    /// Encodes the record as the JSON bytes of a [`CustomMetadata`].
102    ///
103    /// # Errors
104    ///
105    /// Returns [`WormError::MalformedChainRecord`] if `serde_json` cannot
106    /// serialise the record. Every field is a `UUID`, an integer, a fixed-size
107    /// byte array, or a string, so this does not happen in practice.
108    pub fn to_custom_metadata(&self) -> Result<CustomMetadata, WormError> {
109        serde_json::to_vec(&WormChainRecordJson {
110            version: WORM_CHAIN_RECORD_VERSION,
111            epoch_id: self.epoch_id,
112            seq: self.seq,
113            prev_head: self.prev_head,
114            head: self.head,
115            manifest_version_id: self.manifest_version_id.clone(),
116        })
117        .map(CustomMetadata)
118        .map_err(|e| WormError::MalformedChainRecord(e.to_string()))
119    }
120
121    /// Decodes a record from a segment's [`CustomMetadata`].
122    ///
123    /// Never panics. Arbitrary bytes produce an error.
124    ///
125    /// # Errors
126    ///
127    /// Returns [`WormError::UnsupportedChainRecordVersion`] when the object
128    /// has no `"version"`, as a record written before krabka 1.0 does, or a
129    /// version other than [`WORM_CHAIN_RECORD_VERSION`]. Returns
130    /// [`WormError::MalformedChainRecord`] when `cm` does not hold the JSON of
131    /// a chain record: bytes that are not `UTF-8`, text that is not a JSON
132    /// object, an object with a missing or unknown field, or a hex string that
133    /// is not 64 characters.
134    pub fn from_custom_metadata(cm: &CustomMetadata) -> Result<Self, WormError> {
135        let malformed = |e: serde_json::Error| WormError::MalformedChainRecord(e.to_string());
136        let probe: WormChainRecordVersion = serde_json::from_slice(&cm.0).map_err(malformed)?;
137        let version = probe.version.map(|version| version.to_string());
138        if version.as_deref() != Some(WORM_CHAIN_RECORD_VERSION.to_string().as_str()) {
139            return Err(WormError::UnsupportedChainRecordVersion { found: version });
140        }
141        let json: WormChainRecordJson = serde_json::from_slice(&cm.0).map_err(malformed)?;
142        Ok(Self {
143            epoch_id: json.epoch_id,
144            seq: json.seq,
145            prev_head: json.prev_head,
146            head: json.head,
147            manifest_version_id: json.manifest_version_id,
148        })
149    }
150
151    /// The chain position the next manifest takes after this one.
152    ///
153    /// `None` for the request form, which has produced no head to chain onto.
154    #[must_use]
155    pub fn next_stamp(&self) -> Option<ChainStamp> {
156        let head = self.head?;
157        match chain_step(self.seq.0, self.seq.0, true) {
158            ChainStep::Continue(next) => Some(ChainStamp {
159                epoch_id: self.epoch_id,
160                seq: ManifestSeq(next),
161                prev_head: head,
162            }),
163            ChainStep::SequenceMismatch | ChainStep::HeadMismatch | ChainStep::Exhausted => None,
164        }
165    }
166}
167
168/// Next chain position for a partition, given every segment the metadata
169/// manager knows about.
170///
171/// Picks the receipt on the segment with the greatest `start_offset`, breaking
172/// a tie by the greatest `seq`, and continues from it. A segment in a delete
173/// state is ignored, and so is a record in the request form, which carries no
174/// head.
175///
176/// Returns a **fresh epoch at genesis** when no receipt survives. That is a new
177/// partition, or a restart on the non-durable in-memory metadata manager, and
178/// in both cases the old chain cannot be continued. A new epoch says so, rather
179/// than restarting the old chain at sequence zero and looking like a rewrite.
180/// Returns `None` when the selected receipt is at `u64::MAX`, because no later
181/// sequence exists and restarting at genesis would hide exhaustion as a new
182/// chain run. Returns `None` too when a live segment carries a chain record of
183/// a version other than [`WORM_CHAIN_RECORD_VERSION`], because this build
184/// cannot tell where that chain ends.
185///
186/// `new_epoch_id` is a parameter and not a `Uuid::new_v4()` call inside, so the
187/// function stays pure and testable.
188#[must_use]
189pub fn next_chain_stamp(
190    segments: &[RemoteLogSegmentMetadata],
191    new_epoch_id: EpochId,
192) -> Option<ChainStamp> {
193    let mut candidates = Vec::with_capacity(segments.len());
194    let mut receipts = Vec::with_capacity(segments.len());
195    for md in segments {
196        let receipt = if matches!(
197            md.state(),
198            RemoteLogSegmentState::DeleteSegmentStarted
199                | RemoteLogSegmentState::DeleteSegmentFinished
200        ) {
201            None
202        } else {
203            match md
204                .custom_metadata()
205                .map(WormChainRecord::from_custom_metadata)
206            {
207                Some(Ok(record)) => Some(record).filter(|record| record.head.is_some()),
208                // A chain record of a version this build does not read: the
209                // chain cannot be continued, and starting a fresh epoch over
210                // it would hide that. Custom metadata with no version at all
211                // is no chain record, like any other backend's metadata: the
212                // object may be another backend's JSON receipt, and a pre-1.0
213                // record never reaches a 1.x broker, whose cluster was
214                // reformatted.
215                Some(Err(WormError::UnsupportedChainRecordVersion { found: Some(_) })) => {
216                    return None;
217                }
218                Some(Err(_)) | None => None,
219            }
220        };
221        let sequence = receipt.as_ref().map_or(0, |record| record.seq.0);
222        candidates.push((md.start_offset(), sequence, receipt.is_some()));
223        receipts.push(receipt);
224    }
225    let Some(index) = select_chain_tip(&candidates) else {
226        return Some(ChainStamp {
227            epoch_id: new_epoch_id,
228            seq: ManifestSeq(0),
229            prev_head: ChainHead::GENESIS,
230        });
231    };
232    receipts.get(index)?.as_ref()?.next_stamp()
233}
234
235#[cfg(test)]
236mod tests {
237    use {
238        super::*,
239        crate::metadata::{RemoteLogSegmentDetails, RemoteLogSegmentId, TopicIdPartition},
240        assert2::check,
241        krabka_ids::LeaderEpoch,
242        proptest::{collection::vec as prop_vec, num::u8::ANY as ANY_U8, proptest},
243        uuid::Uuid,
244    };
245
246    fn epoch() -> EpochId { EpochId(Uuid::from_u128(0x1234)) }
247
248    fn head(byte: u8) -> ChainHead { ChainHead([byte; 32]) }
249
250    fn request_stamp() -> ChainStamp {
251        ChainStamp {
252            epoch_id: epoch(),
253            seq: ManifestSeq(11),
254            prev_head: head(0x5a),
255        }
256    }
257
258    fn sample_metadata(
259        start_offset: i64,
260        state: RemoteLogSegmentState,
261        custom: Option<CustomMetadata>,
262    ) -> RemoteLogSegmentMetadata {
263        let md = RemoteLogSegmentMetadata::new(
264            RemoteLogSegmentId::new(
265                TopicIdPartition::new(Uuid::from_u128(1), "orders", 0),
266                Uuid::from_u128(u128::try_from(start_offset).unwrap() + 100),
267            ),
268            start_offset,
269            start_offset + 99,
270            123,
271            1,
272            456,
273            RemoteLogSegmentDetails::new(
274                8,
275                state,
276                maplit::btreemap! {LeaderEpoch(0) => start_offset},
277            ),
278        )
279        .unwrap();
280        match custom {
281            Some(cm) => md.with_custom_metadata(cm),
282            None => md,
283        }
284    }
285
286    fn receipt(seq: u64, prev: u8, produced: u8) -> CustomMetadata {
287        WormChainRecord::request(ChainStamp {
288            epoch_id: epoch(),
289            seq: ManifestSeq(seq),
290            prev_head: head(prev),
291        })
292        .with_head(head(produced))
293        .to_custom_metadata()
294        .unwrap()
295    }
296
297    #[test]
298    fn next_chain_stamp_starts_new_epoch_on_empty_partition() {
299        let fresh = EpochId(Uuid::from_u128(0xfeed));
300        check!(
301            next_chain_stamp(&[], fresh)
302                == Some(ChainStamp {
303                    epoch_id: fresh,
304                    seq: ManifestSeq(0),
305                    prev_head: ChainHead::GENESIS,
306                })
307        );
308    }
309
310    #[test]
311    fn next_chain_stamp_continues_from_highest_offset_receipt() {
312        let segments = [
313            sample_metadata(
314                0,
315                RemoteLogSegmentState::CopySegmentFinished,
316                Some(receipt(0, 0x00, 0xaa)),
317            ),
318            sample_metadata(
319                200,
320                RemoteLogSegmentState::CopySegmentFinished,
321                Some(receipt(2, 0xbb, 0xcc)),
322            ),
323            sample_metadata(
324                100,
325                RemoteLogSegmentState::CopySegmentFinished,
326                Some(receipt(1, 0xaa, 0xbb)),
327            ),
328        ];
329        check!(
330            next_chain_stamp(&segments, EpochId(Uuid::from_u128(0xfeed)))
331                == Some(ChainStamp {
332                    epoch_id: epoch(),
333                    seq: ManifestSeq(3),
334                    prev_head: head(0xcc),
335                })
336        );
337    }
338
339    #[test]
340    fn next_chain_stamp_breaks_offset_ties_by_greatest_seq() {
341        let request_at_same_offset = WormChainRecord::request(ChainStamp {
342            epoch_id: epoch(),
343            seq: ManifestSeq(9),
344            prev_head: head(0xcc),
345        })
346        .with_head(head(0xdd))
347        .to_custom_metadata()
348        .unwrap();
349        let segments = [
350            sample_metadata(
351                100,
352                RemoteLogSegmentState::CopySegmentFinished,
353                Some(receipt(1, 0xaa, 0xbb)),
354            ),
355            sample_metadata(
356                100,
357                RemoteLogSegmentState::CopySegmentFinished,
358                Some(request_at_same_offset),
359            ),
360        ];
361        check!(
362            next_chain_stamp(&segments, EpochId(Uuid::from_u128(0xfeed)))
363                == Some(ChainStamp {
364                    epoch_id: epoch(),
365                    seq: ManifestSeq(10),
366                    prev_head: head(0xdd),
367                })
368        );
369    }
370
371    #[test]
372    fn next_chain_stamp_ignores_request_form_records() {
373        let request_only = WormChainRecord::request(ChainStamp {
374            epoch_id: epoch(),
375            seq: ManifestSeq(7),
376            prev_head: head(0xbb),
377        })
378        .to_custom_metadata()
379        .unwrap();
380        let segments = [
381            sample_metadata(
382                100,
383                RemoteLogSegmentState::CopySegmentFinished,
384                Some(receipt(1, 0xaa, 0xbb)),
385            ),
386            // Highest offset, but the copy never finished, so no head.
387            sample_metadata(
388                200,
389                RemoteLogSegmentState::CopySegmentStarted,
390                Some(request_only),
391            ),
392        ];
393        check!(
394            next_chain_stamp(&segments, EpochId(Uuid::from_u128(0xfeed)))
395                == Some(ChainStamp {
396                    epoch_id: epoch(),
397                    seq: ManifestSeq(2),
398                    prev_head: head(0xbb),
399                })
400        );
401    }
402
403    #[yare::parameterized(
404        delete_started = { RemoteLogSegmentState::DeleteSegmentStarted },
405        delete_finished = { RemoteLogSegmentState::DeleteSegmentFinished },
406    )]
407    fn next_chain_stamp_ignores_deleted_segments(deleted_state: RemoteLogSegmentState) {
408        let expected_stamp = Some(ChainStamp {
409            epoch_id: epoch(),
410            seq: ManifestSeq(2),
411            prev_head: head(0xbb),
412        });
413        let segments = [
414            sample_metadata(
415                100,
416                RemoteLogSegmentState::CopySegmentFinished,
417                Some(receipt(1, 0xaa, 0xbb)),
418            ),
419            sample_metadata(300, deleted_state, Some(receipt(5, 0xee, 0xff))),
420        ];
421        let actual = next_chain_stamp(&segments, EpochId(Uuid::from_u128(0xfeed)));
422        check!(actual == expected_stamp);
423    }
424
425    #[yare::parameterized(
426        no_custom_metadata = { None },
427        empty_custom_metadata = { Some(CustomMetadata(Vec::new())) },
428        another_backend = { Some(CustomMetadata(b"s3://bucket/key".to_vec())) },
429        unrelated_json = { Some(CustomMetadata(br#"{"key":"value"}"#.to_vec())) },
430        bad_chain_head = {
431            Some(CustomMetadata(br#"{"epoch_id":"00000000-0000-0000-0000-000000000001","seq":1,"prev_head":"zz","head":null,"manifest_version_id":null}"#.to_vec())),
432        },
433    )]
434    fn next_chain_stamp_starts_new_epoch_when_no_receipt_decodes(custom: Option<CustomMetadata>) {
435        let fresh = EpochId(Uuid::from_u128(0xfeed));
436        let expected = ChainStamp {
437            epoch_id: fresh,
438            seq: ManifestSeq(0),
439            prev_head: ChainHead::GENESIS,
440        };
441        let segments = [sample_metadata(
442            100,
443            RemoteLogSegmentState::CopySegmentFinished,
444            custom,
445        )];
446        check!(next_chain_stamp(&segments, fresh) == Some(expected));
447    }
448
449    #[test]
450    fn next_chain_stamp_rejects_sequence_exhaustion() {
451        let exhausted = WormChainRecord::request(ChainStamp {
452            epoch_id: epoch(),
453            seq: ManifestSeq(u64::MAX),
454            prev_head: head(0xaa),
455        })
456        .with_head(head(0xbb))
457        .to_custom_metadata()
458        .unwrap();
459        let segments = [sample_metadata(
460            100,
461            RemoteLogSegmentState::CopySegmentFinished,
462            Some(exhausted),
463        )];
464
465        check!(next_chain_stamp(&segments, EpochId(Uuid::from_u128(0xfeed))) == None);
466    }
467
468    #[yare::parameterized(
469        request = {
470            WormChainRecord::request(request_stamp()),
471        },
472        receipt = {
473            WormChainRecord::request(request_stamp())
474            .with_head(head(0x6b))
475            .with_manifest_version(Some("3HL4kqtJlcpXroDTDmjVBH40Nrjfkd".to_owned())),
476        },
477    )]
478    fn chain_record_round_trips_through_custom_metadata(record: WormChainRecord) {
479        let encoded = record.to_custom_metadata().unwrap();
480        check!(WormChainRecord::from_custom_metadata(&encoded).unwrap() == record);
481    }
482
483    #[test]
484    fn chain_request_and_receipt_next_stamps_are_defined() {
485        let stamp = request_stamp();
486        let request = WormChainRecord::request(stamp);
487        let full = request
488            .clone()
489            .with_head(head(0x6b))
490            .with_manifest_version(Some("3HL4kqtJlcpXroDTDmjVBH40Nrjfkd".to_owned()));
491        check!(
492            request
493                == WormChainRecord {
494                    epoch_id: epoch(),
495                    seq: ManifestSeq(11),
496                    prev_head: head(0x5a),
497                    head: None,
498                    manifest_version_id: None,
499                }
500        );
501        check!(request.next_stamp() == None);
502        check!(
503            full.next_stamp()
504                == Some(ChainStamp {
505                    epoch_id: epoch(),
506                    seq: ManifestSeq(12),
507                    prev_head: head(0x6b),
508                })
509        );
510    }
511
512    fn golden_record() -> WormChainRecord {
513        WormChainRecord::request(request_stamp())
514            .with_head(head(0x6b))
515            .with_manifest_version(Some("v1".to_owned()))
516    }
517
518    /// The exact custom-metadata bytes of [`golden_record`]. A change here is
519    /// a change to the 1.x `__remote_log_metadata` contract.
520    const GOLDEN_RECORD: &str = concat!(
521        r#"{"version":0,"epoch_id":"00000000-0000-0000-0000-000000001234","seq":11,"#,
522        r#""prev_head":"5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a","#,
523        r#""head":"6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b6b","#,
524        r#""manifest_version_id":"v1"}"#,
525    );
526
527    #[test]
528    fn chain_record_encodes_to_the_golden_bytes_and_decodes_back() {
529        check!(golden_record().to_custom_metadata().unwrap().0 == GOLDEN_RECORD.as_bytes());
530        check!(
531            WormChainRecord::from_custom_metadata(&CustomMetadata(
532                GOLDEN_RECORD.as_bytes().to_vec()
533            ))
534            .unwrap()
535                == golden_record()
536        );
537    }
538
539    fn custom_record_with_version(version: Option<serde_json::Value>) -> CustomMetadata {
540        let mut record: serde_json::Value = serde_json::from_str(GOLDEN_RECORD).unwrap();
541        let object = record.as_object_mut().unwrap();
542        object.remove("version");
543        if let Some(version) = version {
544            object.insert("version".to_owned(), version);
545        }
546        CustomMetadata(serde_json::to_vec(&record).unwrap())
547    }
548
549    #[yare::parameterized(
550        pre_1_0_no_version = { custom_record_with_version(None), None },
551        future_version = { custom_record_with_version(Some(serde_json::json!(1))), Some("1") },
552        wrong_type = { custom_record_with_version(Some(serde_json::json!("0"))), Some("\"0\"") },
553    )]
554    fn chain_record_of_a_missing_or_unknown_version_is_refused(
555        custom: CustomMetadata,
556        found: Option<&str>,
557    ) {
558        let err = WormChainRecord::from_custom_metadata(&custom).unwrap_err();
559        assert2::assert!(
560            let WormError::UnsupportedChainRecordVersion { found: actual } = &err,
561            "{err}"
562        );
563        check!(actual.as_deref() == found);
564    }
565
566    #[test]
567    fn next_chain_stamp_refuses_to_continue_past_a_record_of_an_unknown_version() {
568        let mut future: serde_json::Value = serde_json::from_str(GOLDEN_RECORD).unwrap();
569        future["version"] = serde_json::json!(1);
570        let segments = [
571            sample_metadata(
572                100,
573                RemoteLogSegmentState::CopySegmentFinished,
574                Some(receipt(3, 0x01, 0x02)),
575            ),
576            sample_metadata(
577                50,
578                RemoteLogSegmentState::CopySegmentFinished,
579                Some(CustomMetadata(serde_json::to_vec(&future).unwrap())),
580            ),
581        ];
582
583        check!(next_chain_stamp(&segments, EpochId(Uuid::from_u128(0xfeed))) == None);
584    }
585
586    #[test]
587    fn malformed_chain_record_is_an_error_not_a_panic() {
588        let err = WormChainRecord::from_custom_metadata(&CustomMetadata(b"not json".to_vec()))
589            .unwrap_err();
590        check!(matches!(err, WormError::MalformedChainRecord(_)));
591    }
592
593    proptest! {
594        #[test]
595        fn from_custom_metadata_never_panics_on_arbitrary_bytes(
596            bytes in prop_vec(ANY_U8, 0..512usize),
597        ) {
598            let _ = WormChainRecord::from_custom_metadata(&CustomMetadata(bytes));
599        }
600    }
601}
krabka_remote_storage/worm/chain.rs

krabka_remote_storage/worm/
chain.rs
