rocketmq_protocol/protocol/body/
release_checkpoint.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15//! Versioned wire contracts for release snapshots and Store checkpoints.
16
17use std::collections::BTreeSet;
18use std::fmt;
19
20use rocketmq_error::Sensitive;
21use serde::Deserialize;
22use serde::Serialize;
23
24use crate::ProtocolContractViolation;
25
26/// Current checkpoint manifest schema.
27pub const RELEASE_CHECKPOINT_SCHEMA_VERSION: u16 = 1;
28
29/// Non-sensitive capabilities returned after privileged authentication.
30#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
31#[serde(deny_unknown_fields, rename_all = "camelCase")]
32pub struct MaintenanceCapabilitiesResponse {
33    pub schema_version: u16,
34    pub policy_id: String,
35    pub policy_version: u64,
36    pub operations: Vec<String>,
37    pub max_checkpoint_bytes: u64,
38    pub max_store_members: u32,
39    pub max_concurrent_operations: u32,
40    #[serde(default, skip_serializing_if = "Option::is_none")]
41    pub store: Option<MaintenanceStoreCapabilities>,
42}
43
44/// Store identity returned by a Broker maintenance endpoint.
45#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
46#[serde(deny_unknown_fields, rename_all = "camelCase")]
47pub struct MaintenanceStoreCapabilities {
48    pub member_id: String,
49    pub backend: ReleaseCheckpointBackend,
50    pub storage_identity: ReleaseCheckpointStorageIdentity,
51}
52
53/// Store implementation that produced a checkpoint.
54#[derive(Clone, Copy, Debug, Deserialize, Eq, Hash, Ord, PartialEq, PartialOrd, Serialize)]
55#[serde(rename_all = "snake_case")]
56pub enum ReleaseCheckpointBackend {
57    /// Local CommitLog/ConsumeQueue/Index files.
58    Local,
59    /// RocksDB-managed Store state.
60    RocksDb,
61}
62
63/// Consistent Store offsets captured after the checkpoint flush barrier.
64#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
65#[serde(deny_unknown_fields, rename_all = "camelCase")]
66pub struct ReleaseCheckpointOffsets {
67    /// Exclusive end of data accepted by the primary log.
68    pub appended_offset: i64,
69    /// Exclusive end durably flushed before checkpoint capture.
70    pub durable_offset: i64,
71    /// Highest physical source offset represented by ConsumeQueue.
72    pub consume_queue_offset: i64,
73    /// Highest physical source offset represented by Index.
74    pub index_offset: i64,
75}
76
77impl ReleaseCheckpointOffsets {
78    /// Validates ordering between append, durable, and derived progress.
79    ///
80    /// # Errors
81    ///
82    /// Returns a typed invariant error for a negative or out-of-order offset.
83    pub fn validate(self) -> Result<(), ProtocolContractViolation> {
84        if self.appended_offset < 0 || self.durable_offset < 0 || self.consume_queue_offset < 0 || self.index_offset < 0
85        {
86            return Err(ProtocolContractViolation::InvalidCheckpointOffsets {
87                reason: "checkpoint offsets cannot be negative",
88            });
89        }
90        if self.durable_offset > self.appended_offset {
91            return Err(ProtocolContractViolation::InvalidCheckpointOffsets {
92                reason: "durable_offset cannot exceed appended_offset",
93            });
94        }
95        if self.consume_queue_offset > self.durable_offset || self.index_offset > self.durable_offset {
96            return Err(ProtocolContractViolation::InvalidCheckpointOffsets {
97                reason: "derived offsets cannot exceed durable_offset",
98            });
99        }
100        Ok(())
101    }
102}
103
104/// Persistent storage identity that rollback must preserve.
105#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
106#[serde(deny_unknown_fields, rename_all = "camelCase")]
107pub struct ReleaseCheckpointStorageIdentity {
108    /// Stable PersistentVolume or local-volume identity.
109    pub volume_id: String,
110    /// Store/WAL generation expected after restore.
111    pub wal_generation: u64,
112}
113
114impl ReleaseCheckpointStorageIdentity {
115    fn validate(&self) -> Result<(), ProtocolContractViolation> {
116        require_identifier("storageIdentity.volumeId", &self.volume_id)?;
117        if self.wal_generation == 0 {
118            return Err(ProtocolContractViolation::InvalidCheckpointField {
119                field: "storageIdentity.walGeneration",
120                reason: "must be greater than zero",
121            });
122        }
123        Ok(())
124    }
125}
126
127/// Shared identity and integrity fields for one checkpoint artifact.
128#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
129#[serde(deny_unknown_fields, rename_all = "camelCase")]
130pub struct ReleaseCheckpointArtifact {
131    /// Manifest schema.
132    pub schema_version: u16,
133    /// Globally unique checkpoint ID.
134    pub checkpoint_id: String,
135    /// Set containing this artifact.
136    pub checkpoint_set_id: String,
137    /// Storage/release generation.
138    pub generation: u64,
139    /// Shared write barrier ID.
140    pub barrier_id: String,
141    /// Artifact creation timestamp.
142    pub created_at_unix_millis: u64,
143    /// Uncompressed payload length.
144    pub length_bytes: u64,
145    /// Lowercase SHA-256 of the deterministic artifact payload.
146    pub sha256: String,
147    /// Durable artifact location.
148    pub uri: String,
149}
150
151impl ReleaseCheckpointArtifact {
152    /// Validates schema, identity, length, checksum, and URI.
153    ///
154    /// # Errors
155    ///
156    /// Returns a typed validation error when the artifact is incomplete.
157    pub fn validate(&self) -> Result<(), ProtocolContractViolation> {
158        if self.schema_version != RELEASE_CHECKPOINT_SCHEMA_VERSION {
159            return Err(ProtocolContractViolation::CheckpointSchemaVersion {
160                expected: RELEASE_CHECKPOINT_SCHEMA_VERSION,
161                actual: self.schema_version,
162            });
163        }
164        require_identifier("checkpointId", &self.checkpoint_id)?;
165        require_identifier("checkpointSetId", &self.checkpoint_set_id)?;
166        require_identifier("barrierId", &self.barrier_id)?;
167        if self.generation == 0 {
168            return Err(ProtocolContractViolation::InvalidCheckpointField {
169                field: "generation",
170                reason: "must be greater than zero",
171            });
172        }
173        if self.created_at_unix_millis == 0 {
174            return Err(ProtocolContractViolation::InvalidCheckpointField {
175                field: "createdAtUnixMillis",
176                reason: "must be greater than zero",
177            });
178        }
179        if self.length_bytes == 0 {
180            return Err(ProtocolContractViolation::InvalidCheckpointField {
181                field: "lengthBytes",
182                reason: "must be greater than zero",
183            });
184        }
185        require_sha256(&self.sha256)?;
186        if self.uri.trim().is_empty() || self.uri.bytes().any(|byte| matches!(byte, b'\r' | b'\n')) {
187            return Err(ProtocolContractViolation::InvalidCheckpointField {
188                field: "uri",
189                reason: "must be a non-empty single-line URI",
190            });
191        }
192        Ok(())
193    }
194}
195
196/// Request body for creating a Controller snapshot under a shared barrier.
197#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
198#[serde(deny_unknown_fields, rename_all = "camelCase")]
199pub struct ControllerReleaseSnapshotRequest {
200    pub checkpoint_id: String,
201    pub checkpoint_set_id: String,
202    pub generation: u64,
203    pub barrier_id: String,
204}
205
206impl ControllerReleaseSnapshotRequest {
207    /// Validates the checkpoint-set identity supplied to the Controller.
208    ///
209    /// # Errors
210    ///
211    /// Returns a typed validation error when an identifier is non-canonical or
212    /// the requested storage generation is zero.
213    pub fn validate(&self) -> Result<(), ProtocolContractViolation> {
214        require_identifier("checkpointId", &self.checkpoint_id)?;
215        require_identifier("checkpointSetId", &self.checkpoint_set_id)?;
216        require_identifier("barrierId", &self.barrier_id)?;
217        if self.generation == 0 {
218            return Err(ProtocolContractViolation::InvalidCheckpointField {
219                field: "generation",
220                reason: "must be greater than zero",
221            });
222        }
223        Ok(())
224    }
225}
226
227/// Checksummed Controller snapshot manifest.
228#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
229#[serde(deny_unknown_fields, rename_all = "camelCase")]
230pub struct ControllerReleaseSnapshotManifest {
231    /// Shared artifact identity and integrity metadata.
232    pub artifact: ReleaseCheckpointArtifact,
233    /// OpenRaft snapshot identity embedded in the checksummed payload.
234    pub snapshot_id: String,
235    /// Last applied Raft log index captured after ReadIndex.
236    pub last_applied_index: u64,
237    /// Raft term of the last applied entry.
238    pub last_applied_term: u64,
239    /// Voter node IDs captured in the snapshot membership.
240    pub voter_ids: Vec<u64>,
241}
242
243impl ControllerReleaseSnapshotManifest {
244    /// Validates Controller snapshot invariants.
245    ///
246    /// # Errors
247    ///
248    /// Returns a typed validation error for invalid artifact metadata, missing
249    /// applied state, or empty/duplicate voter membership.
250    pub fn validate(&self) -> Result<(), ProtocolContractViolation> {
251        self.artifact.validate()?;
252        require_identifier("snapshotId", &self.snapshot_id)?;
253        if self.last_applied_index == 0 || self.last_applied_term == 0 {
254            return Err(ProtocolContractViolation::InvalidCheckpointField {
255                field: "lastApplied",
256                reason: "term and index must be greater than zero",
257            });
258        }
259        let voters = self.voter_ids.iter().copied().collect::<BTreeSet<_>>();
260        if voters.is_empty() || voters.len() != self.voter_ids.len() || voters.contains(&0) {
261            return Err(ProtocolContractViolation::InvalidCheckpointField {
262                field: "voterIds",
263                reason: "must contain unique non-zero Controller voters",
264            });
265        }
266        Ok(())
267    }
268}
269
270/// Request body for creating one Store member checkpoint.
271#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
272#[serde(deny_unknown_fields, rename_all = "camelCase")]
273pub struct StoreReleaseCheckpointRequest {
274    pub checkpoint_id: String,
275    pub checkpoint_set_id: String,
276    pub generation: u64,
277    pub barrier_id: String,
278    pub member_id: String,
279    pub offsets: ReleaseCheckpointOffsets,
280    pub storage_identity: ReleaseCheckpointStorageIdentity,
281}
282
283impl StoreReleaseCheckpointRequest {
284    /// Validates the set binding and pre-flush Store observations.
285    ///
286    /// # Errors
287    ///
288    /// Returns a typed validation error when any required identity, generation,
289    /// offset, or storage identity is invalid.
290    pub fn validate(&self) -> Result<(), ProtocolContractViolation> {
291        require_identifier("checkpointId", &self.checkpoint_id)?;
292        require_identifier("checkpointSetId", &self.checkpoint_set_id)?;
293        require_identifier("barrierId", &self.barrier_id)?;
294        require_identifier("memberId", &self.member_id)?;
295        if self.generation == 0 {
296            return Err(ProtocolContractViolation::InvalidCheckpointField {
297                field: "generation",
298                reason: "must be greater than zero",
299            });
300        }
301        self.offsets.validate()?;
302        self.storage_identity.validate()
303    }
304}
305
306/// Checksummed Store checkpoint manifest.
307#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
308#[serde(deny_unknown_fields, rename_all = "camelCase")]
309pub struct StoreReleaseCheckpointManifest {
310    /// Shared artifact identity and integrity metadata.
311    pub artifact: ReleaseCheckpointArtifact,
312    /// Store member bound by the checkpoint set.
313    pub member_id: String,
314    /// Backend that created the artifact.
315    pub backend: ReleaseCheckpointBackend,
316    /// Offsets observed after all required flushes completed.
317    pub offsets: ReleaseCheckpointOffsets,
318    /// Persistent storage identity that rollback must retain.
319    pub storage_identity: ReleaseCheckpointStorageIdentity,
320    /// Explicit proof that WAL files remain part of the checkpoint contract.
321    pub wal_retained: bool,
322    /// Explicit proof that rollback must reuse the existing persistent volume.
323    pub persistent_volume_retained: bool,
324}
325
326impl StoreReleaseCheckpointManifest {
327    /// Validates Store checkpoint integrity and non-destructive rollback invariants.
328    ///
329    /// # Errors
330    ///
331    /// Returns a typed validation error for an incomplete artifact, invalid
332    /// offsets/storage identity, or a policy that permits WAL/PVC replacement.
333    pub fn validate(&self) -> Result<(), ProtocolContractViolation> {
334        self.artifact.validate()?;
335        require_identifier("memberId", &self.member_id)?;
336        self.offsets.validate()?;
337        self.storage_identity.validate()?;
338        if !self.wal_retained || !self.persistent_volume_retained {
339            return Err(ProtocolContractViolation::DestructiveRollback);
340        }
341        Ok(())
342    }
343}
344
345/// Complete checkpoint set that binds Controller and all Store members.
346#[derive(Clone, Deserialize, Eq, PartialEq, Serialize)]
347#[serde(deny_unknown_fields, rename_all = "camelCase")]
348pub struct ReleaseCheckpointSetManifest {
349    pub schema_version: u16,
350    pub checkpoint_set_id: String,
351    pub release_id: String,
352    pub generation: u64,
353    pub barrier_id: String,
354    pub policy_version: u64,
355    pub fencing_token: u64,
356    pub created_at_unix_millis: u64,
357    pub controller: ControllerReleaseSnapshotManifest,
358    pub stores: Vec<StoreReleaseCheckpointManifest>,
359}
360
361impl fmt::Debug for ReleaseCheckpointSetManifest {
362    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
363        formatter
364            .debug_struct("ReleaseCheckpointSetManifest")
365            .field("schema_version", &self.schema_version)
366            .field("checkpoint_set_id", &self.checkpoint_set_id)
367            .field("release_id", &self.release_id)
368            .field("generation", &self.generation)
369            .field("barrier_id", &self.barrier_id)
370            .field("policy_version", &self.policy_version)
371            .field("fencing_token", &Sensitive::new(self.fencing_token))
372            .field("created_at_unix_millis", &self.created_at_unix_millis)
373            .field("controller", &self.controller)
374            .field("stores", &self.stores)
375            .finish()
376    }
377}
378
379impl ReleaseCheckpointSetManifest {
380    /// Validates the complete Controller/Store barrier and member set.
381    ///
382    /// # Errors
383    ///
384    /// Returns a typed validation error when schema, release identity, policy
385    /// version, fencing, membership, or any cross-artifact binding differs.
386    pub fn validate(&self) -> Result<(), ProtocolContractViolation> {
387        if self.schema_version != RELEASE_CHECKPOINT_SCHEMA_VERSION {
388            return Err(ProtocolContractViolation::CheckpointSchemaVersion {
389                expected: RELEASE_CHECKPOINT_SCHEMA_VERSION,
390                actual: self.schema_version,
391            });
392        }
393        require_identifier("checkpointSetId", &self.checkpoint_set_id)?;
394        require_identifier("releaseId", &self.release_id)?;
395        require_identifier("barrierId", &self.barrier_id)?;
396        if self.generation == 0 || self.policy_version == 0 || self.fencing_token == 0 {
397            return Err(ProtocolContractViolation::InvalidCheckpointField {
398                field: "generation/policyVersion/fencingToken",
399                reason: "must be greater than zero",
400            });
401        }
402        if self.created_at_unix_millis == 0 {
403            return Err(ProtocolContractViolation::InvalidCheckpointField {
404                field: "createdAtUnixMillis",
405                reason: "must be greater than zero",
406            });
407        }
408        self.controller.validate()?;
409        if self.stores.is_empty() {
410            return Err(ProtocolContractViolation::MissingStoreMembers);
411        }
412
413        self.validate_binding("Controller", &self.controller.artifact)?;
414        let mut members = BTreeSet::new();
415        for store in &self.stores {
416            store.validate()?;
417            self.validate_binding("Store", &store.artifact)?;
418            if !members.insert(store.member_id.as_str()) {
419                return Err(ProtocolContractViolation::DuplicateStoreMember);
420            }
421        }
422        Ok(())
423    }
424
425    fn validate_binding(
426        &self,
427        member: &'static str,
428        artifact: &ReleaseCheckpointArtifact,
429    ) -> Result<(), ProtocolContractViolation> {
430        if artifact.checkpoint_set_id != self.checkpoint_set_id
431            || artifact.generation != self.generation
432            || artifact.barrier_id != self.barrier_id
433        {
434            return Err(ProtocolContractViolation::CheckpointSetBindingMismatch { member });
435        }
436        Ok(())
437    }
438}
439
440/// Result of non-destructive restore verification.
441#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
442#[serde(deny_unknown_fields, rename_all = "camelCase")]
443pub struct ReleaseCheckpointRestoreVerification {
444    pub checkpoint_id: String,
445    pub generation: u64,
446    pub verified_at_unix_millis: u64,
447    pub checksum_verified: bool,
448    pub offsets_verified: bool,
449    pub storage_identity_verified: bool,
450    pub wal_retained: bool,
451    pub persistent_volume_retained: bool,
452}
453
454impl ReleaseCheckpointRestoreVerification {
455    /// Validates that restore verification proved every production invariant.
456    ///
457    /// # Errors
458    ///
459    /// Returns a typed validation error when any integrity, offset, WAL, or PVC
460    /// proof is absent.
461    pub fn validate(&self) -> Result<(), ProtocolContractViolation> {
462        require_identifier("checkpointId", &self.checkpoint_id)?;
463        if self.generation == 0 || self.verified_at_unix_millis == 0 {
464            return Err(ProtocolContractViolation::InvalidCheckpointField {
465                field: "generation/verifiedAtUnixMillis",
466                reason: "must be greater than zero",
467            });
468        }
469        if !self.checksum_verified
470            || !self.offsets_verified
471            || !self.storage_identity_verified
472            || !self.wal_retained
473            || !self.persistent_volume_retained
474        {
475            return Err(ProtocolContractViolation::RestoreVerificationIncomplete);
476        }
477        Ok(())
478    }
479}
480
481fn require_identifier(field: &'static str, value: &str) -> Result<(), ProtocolContractViolation> {
482    if value.is_empty()
483        || value.len() > 256
484        || !value
485            .bytes()
486            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-' | b':' | b'/' | b'@'))
487    {
488        return Err(ProtocolContractViolation::InvalidCheckpointField {
489            field,
490            reason: "must be a canonical identifier",
491        });
492    }
493    Ok(())
494}
495
496fn require_sha256(value: &str) -> Result<(), ProtocolContractViolation> {
497    if value.len() != 64
498        || !value
499            .bytes()
500            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
501    {
502        return Err(ProtocolContractViolation::InvalidCheckpointField {
503            field: "sha256",
504            reason: "must be 64 lowercase hexadecimal characters",
505        });
506    }
507    Ok(())
508}
509
510#[cfg(test)]
511mod tests {
512    use super::*;
513
514    fn artifact(id: &str) -> ReleaseCheckpointArtifact {
515        ReleaseCheckpointArtifact {
516            schema_version: 1,
517            checkpoint_id: id.to_string(),
518            checkpoint_set_id: "set-7".to_string(),
519            generation: 7,
520            barrier_id: "barrier-42".to_string(),
521            created_at_unix_millis: 1_800_000_000_000,
522            length_bytes: 512,
523            sha256: "a".repeat(64),
524            uri: format!("file:///checkpoints/{id}"),
525        }
526    }
527
528    fn store(member: &str) -> StoreReleaseCheckpointManifest {
529        StoreReleaseCheckpointManifest {
530            artifact: artifact(&format!("store-{member}")),
531            member_id: member.to_string(),
532            backend: ReleaseCheckpointBackend::Local,
533            offsets: ReleaseCheckpointOffsets {
534                appended_offset: 120,
535                durable_offset: 120,
536                consume_queue_offset: 100,
537                index_offset: 100,
538            },
539            storage_identity: ReleaseCheckpointStorageIdentity {
540                volume_id: format!("pvc-{member}"),
541                wal_generation: 7,
542            },
543            wal_retained: true,
544            persistent_volume_retained: true,
545        }
546    }
547
548    #[test]
549    fn release_checkpoint_set_binds_controller_stores_and_offsets() {
550        let manifest = ReleaseCheckpointSetManifest {
551            schema_version: 1,
552            checkpoint_set_id: "set-7".to_string(),
553            release_id: "release-7".to_string(),
554            generation: 7,
555            barrier_id: "barrier-42".to_string(),
556            policy_version: 3,
557            fencing_token: 42,
558            created_at_unix_millis: 1_800_000_000_000,
559            controller: ControllerReleaseSnapshotManifest {
560                artifact: artifact("controller"),
561                snapshot_id: "snapshot-99".to_string(),
562                last_applied_index: 99,
563                last_applied_term: 3,
564                voter_ids: vec![1, 2, 3],
565            },
566            stores: vec![store("broker-a"), store("broker-b")],
567        };
568
569        manifest.validate().expect("complete set should validate");
570    }
571
572    #[test]
573    fn release_checkpoint_set_rejects_barrier_drift_and_destructive_restore() {
574        let mut store = store("broker-a");
575        store.artifact.barrier_id = "other-barrier".to_string();
576        store.wal_retained = false;
577        let manifest = ReleaseCheckpointSetManifest {
578            schema_version: 1,
579            checkpoint_set_id: "set-7".to_string(),
580            release_id: "release-7".to_string(),
581            generation: 7,
582            barrier_id: "barrier-42".to_string(),
583            policy_version: 3,
584            fencing_token: 42,
585            created_at_unix_millis: 1_800_000_000_000,
586            controller: ControllerReleaseSnapshotManifest {
587                artifact: artifact("controller"),
588                snapshot_id: "snapshot-99".to_string(),
589                last_applied_index: 99,
590                last_applied_term: 3,
591                voter_ids: vec![1],
592            },
593            stores: vec![store],
594        };
595
596        assert!(manifest.validate().is_err());
597    }
598
599    #[test]
600    fn release_checkpoint_set_debug_redacts_fencing_token() {
601        let manifest = ReleaseCheckpointSetManifest {
602            schema_version: 1,
603            checkpoint_set_id: "set-7".to_string(),
604            release_id: "release-7".to_string(),
605            generation: 7,
606            barrier_id: "barrier-42".to_string(),
607            policy_version: 3,
608            fencing_token: 987_654_321,
609            created_at_unix_millis: 1_800_000_000_000,
610            controller: ControllerReleaseSnapshotManifest {
611                artifact: artifact("controller"),
612                snapshot_id: "snapshot-99".to_string(),
613                last_applied_index: 99,
614                last_applied_term: 3,
615                voter_ids: vec![1],
616            },
617            stores: vec![store("broker-a")],
618        };
619
620        let debug = format!("{manifest:?}");
621
622        assert!(debug.contains("<redacted>"));
623        assert!(!debug.contains("987654321"));
624    }
625}
rocketmq_protocol/protocol/body/release_checkpoint.rs

rocketmq_protocol/protocol/body/
release_checkpoint.rs
