haematite_cli/serve/storage/
mutations.rs

1use haematite::branch::history::Key as HistoryKey;
2use haematite::branch::{
3    CommitDurability, CommitRequest, ForkWithChanges, ForkWrite, RefRecord, current_timestamp,
4    open as open_branch,
5};
6use haematite::db::branches::ForkKeptError;
7use haematite::tree::Hash;
8
9use super::{
10    Captured, Collection, DocumentRevision, Failure, Receipt, Storage, branch_name, collection_key,
11    counts, document_key, identifier, reads, receipt_key, revision_key, version,
12};
13use crate::serve::protocol::{Change, DocumentImport, Mutation, ReceiptQuery, Selection, Version};
14use crate::serve::schema::Collection as CollectionSchema;
15
16pub(super) type PreparedWrite = (Vec<u8>, Option<Vec<u8>>);
17
18impl Storage {
19    /// Reconcile a receipt and retry only its reference fence, not the write.
20    ///
21    /// # Errors
22    ///
23    /// Propagates any error from `identifier`, `get_bytes`, `from_slice` or `publish_record`.
24    pub fn receipt(&mut self, query: &ReceiptQuery) -> Result<Option<Receipt>, Failure> {
25        identifier(&query.request_id)?;
26        identifier(&query.branch)?;
27        if let Some(receipt) = self.discard_receipt(query)? {
28            return Ok(Some(receipt));
29        }
30        if self.pending_discards.contains(&branch_name(&query.branch)) {
31            return Err(Failure::new(
32                "discard_pending",
33                "branch removal requires receipt reconciliation",
34            ));
35        }
36        // Reconciliation deliberately sees an installed but not yet published
37        // head. Ordinary readers use only the separately confirmed catalogue.
38        let record = self
39            .refs
40            .get(&branch_name(&query.branch))
41            .ok_or_else(|| Failure::new("branch_missing", &query.branch))?;
42        let at = Captured {
43            branch: query.branch.clone(),
44            version: version(record),
45            shards: record.shards.clone(),
46        };
47        let Some(bytes) = reads::get_bytes(
48            &self.reader,
49            &at,
50            &receipt_key(&query.branch, at.version.generation, &query.request_id),
51        )?
52        else {
53            return Ok(None);
54        };
55        let receipt: Receipt = serde_json::from_slice(&bytes).map_err(Failure::storage)?;
56        if receipt.request_id != query.request_id
57            || receipt.branch != query.branch
58            || receipt.version.generation != at.version.generation
59        {
60            return Err(Failure::new(
61                "receipt_corrupt",
62                "receipt identity does not match its stored key",
63            ));
64        }
65        self.refs
66            .confirm_version(
67                &branch_name(&query.branch),
68                at.version.generation,
69                at.version.sequence,
70            )
71            .map_err(|error| Failure::new("durability_unknown", error.to_string()))?;
72        self.publish_record(&branch_name(&query.branch))?;
73        Ok(Some(receipt))
74    }
75
76    /// One durable request identity, validation and effect within admission.
77    ///
78    /// # Errors
79    ///
80    /// Propagates any error from `identifier`, `to_vec`, `replay` or `prepare_change`.
81    pub fn mutate(&mut self, request: &Mutation) -> Result<Receipt, Failure> {
82        self.mutate_with_import_bound(request, None, super::AppendPruneLimits::default())
83    }
84
85    /// Admit imports up to sixty-four pages, preserving receipt-first replay.
86    ///
87    /// # Errors
88    /// Propagates mutation preparation, admission, replay and durable commit refusals without changing their outcome classification.
89    pub fn mutate_bounded(
90        &mut self,
91        request: &Mutation,
92        max_page: std::num::NonZeroU32,
93    ) -> Result<Receipt, Failure> {
94        self.mutate_with_import_bound(request, Some(max_page), super::AppendPruneLimits::default())
95    }
96
97    /// Admit a mutation with the configured retention and import bounds.
98    ///
99    /// # Errors
100    /// Preserves receipt-first replay and propagates preparation or commit refusals.
101    pub fn mutate_with_append_prune_limits(
102        &mut self,
103        request: &Mutation,
104        max_page: std::num::NonZeroU32,
105        limits: super::AppendPruneLimits,
106    ) -> Result<Receipt, Failure> {
107        self.mutate_with_import_bound(request, Some(max_page), limits)
108    }
109
110    /// The live record a mutation names, at exactly the version it expects.
111    fn expected_record(&self, request: &Mutation) -> Result<RefRecord, Failure> {
112        let record = self
113            .refs
114            .get(&branch_name(&request.branch))
115            .ok_or_else(|| Failure::new("branch_missing", &request.branch))?;
116        if version(record) != request.expected {
117            return Err(Failure::new(
118                "version_conflict",
119                "expected branch generation or sequence differs",
120            ));
121        }
122        if self.pending_discards.contains(&record.name) {
123            return Err(Failure::new(
124                "discard_pending",
125                "branch removal requires receipt reconciliation",
126            ));
127        }
128        Ok(record.clone())
129    }
130
131    fn mutate_with_import_bound(
132        &mut self,
133        request: &Mutation,
134        import_max_page: Option<std::num::NonZeroU32>,
135        retention_limits: super::AppendPruneLimits,
136    ) -> Result<Receipt, Failure> {
137        identifier(&request.request_id)?;
138        identifier(&request.branch)?;
139        let destination = match &request.change {
140            Change::Fork(fork) => &fork.name,
141            _ => &request.branch,
142        };
143        identifier(destination)?;
144        let fingerprint =
145            Hash::of(&serde_json::to_vec(request).map_err(Failure::storage)?).to_string();
146        if let Some(receipt) = self.replay(request, destination, &fingerprint)? {
147            return Ok(receipt);
148        }
149        validate_import_bound(&request.change, import_max_page)?;
150        let record = self.expected_record(request)?;
151        let at = Captured {
152            branch: request.branch.clone(),
153            version: request.expected,
154            shards: record.shards.clone(),
155        };
156        if let Change::Fork(fork) = &request.change {
157            return self.fork_request(request, &fingerprint, &fork.name, fork.from);
158        }
159        if let Change::Discard(discard) = &request.change {
160            return self.discard_request(request, &fingerprint, &discard.name, &record);
161        }
162        let next = Version {
163            generation: record.created,
164            sequence: record.seq.checked_add(1).ok_or_else(|| {
165                Failure::new("sequence_exhausted", "branch sequence cannot advance")
166            })?,
167        };
168        let (mut writes, retention) = match &request.change {
169            Change::AppendPrune(write) => {
170                let (writes, result) =
171                    self.prepare_append_prune(&at, write, next, retention_limits)?;
172                (writes, Some(result))
173            }
174            change => (self.prepare_change(&at, change, next)?, None),
175        };
176        counts::append(&self.reader, &at, &request.change, &mut writes)?;
177        // A version the advance drops from the kept window has its row first
178        // (STOR-003 MD5a).
179        self.rows_before_advance(&request.branch, &record)?;
180        #[cfg(test)]
181        super::history::after_rows::reached()?;
182        let branch = open_branch(&record.name, &self.refs, &self.registry)
183            .map_err(Failure::branch_commit)?;
184        for (key, value) in writes {
185            let shard_id = self.writer.shard_for(&key);
186            match value {
187                Some(value) => branch.put(shard_id, key, value),
188                None => branch.delete(shard_id, key),
189            }
190            .map_err(Failure::branch)?;
191        }
192        let receipt = Receipt {
193            request_id: request.request_id.clone(),
194            fingerprint,
195            branch: request.branch.clone(),
196            version: next,
197            retention,
198        };
199        let key = receipt_key(&request.branch, next.generation, &request.request_id);
200        branch
201            .put(
202                self.writer.shard_for(&key),
203                key,
204                serde_json::to_vec(&receipt).map_err(Failure::storage)?,
205            )
206            .map_err(Failure::branch)?;
207        let result = self.writer.commit(
208            &branch,
209            &self.registry,
210            CommitRequest {
211                durability: CommitDurability::Durable {
212                    refs: &mut self.refs,
213                },
214                extra_parents: &[],
215                timestamp: current_timestamp(),
216            },
217        );
218        if let Err(error) = result {
219            let installed = self
220                .refs
221                .get(&record.name)
222                .is_some_and(|now| version(now) != request.expected);
223            let mut failure = Failure::branch_commit(error);
224            let code = if installed {
225                "durability_unknown"
226            } else {
227                "write_refused"
228            };
229            code.clone_into(&mut failure.code);
230            return Err(failure);
231        }
232        #[cfg(test)]
233        lose_mutation_answer()?;
234        #[cfg(test)]
235        publish_pause_point();
236        self.publish_record(&record.name)?;
237        Ok(receipt)
238    }
239
240    fn replay(
241        &mut self,
242        request: &Mutation,
243        destination: &str,
244        fingerprint: &str,
245    ) -> Result<Option<Receipt>, Failure> {
246        // A fork receipt lives in its destination and is absent before creation.
247        if matches!(&request.change, Change::Discard(_))
248            || self.refs.get(&branch_name(destination)).is_some()
249        {
250            if let Some(receipt) = self.receipt(&ReceiptQuery {
251                branch: destination.to_owned(),
252                request_id: request.request_id.clone(),
253            })? {
254                return if receipt.fingerprint == fingerprint {
255                    Ok(Some(receipt))
256                } else {
257                    Err(Failure::new(
258                        "request_conflict",
259                        "request identity was used with different content",
260                    ))
261                };
262            }
263            if matches!(&request.change, Change::Fork(_)) {
264                return Err(Failure::new(
265                    "branch_exists",
266                    "destination exists without this fork receipt",
267                ));
268            }
269        }
270        Ok(None)
271    }
272
273    pub(super) fn prepare_change(
274        &self,
275        at: &Captured,
276        change: &Change,
277        next: Version,
278    ) -> Result<Vec<PreparedWrite>, Failure> {
279        match change {
280            Change::Collection(write) => {
281                identifier(&write.collection)?;
282                if let Some(schema) = &write.schema {
283                    CollectionSchema::compile(schema).map_err(schema_failure)?;
284                }
285                let collection = Collection {
286                    name: write.collection.clone(),
287                    schema: write.schema.clone(),
288                    changed_in: at.branch.clone(),
289                    changed_at: next,
290                };
291                Ok(vec![(
292                    collection_key(&write.collection),
293                    Some(serde_json::to_vec(&collection).map_err(Failure::storage)?),
294                )])
295            }
296            Change::Put(write) => {
297                identifier(&write.document_id)?;
298                let collection = reads::collection(&self.reader, at, &write.collection)?
299                    .ok_or_else(|| Failure::new("collection_missing", &write.collection))?;
300                if let Some(schema) = &collection.schema {
301                    CollectionSchema::compile(schema)
302                        .map_err(schema_failure)?
303                        .validate(
304                            &serde_json::from_str(write.document.get())
305                                .map_err(Failure::storage)?,
306                        )
307                        .map_err(schema_failure)?;
308                }
309                let revision = DocumentRevision {
310                    collection_branch: collection.changed_in,
311                    collection_version: collection.changed_at,
312                    written_in: at.branch.clone(),
313                    written_at: next,
314                };
315                Ok(vec![
316                    (
317                        document_key(&write.collection, &write.document_id),
318                        Some(write.document.get().as_bytes().to_vec()),
319                    ),
320                    (
321                        revision_key(&write.collection, &write.document_id),
322                        Some(serde_json::to_vec(&revision).map_err(Failure::storage)?),
323                    ),
324                ])
325            }
326            Change::AppendPrune(_) => Err(Failure::new(
327                "append_prune_preparation_required",
328                "retention uses bounded mutation preparation",
329            )),
330            Change::Delete(write) => {
331                identifier(&write.collection)?;
332                identifier(&write.document_id)?;
333                // A deleted document takes its edges, in both directions, and its
334                // vectors with it in the same version (D7).
335                let mut writes = vec![
336                    (document_key(&write.collection, &write.document_id), None),
337                    (revision_key(&write.collection, &write.document_id), None),
338                ];
339                writes.extend(super::edges::removals_for_document(
340                    &self.reader,
341                    at,
342                    &write.collection,
343                    &write.document_id,
344                )?);
345                writes.extend(super::vectors::removals_for_document(
346                    &self.reader,
347                    at,
348                    &write.collection,
349                    &write.document_id,
350                )?);
351                Ok(writes)
352            }
353            Change::DeleteMany(delete) => self.prepare_delete_many(at, delete, next),
354            Change::Merge(merge) => self.merge_writes(at, merge),
355            Change::Fork(_) | Change::Discard(_) => Err(Failure::new(
356                "invalid_operation",
357                "branch control uses the publication path",
358            )),
359            Change::EdgePut(edge) => super::edges::put(&self.reader, at, edge),
360            Change::EdgeDelete(edge) => super::edges::delete(edge),
361            Change::EdgePutMany(import) => super::edges::put_many(&self.reader, at, import),
362            Change::IndexDeclare(field) => {
363                super::field_index::declare(&self.reader, at, field, next)
364            }
365            Change::VectorDeclare(field) => super::vectors::declare(&self.reader, at, field),
366            Change::VectorPut(vector) => super::vectors::put(&self.reader, at, vector),
367            Change::VectorPutMany(import) => {
368                super::vector_import::prepare(&self.reader, at, import)
369            }
370            Change::VectorDelete(vector) => super::vectors::delete(vector),
371            Change::PutMany(import) => self.prepare_import(at, import, next),
372            Change::RelationPut(relation) => super::relations::put(&self.reader, at, relation),
373            Change::RelationDelete(relation) => super::relations::delete(relation),
374        }
375    }
376
377    fn prepare_import(
378        &self,
379        at: &Captured,
380        import: &DocumentImport,
381        next: Version,
382    ) -> Result<Vec<PreparedWrite>, Failure> {
383        if import.documents.is_empty() {
384            return Err(Failure {
385                code: "import_empty".to_owned(),
386                message: "an import must contain at least one document".to_owned(),
387                path: Some("/documents".to_owned()),
388                cause: None,
389            });
390        }
391        let collection = reads::collection(&self.reader, at, &import.collection)
392            .map_err(|mut failure| {
393                failure.path = Some("/collection".to_owned());
394                failure
395            })?
396            .ok_or_else(|| Failure {
397                code: "collection_missing".to_owned(),
398                message: import.collection.clone(),
399                path: Some("/collection".to_owned()),
400                cause: None,
401            })?;
402        let schema = collection
403            .schema
404            .as_ref()
405            .map(CollectionSchema::compile)
406            .transpose()
407            .map_err(schema_failure)?;
408        validate_import_rows(import, schema.as_ref())?;
409        let revision = serde_json::to_vec(&DocumentRevision {
410            collection_branch: collection.changed_in,
411            collection_version: collection.changed_at,
412            written_in: at.branch.clone(),
413            written_at: next,
414        })
415        .map_err(Failure::storage)?;
416        let mut writes = Vec::new();
417        for document in &import.documents {
418            writes.push((
419                document_key(&import.collection, &document.document_id),
420                Some(document.document.get().as_bytes().to_vec()),
421            ));
422            writes.push((
423                revision_key(&import.collection, &document.document_id),
424                Some(revision.clone()),
425            ));
426        }
427        Ok(writes)
428    }
429
430    fn fork_request(
431        &mut self,
432        request: &Mutation,
433        fingerprint: &str,
434        destination: &str,
435        from: Version,
436    ) -> Result<Receipt, Failure> {
437        if self.pending_discards.contains(&branch_name(destination)) {
438            return Err(Failure::new(
439                "discard_pending",
440                "destination removal requires receipt reconciliation",
441            ));
442        }
443        self.capture(&Selection {
444            branch: request.branch.clone(),
445            version: Some(from),
446        })?;
447        let current = self
448            .refs
449            .get(&branch_name(&request.branch))
450            .ok_or_else(|| Failure::new("branch_missing", &request.branch))?
451            .clone();
452        // `fork_kept` reads the row of the version forked from: the head's is
453        // written only when the head is that version (STOR-003 MD5b).
454        if from == version(&current) {
455            self.retain_current(&request.branch, &current)?;
456        }
457        // A kept version forked from has its row before `fork_kept` reads it
458        // (STOR-003 MD5a).
459        self.row_for_kept(&request.branch, &current, from)?;
460        #[cfg(test)]
461        super::history::after_rows::reached()?;
462        let service =
463            open_branch(super::SERVICE, &self.refs, &self.registry).map_err(|source| {
464                fork_kept_failure(&ForkKeptError::ServiceBindingStale { source }, false)
465            })?;
466        let history_key = HistoryKey {
467            branch: Hash::of(request.branch.as_bytes()),
468            generation: from.generation,
469            sequence: from.sequence,
470        };
471        let created = current_timestamp();
472        let receipt = Receipt {
473            request_id: request.request_id.clone(),
474            fingerprint: fingerprint.to_owned(),
475            branch: destination.to_owned(),
476            version: Version {
477                generation: created,
478                sequence: 0,
479            },
480            retention: None,
481        };
482        let key = receipt_key(destination, created, &request.request_id);
483        let writes = [
484            ForkWrite {
485                shard_id: self.writer.shard_for(&key),
486                key,
487                value: Some(serde_json::to_vec(&receipt).map_err(Failure::storage)?),
488            },
489            super::branches::provenance_write(&self.writer, &request.branch, from)?,
490        ];
491        let name = branch_name(destination);
492        self.writer
493            .fork_kept(
494                &service,
495                history_key,
496                &mut self.refs,
497                &self.registry,
498                ForkWithChanges {
499                    name: &name,
500                    timestamp: created,
501                    writes: &writes,
502                },
503            )
504            .map_err(|error| fork_kept_failure(&error, self.refs.get(&name).is_some()))?;
505        #[cfg(test)]
506        lose_mutation_answer()?;
507        self.publish_record(&name)?;
508        Ok(receipt)
509    }
510
511    /// Stop this thread's next committed mutation before it publishes.
512    ///
513    /// The pause point signals on `paused` once the store commit has returned,
514    /// then waits on `release` while still holding the mutation admission.
515    #[cfg(test)]
516    pub fn arm_publish_pause(
517        paused: std::sync::mpsc::Sender<()>,
518        release: std::sync::mpsc::Receiver<()>,
519    ) {
520        PUBLISH_PAUSE.with(|slot| {
521            *slot.borrow_mut() = Some(PublishPause { paused, release });
522        });
523    }
524
525    #[cfg(test)]
526    pub(crate) fn lose_next_mutation_answer() {
527        Self::lose_next_mutation_answer_with_message(
528            "injected lost answer after reference commit".to_owned(),
529        );
530    }
531
532    #[cfg(test)]
533    pub(crate) fn lose_next_mutation_answer_with_message(message: String) {
534        LOST_MUTATION_ANSWER.with(|slot| {
535            *slot.borrow_mut() = Some(message);
536        });
537    }
538}
539
540#[cfg(test)]
541thread_local! {
542    static LOST_MUTATION_ANSWER: std::cell::RefCell<Option<String>> = const { std::cell::RefCell::new(None) };
543}
544
545#[cfg(test)]
546fn lose_mutation_answer() -> Result<(), Failure> {
547    if let Some(message) = LOST_MUTATION_ANSWER.with(|slot| slot.borrow_mut().take()) {
548        return Err(Failure::new("durability_unknown", message));
549    }
550    Ok(())
551}
552
553fn fork_kept_failure(error: &ForkKeptError, installed: bool) -> Failure {
554    let code = match error {
555        ForkKeptError::ServiceBindingStale { .. } => "kept_service_binding_stale",
556        ForkKeptError::ServiceInvalid { .. } => "kept_service_invalid",
557        ForkKeptError::ShardSetIncomplete { .. } => "kept_shard_set_incomplete",
558        ForkKeptError::RowMissing { .. } => "kept_row_missing",
559        ForkKeptError::RowInvalid { .. } => "kept_row_invalid",
560        ForkKeptError::RowRead { .. } => "kept_row_read",
561        ForkKeptError::RootUnavailable { .. } => "kept_root_unavailable",
562        ForkKeptError::Fork(_) if installed => "durability_unknown",
563        ForkKeptError::Fork(_) => "write_refused",
564    };
565    Failure::new(code, error.to_string())
566}
567
568#[cfg(test)]
569#[path = "mutations_kept_tests.rs"]
570mod kept_tests;
571
572fn validate_import_rows(
573    import: &DocumentImport,
574    schema: Option<&CollectionSchema>,
575) -> Result<(), Failure> {
576    let mut ids = std::collections::BTreeSet::new();
577    for (index, document) in import.documents.iter().enumerate() {
578        identifier(&document.document_id).map_err(|mut failure| {
579            failure.path = Some(format!("/documents/{index}/document_id"));
580            failure
581        })?;
582        if !ids.insert(document.document_id.as_str()) {
583            return Err(Failure {
584                code: "import_duplicate".to_owned(),
585                message: format!("repeated document_id {}", document.document_id),
586                path: Some(format!("/documents/{index}/document_id")),
587                cause: None,
588            });
589        }
590        if let Some(schema) = schema {
591            let value = serde_json::from_str(document.document.get()).map_err(|error| Failure {
592                code: "invalid_document".to_owned(),
593                message: error.to_string(),
594                path: Some(format!("/documents/{index}/document")),
595                cause: None,
596            })?;
597            schema.validate(&value).map_err(|error| Failure {
598                code: error.code().to_owned(),
599                message: error.reason,
600                path: Some(format!("/documents/{index}/document{}", error.path)),
601                cause: None,
602            })?;
603        }
604    }
605    Ok(())
606}
607
608fn validate_import_bound(
609    change: &Change,
610    max_page: Option<std::num::NonZeroU32>,
611) -> Result<(), Failure> {
612    let (rows, kind) = match change {
613        Change::PutMany(import) => (import.documents.len(), "documents"),
614        Change::DeleteMany(delete) => (delete.document_ids.len(), "document_ids"),
615        Change::EdgePutMany(import) => (import.edges.len(), "edges"),
616        Change::VectorPutMany(import) => (import.vectors.len(), "vectors"),
617        _ => return Ok(()),
618    };
619    let max_page = max_page.ok_or_else(|| {
620        Failure::new(
621            "import_limit_required",
622            "an import requires the service page bound",
623        )
624    })?;
625    let bound = u64::from(max_page.get())
626        .checked_mul(64)
627        .ok_or_else(|| Failure::new("import_limit", "import row bound exceeds u64"))?;
628    if u64::try_from(rows).map_err(Failure::storage)? > bound {
629        return Err(Failure {
630            code: "import_limit".to_owned(),
631            message: format!("maximum import size is {bound} {kind}"),
632            path: Some(format!("/{kind}/{bound}")),
633            cause: None,
634        });
635    }
636    Ok(())
637}
638
639fn schema_failure(error: crate::serve::schema::Violation) -> Failure {
640    Failure {
641        code: error.code().to_owned(),
642        message: error.reason,
643        path: Some(error.path),
644        cause: None,
645    }
646}
647
648#[cfg(test)]
649struct PublishPause {
650    paused: std::sync::mpsc::Sender<()>,
651    release: std::sync::mpsc::Receiver<()>,
652}
653
654#[cfg(test)]
655thread_local! {
656    static PUBLISH_PAUSE: std::cell::RefCell<Option<PublishPause>> =
657        const { std::cell::RefCell::new(None) };
658}
659
660/// Between the durable store commit and the catalogue publication. The ten
661/// second bound turns a lost signal into a named failure, not a hung test.
662#[cfg(test)]
663fn publish_pause_point() {
664    let Some(pause) = PUBLISH_PAUSE.with(|slot| slot.borrow_mut().take()) else {
665        return;
666    };
667    assert!(
668        pause.paused.send(()).is_ok(),
669        "paused signal receiver was dropped"
670    );
671    let wait = std::time::Duration::from_secs(10);
672    let released = pause.release.recv_timeout(wait);
673    assert!(released.is_ok(), "release signal not received within 10 s");
674}
haematite_cli/serve/storage/mutations.rs

haematite_cli/serve/storage/
mutations.rs
