loonfs_core/namespace/
control.rs

1//! Typed loaders for the namespace's control objects: head, metadata root,
2//! and WAL floor.
3
4use crate::error::StoreFailureClass;
5use crate::namespace::basis::{read_head_and_metadata_basis, MetadataBasis};
6use loonfs_api::wire::control::{
7    decode_control_object, ControlObjectKind, HeadState, HeadStateEnvelope, MetadataRootEnvelope,
8    MetadataRootState, WalFloorEnvelope,
9};
10use loonfs_api::wire::envelope::EnvelopeCodecError;
11use loonfs_api::NamespaceId;
12use loonfs_objectstore::keys::{metadata_root, wal_floor, wal_head};
13use loonfs_objectstore::ObjectStoreError;
14use loonfs_objectstore::{ObjectMetadata, ObjectStore};
15use serde::{Deserialize, Serialize};
16use thiserror::Error;
17
18#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
19pub(crate) struct LoadedHeadObject {
20    pub(crate) object_key: String,
21    pub(crate) metadata: ObjectMetadata,
22    pub(crate) envelope: HeadStateEnvelope,
23}
24
25#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
26pub(crate) struct LoadedMetadataRootObject {
27    pub(crate) object_key: String,
28    pub(crate) metadata: ObjectMetadata,
29    pub(crate) envelope: MetadataRootEnvelope,
30}
31
32#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
33pub(crate) struct LoadedWalFloorObject {
34    pub(crate) object_key: String,
35    pub(crate) metadata: ObjectMetadata,
36    pub(crate) envelope: WalFloorEnvelope,
37}
38
39#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
40pub struct ControlObjectIdentity {
41    pub etag: String,
42}
43
44#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
45pub struct LoadedMetadataRootControl {
46    pub object_key: String,
47    pub identity: ControlObjectIdentity,
48    pub state: MetadataRootState,
49}
50
51#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
52pub struct LoadedWalFloorControl {
53    pub object_key: String,
54    pub identity: ControlObjectIdentity,
55    pub state: loonfs_api::wire::control::WalFloorState,
56}
57
58#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
59pub struct LoadedHeadControl {
60    pub object_key: String,
61    pub identity: ControlObjectIdentity,
62    pub state: HeadState,
63}
64
65#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Error)]
66pub enum ControlObjectLoadError {
67    #[error("missing control object `{object_key}`")]
68    MissingObject { object_key: String },
69    #[error(
70        "metadata root references seq `{root_manifest_head_seq}` beyond the reloaded head seq `{head_seq}`"
71    )]
72    RootAheadOfHead {
73        root_manifest_head_seq: loonfs_api::ChangeSeq,
74        head_seq: loonfs_api::ChangeSeq,
75    },
76    #[error(
77        "control object namespace mismatch for `{object_key}`: expected `{expected}`, actual `{actual}`"
78    )]
79    NamespaceMismatch {
80        object_key: String,
81        expected: NamespaceId,
82        actual: NamespaceId,
83    },
84    #[error(
85        "control object checksum mismatch for `{object_key}`: expected `{expected}`, actual `{actual}`"
86    )]
87    ChecksumMismatch {
88        object_key: String,
89        expected: String,
90        actual: String,
91    },
92    #[error("control object codec error for `{object_key}`: {message}")]
93    Codec { object_key: String, message: String },
94    #[error("control object store error for `{object_key}`: {message}")]
95    Store {
96        object_key: String,
97        message: String,
98        class: StoreFailureClass,
99    },
100}
101
102pub(crate) async fn read_wal_floor_object<S: ObjectStore + ?Sized>(
103    store: &S,
104    expected_namespace_id: &NamespaceId,
105) -> Result<LoadedWalFloorObject, ControlObjectLoadError> {
106    let object_key = wal_floor(expected_namespace_id.as_str());
107    let (metadata, encoded_bytes) = read_control_object_bytes(store, &object_key).await?;
108    let envelope: WalFloorEnvelope =
109        decode_control_object(&encoded_bytes, ControlObjectKind::WalFloor)
110            .map_err(|err| map_control_codec_error(&object_key, err))?;
111    validate_expected_namespace(
112        &object_key,
113        expected_namespace_id,
114        &envelope.state.namespace_id,
115    )?;
116
117    Ok(LoadedWalFloorObject {
118        object_key,
119        metadata,
120        envelope,
121    })
122}
123
124pub(crate) async fn read_metadata_root_object<S: ObjectStore + ?Sized>(
125    store: &S,
126    expected_namespace_id: &NamespaceId,
127) -> Result<LoadedMetadataRootObject, ControlObjectLoadError> {
128    let object_key = metadata_root(expected_namespace_id.as_str());
129    let (metadata, encoded_bytes) = read_control_object_bytes(store, &object_key).await?;
130    let envelope: MetadataRootEnvelope =
131        decode_control_object(&encoded_bytes, ControlObjectKind::MetadataRoot)
132            .map_err(|err| map_control_codec_error(&object_key, err))?;
133    validate_expected_namespace(
134        &object_key,
135        expected_namespace_id,
136        &envelope.state.namespace_id,
137    )?;
138
139    Ok(LoadedMetadataRootObject {
140        object_key,
141        metadata,
142        envelope,
143    })
144}
145
146pub(crate) async fn read_metadata_root_object_if_present<S: ObjectStore + ?Sized>(
147    store: &S,
148    expected_namespace_id: &NamespaceId,
149) -> Result<Option<LoadedMetadataRootObject>, ControlObjectLoadError> {
150    match read_metadata_root_object(store, expected_namespace_id).await {
151        Ok(loaded) => Ok(Some(loaded)),
152        Err(ControlObjectLoadError::MissingObject { .. }) => Ok(None),
153        Err(error) => Err(error),
154    }
155}
156
157/// Reads the WAL head and metadata root together (concurrently), tolerating
158/// a namespace that has not materialized a root yet.
159///
160/// The root is published by the first flush, so its absence is ordinary for
161/// a young namespace and the caller resolves the basis from the head
162/// instead. When the root is there it can only ever reference published
163/// state, so observing `root.manifest_head_seq > head.seq` means the head
164/// read raced an in-flight commit CAS; a fresh head read observes at least
165/// the root's seq. Bounded reloads resolve the race without treating it as
166/// corruption (format spec, "metadata/root.json").
167pub(crate) async fn read_head_and_metadata_root_if_present<S: ObjectStore + ?Sized>(
168    store: &S,
169    expected_namespace_id: &NamespaceId,
170) -> Result<(LoadedHeadObject, Option<LoadedMetadataRootObject>), ControlObjectLoadError> {
171    const ROOT_AHEAD_HEAD_RELOADS: usize = 3;
172    let (head, root) = futures::join!(
173        read_head_object(store, expected_namespace_id),
174        read_metadata_root_object_if_present(store, expected_namespace_id)
175    );
176    let mut head = head?;
177    let Some(root) = root? else {
178        return Ok((head, None));
179    };
180    for _reload in 0..=ROOT_AHEAD_HEAD_RELOADS {
181        if root.envelope.state.manifest_head_seq <= head.envelope.state.seq {
182            return Ok((head, Some(root)));
183        }
184        head = read_head_object(store, expected_namespace_id).await?;
185    }
186    Err(ControlObjectLoadError::RootAheadOfHead {
187        root_manifest_head_seq: root.envelope.state.manifest_head_seq,
188        head_seq: head.envelope.state.seq,
189    })
190}
191
192pub(crate) async fn read_head_object<S: ObjectStore + ?Sized>(
193    store: &S,
194    expected_namespace_id: &NamespaceId,
195) -> Result<LoadedHeadObject, ControlObjectLoadError> {
196    let object_key = wal_head(expected_namespace_id.as_str());
197    let (metadata, encoded_bytes) = read_control_object_bytes(store, &object_key).await?;
198    let envelope: HeadStateEnvelope =
199        decode_control_object(&encoded_bytes, ControlObjectKind::WalHead)
200            .map_err(|err| map_control_codec_error(&object_key, err))?;
201    validate_expected_namespace(
202        &object_key,
203        expected_namespace_id,
204        &envelope.state.namespace_id,
205    )?;
206
207    Ok(LoadedHeadObject {
208        object_key,
209        metadata,
210        envelope,
211    })
212}
213
214pub async fn load_namespace_wal_floor_control<S: ObjectStore + ?Sized>(
215    store: &S,
216    expected_namespace_id: &NamespaceId,
217) -> Result<LoadedWalFloorControl, ControlObjectLoadError> {
218    let loaded = read_wal_floor_object(store, expected_namespace_id).await?;
219    let identity = control_identity(&loaded.object_key, &loaded.metadata)?;
220    Ok(LoadedWalFloorControl {
221        object_key: loaded.object_key,
222        identity,
223        state: loaded.envelope.state,
224    })
225}
226
227pub async fn load_namespace_checkpoint_record_control<S: ObjectStore + ?Sized>(
228    store: &S,
229    expected_namespace_id: &NamespaceId,
230    checkpoint_id: &loonfs_api::CheckpointId,
231) -> Result<Option<loonfs_api::wire::control::CheckpointRecordState>, crate::error::CoreError> {
232    Ok(
233        crate::checkpoint::read_checkpoint_record(store, expected_namespace_id, checkpoint_id)
234            .await?
235            .map(|loaded| loaded.state),
236    )
237}
238
239pub async fn load_namespace_metadata_root_control<S: ObjectStore + ?Sized>(
240    store: &S,
241    expected_namespace_id: &NamespaceId,
242) -> Result<LoadedMetadataRootControl, ControlObjectLoadError> {
243    let loaded = read_metadata_root_object(store, expected_namespace_id).await?;
244    let identity = control_identity(&loaded.object_key, &loaded.metadata)?;
245    Ok(LoadedMetadataRootControl {
246        object_key: loaded.object_key,
247        identity,
248        state: loaded.envelope.state,
249    })
250}
251
252/// Loads the head together with the metadata basis it authorizes, as one
253/// consistent read anchor (see [`read_head_and_metadata_root_if_present`]
254/// for the race rule).
255pub async fn load_namespace_read_anchor<S: ObjectStore + ?Sized>(
256    store: &S,
257    expected_namespace_id: &NamespaceId,
258) -> Result<(LoadedHeadControl, MetadataBasis), ControlObjectLoadError> {
259    let loaded = read_head_and_metadata_basis(store, expected_namespace_id).await?;
260    let head_identity = control_identity(&loaded.head.object_key, &loaded.head.metadata)?;
261    Ok((
262        LoadedHeadControl {
263            object_key: loaded.head.object_key,
264            identity: head_identity,
265            state: loaded.head.envelope.state,
266        },
267        loaded.basis,
268    ))
269}
270
271pub async fn load_namespace_head_control<S: ObjectStore + ?Sized>(
272    store: &S,
273    expected_namespace_id: &NamespaceId,
274) -> Result<LoadedHeadControl, ControlObjectLoadError> {
275    let loaded = read_head_object(store, expected_namespace_id).await?;
276    let identity = control_identity(&loaded.object_key, &loaded.metadata)?;
277    Ok(LoadedHeadControl {
278        object_key: loaded.object_key,
279        identity,
280        state: loaded.envelope.state,
281    })
282}
283
284fn control_identity(
285    object_key: &str,
286    metadata: &ObjectMetadata,
287) -> Result<ControlObjectIdentity, ControlObjectLoadError> {
288    let etag = metadata
289        .etag
290        .clone()
291        .ok_or_else(|| ControlObjectLoadError::Store {
292            object_key: object_key.to_owned(),
293            message: "missing control object etag".to_owned(),
294            class: StoreFailureClass::Other,
295        })?;
296    Ok(ControlObjectIdentity { etag })
297}
298
299async fn read_control_object_bytes<S: ObjectStore + ?Sized>(
300    store: &S,
301    object_key: &str,
302) -> Result<(ObjectMetadata, Vec<u8>), ControlObjectLoadError> {
303    let body = store
304        .get_with_metadata(object_key)
305        .await
306        .map_err(|err| map_store_load_error(object_key, err))?
307        .ok_or_else(|| ControlObjectLoadError::MissingObject {
308            object_key: object_key.to_owned(),
309        })?;
310    Ok((body.metadata, body.bytes))
311}
312
313fn validate_expected_namespace(
314    object_key: &str,
315    expected: &NamespaceId,
316    actual: &NamespaceId,
317) -> Result<(), ControlObjectLoadError> {
318    if actual != expected {
319        return Err(ControlObjectLoadError::NamespaceMismatch {
320            object_key: object_key.to_owned(),
321            expected: expected.clone(),
322            actual: actual.clone(),
323        });
324    }
325    Ok(())
326}
327
328pub(crate) fn map_control_codec_error(
329    object_key: &str,
330    err: EnvelopeCodecError,
331) -> ControlObjectLoadError {
332    match err {
333        EnvelopeCodecError::ChecksumMismatch { expected, actual } => {
334            ControlObjectLoadError::ChecksumMismatch {
335                object_key: object_key.to_owned(),
336                expected,
337                actual,
338            }
339        }
340        other => ControlObjectLoadError::Codec {
341            object_key: object_key.to_owned(),
342            message: other.to_string(),
343        },
344    }
345}
346
347fn map_store_load_error(object_key: &str, err: ObjectStoreError) -> ControlObjectLoadError {
348    ControlObjectLoadError::Store {
349        object_key: object_key.to_owned(),
350        message: err.message(),
351        class: StoreFailureClass::of(&err),
352    }
353}
loonfs_core/namespace/control.rs

loonfs_core/namespace/
control.rs
