chorus_client/
manifest_store.rs

1//! Pluggable storage for the manifest control register.
2//!
3//! The register is one tiny versioned document holding the WAL's control
4//! fields (writer epoch, tail and seal records, truncation floor). Every
5//! protocol decision is a compare-and-swap on it, so the backend must be a
6//! linearizable CAS register — nothing more. The default implementation
7//! stores the register as object metadata in a regional GCS bucket; any
8//! system offering strongly consistent reads and an atomic conditional
9//! update (Firestore, Spanner, a SQL row with optimistic locking) can stand
10//! in through [`ManifestStore`].
11
12use std::collections::HashMap;
13use std::sync::Arc;
14
15use async_trait::async_trait;
16
17use crate::transport::{Replica, ReplicaSnapshot, TransportCode};
18
19/// Opaque optimistic-concurrency token for one observed register state.
20///
21/// A backend supplies whatever counter expresses "the register has not
22/// changed since this read" — the GCS implementation uses the object
23/// metageneration (the register is never deleted or recreated, so its
24/// generation is constant), a SQL implementation might use a row version.
25/// Callers only thread the token from a read into the conditional update
26/// that follows it.
27#[derive(Clone, Copy, Debug, Eq, PartialEq)]
28pub struct ManifestVersion(pub u64);
29
30/// One consistent read of the register: its fields and the version token
31/// guarding the next conditional update.
32#[derive(Clone, Debug, Eq, PartialEq)]
33pub struct VersionedManifest {
34    /// Token for the conditional update that follows this read.
35    pub version: ManifestVersion,
36    /// The register's string fields (the `chorus.*` keys).
37    pub fields: HashMap<String, String>,
38}
39
40/// Why a register operation did not produce a new state.
41#[derive(Clone, Debug, thiserror::Error)]
42pub enum ManifestStoreError {
43    /// The register changed since the witnessed version; re-read and decide
44    /// again.
45    #[error("the manifest register changed since it was read")]
46    Conflict,
47    /// A concurrent creator initialized the register first; re-read.
48    #[error("the manifest register already exists")]
49    AlreadyExists,
50    /// Transient backend trouble; the operation may be retried as-is.
51    #[error("manifest store unavailable: {0}")]
52    Unavailable(String),
53    /// Terminal backend failure.
54    #[error("manifest store: {0}")]
55    Backend(String),
56}
57
58/// A linearizable compare-and-swap register holding the manifest.
59///
60/// Required semantics, on which every protocol decision rests:
61///
62/// - [`read`](Self::read) is strongly consistent: it observes every update
63///   that completed before it started.
64/// - [`create`](Self::create) initializes the register exactly once;
65///   concurrent creators lose with [`ManifestStoreError::AlreadyExists`].
66/// - [`update`](Self::update) atomically replaces the fields if and only if
67///   the register still carries the witnessed version, and fails with
68///   [`ManifestStoreError::Conflict`] otherwise. Updates are totally
69///   ordered.
70///
71/// An update whose response is lost may have applied; callers re-read and
72/// re-decide, so implementations must not retry internally in ways that
73/// could apply one logical update twice under a stale version.
74///
75/// A Firestore document, Spanner row, or SQL row is the intended extension
76/// point when an operator wants to move beyond GCS object-metadata constraints.
77/// Each backend reports its own sealed-directory byte budget, so a
78/// higher-capacity store can retain more directory entries than GCS metadata.
79#[async_trait]
80pub trait ManifestStore: Send + Sync {
81    /// Maximum encoded byte length accepted for the `chorus.segments` value.
82    ///
83    /// This is a backend capacity property, not a protocol constant. The
84    /// client reserves worst-case room before committing another directory
85    /// entry and returns `SegmentDirectoryFull` rather than exceeding it.
86    fn max_directory_bytes(&self) -> usize;
87
88    /// Read the register; `None` means it has never been created.
89    async fn read(&self) -> Result<Option<VersionedManifest>, ManifestStoreError>;
90
91    /// Create the register with `fields` if it does not exist.
92    async fn create(
93        &self,
94        fields: HashMap<String, String>,
95    ) -> Result<VersionedManifest, ManifestStoreError>;
96
97    /// Replace the fields if the register still carries `version`.
98    async fn update(
99        &self,
100        version: ManifestVersion,
101        fields: HashMap<String, String>,
102    ) -> Result<VersionedManifest, ManifestStoreError>;
103}
104
105/// The default register backend: object metadata on one regional GCS
106/// object, guarded by a metageneration precondition.
107pub(crate) struct GcsManifestStore {
108    replica: Arc<dyn Replica>,
109}
110
111/// GCS caps all custom object metadata at roughly 8 KiB. The fixed manifest
112/// fields stay below 500 bytes, leaving this conservative budget for the
113/// encoded sealed-segment directory.
114pub(crate) const GCS_MAX_DIRECTORY_BYTES: usize = 6144;
115
116impl GcsManifestStore {
117    pub(crate) fn new(replica: Arc<dyn Replica>) -> Self {
118        Self { replica }
119    }
120
121    fn versioned(snapshot: ReplicaSnapshot) -> VersionedManifest {
122        VersionedManifest {
123            version: ManifestVersion(snapshot.metageneration as u64),
124            fields: snapshot.metadata,
125        }
126    }
127}
128
129fn store_error(error: crate::transport::TransportError) -> ManifestStoreError {
130    match error.code {
131        TransportCode::FailedPrecondition => ManifestStoreError::Conflict,
132        TransportCode::AlreadyExists => ManifestStoreError::AlreadyExists,
133        code if code.transient() => ManifestStoreError::Unavailable(error.to_string()),
134        _ => ManifestStoreError::Backend(error.to_string()),
135    }
136}
137
138#[async_trait]
139impl ManifestStore for GcsManifestStore {
140    fn max_directory_bytes(&self) -> usize {
141        GCS_MAX_DIRECTORY_BYTES
142    }
143
144    async fn read(&self) -> Result<Option<VersionedManifest>, ManifestStoreError> {
145        // The register body is permanently empty; all state and CAS
146        // preconditions live in object metadata, so a stat is the read.
147        match self.replica.stat().await {
148            Ok(snapshot) => Ok(Some(Self::versioned(snapshot))),
149            Err(error) if error.code == TransportCode::NotFound => Ok(None),
150            Err(error) => Err(store_error(error)),
151        }
152    }
153
154    async fn create(
155        &self,
156        fields: HashMap<String, String>,
157    ) -> Result<VersionedManifest, ManifestStoreError> {
158        match self.replica.create_register(fields).await {
159            Ok(snapshot) => Ok(Self::versioned(snapshot)),
160            Err(error) => Err(store_error(error)),
161        }
162    }
163
164    async fn update(
165        &self,
166        version: ManifestVersion,
167        fields: HashMap<String, String>,
168    ) -> Result<VersionedManifest, ManifestStoreError> {
169        match self.replica.update_register(version.0 as i64, fields).await {
170            Ok(snapshot) => Ok(Self::versioned(snapshot)),
171            Err(error) => Err(store_error(error)),
172        }
173    }
174}
175
176#[cfg(test)]
177pub(crate) mod test_support {
178    use std::collections::HashMap;
179    use std::sync::Mutex;
180
181    use async_trait::async_trait;
182
183    use super::{ManifestStore, ManifestStoreError, ManifestVersion, VersionedManifest};
184
185    /// A process-local register with exact CAS semantics, standing in for a
186    /// Firestore/Spanner/SQL backend in tests.
187    #[derive(Default)]
188    pub(crate) struct InMemoryManifestStore {
189        state: Mutex<Option<(u64, HashMap<String, String>)>>,
190    }
191
192    #[async_trait]
193    impl ManifestStore for InMemoryManifestStore {
194        fn max_directory_bytes(&self) -> usize {
195            64 * 1024
196        }
197
198        async fn read(&self) -> Result<Option<VersionedManifest>, ManifestStoreError> {
199            Ok(self
200                .state
201                .lock()
202                .unwrap()
203                .as_ref()
204                .map(|(version, fields)| VersionedManifest {
205                    version: ManifestVersion(*version),
206                    fields: fields.clone(),
207                }))
208        }
209
210        async fn create(
211            &self,
212            fields: HashMap<String, String>,
213        ) -> Result<VersionedManifest, ManifestStoreError> {
214            let mut state = self.state.lock().unwrap();
215            if state.is_some() {
216                return Err(ManifestStoreError::AlreadyExists);
217            }
218            *state = Some((1, fields.clone()));
219            Ok(VersionedManifest {
220                version: ManifestVersion(1),
221                fields,
222            })
223        }
224
225        async fn update(
226            &self,
227            version: ManifestVersion,
228            fields: HashMap<String, String>,
229        ) -> Result<VersionedManifest, ManifestStoreError> {
230            let mut state = self.state.lock().unwrap();
231            let Some((current, stored)) = state.as_mut() else {
232                return Err(ManifestStoreError::Backend(
233                    "the register was never created".into(),
234                ));
235            };
236            if ManifestVersion(*current) != version {
237                return Err(ManifestStoreError::Conflict);
238            }
239            *current += 1;
240            *stored = fields.clone();
241            Ok(VersionedManifest {
242                version: ManifestVersion(*current),
243                fields,
244            })
245        }
246    }
247
248    #[tokio::test]
249    async fn in_memory_store_enforces_cas_semantics() {
250        let store = InMemoryManifestStore::default();
251        assert!(store.read().await.unwrap().is_none());
252        let initial = store
253            .create(HashMap::from([("k".to_string(), "1".to_string())]))
254            .await
255            .unwrap();
256        assert!(matches!(
257            store.create(HashMap::new()).await,
258            Err(ManifestStoreError::AlreadyExists)
259        ));
260        let updated = store
261            .update(
262                initial.version,
263                HashMap::from([("k".to_string(), "2".to_string())]),
264            )
265            .await
266            .unwrap();
267        assert!(matches!(
268            store.update(initial.version, HashMap::new()).await,
269            Err(ManifestStoreError::Conflict)
270        ));
271        assert_eq!(
272            store.read().await.unwrap().unwrap().version,
273            updated.version
274        );
275    }
276}
chorus_client/manifest_store.rs

chorus_client/
manifest_store.rs
