hidane_core/store/
mod.rs

1//! The storage boundary (ADR 0002).
2//!
3//! Everything above this module (gRPC, REST, Listen, transactions) talks to [`Store`]; the
4//! engine behind it can change without touching them. [`MemoryStore`] is the engine for now.
5//!
6//! The model:
7//!
8//! - Documents are **versioned by commit time** (microseconds, unique and increasing per
9//!   database). A read names a [`ReadTime`] and sees the newest version at or before it, which
10//!   gives consistent snapshots for queries, transactions and Listen.
11//! - A commit runs a closure against a [`WriteBatch`]: the commit time is fixed before the
12//!   closure runs (server timestamps need it) and reads inside the batch see the batch's own
13//!   writes. The closure's error aborts the commit with nothing applied.
14//! - A commit returns every changed document with its old and new version, so change
15//!   notification costs O(changed documents), never O(stored documents).
16
17mod memory;
18
19use std::{fmt, ops::ControlFlow, sync::Arc};
20
21use hidane_proto::google::firestore::v1::{ArrayValue, MapValue, Value, value::ValueType};
22use hidane_proto::ordered::Fields;
23use prost::Message as _;
24use prost_types::Timestamp;
25
26pub use memory::MemoryStore;
27
28use crate::path::ResourcePath;
29
30/// A point in a database's history, in microseconds since the Unix epoch. Commit times are
31/// read times; reading at a commit time sees that commit.
32#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
33pub struct ReadTime(pub i64);
34
35impl ReadTime {
36    pub const MAX: Self = Self(i64::MAX);
37
38    /// Truncates to microseconds, as Firestore does.
39    pub fn from_timestamp(ts: &Timestamp) -> Self {
40        Self(ts.seconds * 1_000_000 + i64::from(ts.nanos) / 1_000)
41    }
42
43    pub fn to_timestamp(self) -> Timestamp {
44        #[allow(clippy::cast_possible_truncation)]
45        Timestamp {
46            seconds: self.0.div_euclid(1_000_000),
47            nanos: (self.0.rem_euclid(1_000_000) * 1_000) as i32,
48        }
49    }
50}
51
52/// A stored document version.
53///
54/// Fields are kept protobuf-encoded: a document then costs about its wire size instead of the
55/// decoded tree (a two-field document is ~30 bytes encoded, over 600 decoded, because every
56/// map node reserves room for eleven entries). Readers decode with [`StoredDocument::fields`].
57#[derive(Debug, Clone, PartialEq)]
58pub struct StoredDocument {
59    /// Shared by every version of the same document.
60    pub path: Arc<ResourcePath>,
61    encoded_fields: Box<[u8]>,
62    pub create_time: Timestamp,
63    pub update_time: Timestamp,
64}
65
66impl StoredDocument {
67    pub fn new(
68        path: Arc<ResourcePath>,
69        fields: Fields,
70        create_time: Timestamp,
71        update_time: Timestamp,
72    ) -> Self {
73        Self {
74            path,
75            encoded_fields: MapValue { fields }.encode_to_vec().into_boxed_slice(),
76            create_time,
77            update_time,
78        }
79    }
80
81    /// The document's fields, decoded.
82    pub fn fields(&self) -> Fields {
83        MapValue::decode(&*self.encoded_fields)
84            .expect("fields were encoded by StoredDocument::new")
85            .fields
86    }
87
88    /// Size of the stored field encoding, in bytes.
89    pub fn encoded_len(&self) -> usize {
90        self.encoded_fields.len()
91    }
92
93    /// Whether the document holds exactly `fields`, bit for bit: unlike `==` on the decoded
94    /// fields, a NaN equals itself and `-0.0` differs from `0.0`, as for the official emulator,
95    /// to which writing either over itself changes nothing (`tools/oracle/negative_zero.py`).
96    /// The order of the fields does not matter.
97    pub fn has_fields(&self, fields: &Fields) -> bool {
98        let canonical = |fields: &Fields| {
99            MapValue {
100                fields: canonical(fields),
101            }
102            .encode_to_vec()
103        };
104        fields.len() == self.fields().len() && canonical(fields) == canonical(&self.fields())
105    }
106}
107
108/// Whether `a` and `b` are the same value, bit for bit, as [`StoredDocument::has_fields`]
109/// compares documents: a NaN equals itself, `-0.0` differs from `0.0`, and the order of a map's
110/// fields does not matter.
111pub fn identical(a: &Value, b: &Value) -> bool {
112    canonical_value(a).encode_to_vec() == canonical_value(b).encode_to_vec()
113}
114
115/// `fields` with every map, nested ones included, in UTF-8 key order.
116fn canonical(fields: &Fields) -> Fields {
117    fields
118        .sorted()
119        .into_iter()
120        .map(|(key, value)| (key.clone(), canonical_value(value)))
121        .collect()
122}
123
124fn canonical_value(value: &Value) -> Value {
125    let value_type = match &value.value_type {
126        Some(ValueType::MapValue(map)) => ValueType::MapValue(MapValue {
127            fields: canonical(&map.fields),
128        }),
129        Some(ValueType::ArrayValue(array)) => ValueType::ArrayValue(ArrayValue {
130            values: array.values.iter().map(canonical_value).collect(),
131        }),
132        _ => return value.clone(),
133    };
134    Value {
135        value_type: Some(value_type),
136    }
137}
138
139/// One document touched by a commit.
140#[derive(Debug, Clone)]
141pub struct Change {
142    pub path: Arc<ResourcePath>,
143    pub before: Option<Arc<StoredDocument>>,
144    pub after: Option<Arc<StoredDocument>>,
145}
146
147/// The result of a successful commit.
148#[derive(Debug, Clone)]
149pub struct Commit {
150    pub commit_time: ReadTime,
151    pub changes: Vec<Change>,
152}
153
154/// Why a commit was refused. Maps one to one onto gRPC status codes.
155#[derive(Debug, Clone, PartialEq, Eq)]
156pub enum StoreError {
157    InvalidArgument(String),
158    NotFound(String),
159    AlreadyExists(String),
160    FailedPrecondition(String),
161    Aborted(String),
162}
163
164impl fmt::Display for StoreError {
165    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
166        match self {
167            Self::InvalidArgument(m)
168            | Self::NotFound(m)
169            | Self::AlreadyExists(m)
170            | Self::FailedPrecondition(m)
171            | Self::Aborted(m) => f.write_str(m),
172        }
173    }
174}
175
176impl std::error::Error for StoreError {}
177
178/// The writes of one commit, with read-your-writes visibility.
179pub trait WriteBatch {
180    /// The time this commit will be stamped with (and the documents' `update_time`).
181    fn commit_time(&self) -> ReadTime;
182
183    /// The document as it will be if the commit succeeds now: the batch's own writes first,
184    /// then the latest committed version.
185    fn get(&self, path: &ResourcePath) -> Option<Arc<StoredDocument>>;
186
187    /// Replaces the document's fields. Keeps `create_time` if the document exists.
188    fn set(&mut self, path: &ResourcePath, fields: Fields);
189
190    /// Deletes the document. Deleting a missing document is not an error.
191    fn delete(&mut self, path: &ResourcePath);
192}
193
194/// Called for each document of a scan, in `__name__` order; return `Break` to stop early.
195pub type Visit<'a> = dyn FnMut(&Arc<StoredDocument>) -> ControlFlow<()> + 'a;
196
197/// An entry of a collection listing.
198#[derive(Debug)]
199pub enum ListItem<'a> {
200    Document(&'a Arc<StoredDocument>),
201    /// No document exists at this path, but documents exist below it (in subcollections).
202    Missing(&'a ResourcePath),
203}
204
205/// Options of [`Store::list_collection`].
206#[derive(Debug, Default, Clone, Copy)]
207pub struct ListOptions<'a> {
208    /// Resume after this document path (exclusive), for paging.
209    pub after: Option<&'a ResourcePath>,
210    /// Also report missing documents that have descendants (`ListDocuments.show_missing`).
211    pub include_missing: bool,
212}
213
214/// The storage engine. `database` is the full database name,
215/// `projects/{project}/databases/{database}`; databases are created on first write.
216pub trait Store: Send + Sync {
217    /// A read time that sees every commit made so far.
218    fn latest_read_time(&self, database: &str) -> ReadTime;
219
220    /// The oldest read time still served; older versions may have been dropped.
221    fn earliest_read_time(&self, database: &str) -> ReadTime;
222
223    fn get(&self, database: &str, path: &ResourcePath, at: ReadTime)
224    -> Option<Arc<StoredDocument>>;
225
226    /// The documents directly in `collection` (not in its subcollections).
227    fn scan_collection(
228        &self,
229        database: &str,
230        collection: &ResourcePath,
231        at: ReadTime,
232        visit: &mut Visit<'_>,
233    ) {
234        self.list_collection(
235            database,
236            collection,
237            at,
238            ListOptions::default(),
239            &mut |item| match item {
240                ListItem::Document(doc) => visit(doc),
241                ListItem::Missing(_) => ControlFlow::Continue(()),
242            },
243        );
244    }
245
246    /// Like [`Store::scan_collection`], with paging and missing documents.
247    fn list_collection(
248        &self,
249        database: &str,
250        collection: &ResourcePath,
251        at: ReadTime,
252        options: ListOptions<'_>,
253        visit: &mut dyn FnMut(ListItem<'_>) -> ControlFlow<()>,
254    );
255
256    /// Every document under `parent` (a document path, or the root) whose collection ID is
257    /// `collection_id`, at any depth, ordered by full path.
258    fn scan_collection_group(
259        &self,
260        database: &str,
261        parent: &ResourcePath,
262        collection_id: &str,
263        at: ReadTime,
264        visit: &mut Visit<'_>,
265    );
266
267    /// Every document below `parent` (a document path, or the root), at any depth, ordered by
268    /// path. `parent` itself is not included.
269    fn scan_descendants(
270        &self,
271        database: &str,
272        parent: &ResourcePath,
273        at: ReadTime,
274        visit: &mut Visit<'_>,
275    );
276
277    /// IDs of the collections directly under `parent` that contain at least one document at
278    /// any depth, in UTF-8 order.
279    fn list_collection_ids(
280        &self,
281        database: &str,
282        parent: &ResourcePath,
283        at: ReadTime,
284    ) -> Vec<String>;
285
286    /// Runs `write` as one atomic commit.
287    fn commit(
288        &self,
289        database: &str,
290        write: &mut dyn FnMut(&mut dyn WriteBatch) -> Result<(), StoreError>,
291    ) -> Result<Commit, StoreError>;
292
293    /// Drops every document of `database` and every older version, at a time after its last
294    /// commit, which it returns: later commits are later still, so listeners can order the
295    /// clear among them.
296    fn clear_database(&self, database: &str) -> ReadTime;
297
298    /// The databases that hold or held documents.
299    fn database_names(&self) -> Vec<String>;
300
301    /// Drops every database.
302    fn clear(&self);
303}
hidane_core/store/mod.rs

hidane_core/store/
mod.rs
