haematite_cli/serve/storage/
admission.rs

1//! Dormant contract admission, retaining the caller's admission ownership.
2
3use super::{
4    Captured, Failure, SERVICE, Storage, branch_name, discards, reads, receipt_key, version,
5};
6use crate::serve::contracts::{Contract, Receipt, ResolvedMutation};
7use crate::serve::protocol::{Change, Selection};
8
9#[path = "admission_reads.rs"]
10mod dependencies;
11
12/// Contract admission is valid only while the caller retains admission.
13/// Preparing and publishing the logical change remain separate steps.
14#[derive(Debug)]
15pub enum Admission {
16    Replayed(Receipt),
17    Admitted(Captured),
18}
19
20impl Storage {
21    /// Decide a resolved request without publishing or retaining another version.
22    /// Receipt reconciliation precedes every contract and origin check.
23    ///
24    /// # Errors
25    /// Refuses reused identities, changed dependencies, unconfirmed heads and
26    /// mutation kinds whose dependencies cannot yet be proved. Propagates reads.
27    pub fn decide_resolved(&mut self, request: &ResolvedMutation) -> Result<Admission, Failure> {
28        if let Some(receipt) = self.resolved_replay(request)? {
29            return Ok(Admission::Replayed(receipt));
30        }
31        let name = branch_name(request.branch());
32        let record = self
33            .refs
34            .get(&name)
35            .ok_or_else(|| Failure::new("branch_missing", request.branch()))?;
36        let installed = version(record);
37        if let Contract::ExactVersion { expected } = request.contract()
38            && *expected != installed
39        {
40            return Err(Failure::new(
41                "version_conflict",
42                "expected branch generation or sequence differs",
43            ));
44        }
45        if self.pending_discards.contains(&name) {
46            return Err(Failure::new(
47                "discard_pending",
48                "branch removal requires receipt reconciliation",
49            ));
50        }
51        let head = self.capture(&Selection {
52            branch: request.branch().to_owned(),
53            version: None,
54        })?;
55        if head.version != installed {
56            return Err(Failure::new(
57                "durability_unknown",
58                "installed head requires receipt reconciliation before admission",
59            ));
60        }
61        if head.shards.len() != self.reader.shard_count()
62            || head
63                .shards
64                .iter()
65                .enumerate()
66                .any(|(id, shard)| id != shard.shard_id)
67        {
68            return Err(Failure::new(
69                "incomplete_roots",
70                "head does not name every physical shard in order",
71            ));
72        }
73        if let Contract::Dependencies { origin, reads } = request.contract() {
74            dependencies::require_supported(request.change())?;
75            if origin.generation != head.version.generation {
76                return Err(Failure::new(
77                    "generation_mismatch",
78                    "dependency origin names another branch generation",
79                ));
80            }
81            if origin.sequence > head.version.sequence {
82                return Err(Failure::new(
83                    "version_missing",
84                    "dependency origin is ahead of the branch",
85                ));
86            }
87            let at = self.capture(&Selection {
88                branch: request.branch().to_owned(),
89                version: Some(*origin),
90            })?;
91            dependencies::validate(&self.reader, &at, &head, reads, request.change())?;
92        }
93        Ok(Admission::Admitted(head))
94    }
95
96    fn resolved_replay(&mut self, request: &ResolvedMutation) -> Result<Option<Receipt>, Failure> {
97        let service = self
98            .refs
99            .get(SERVICE)
100            .ok_or_else(|| Failure::new("reference_missing", "missing service namespace"))?;
101        let at = Captured {
102            branch: SERVICE.to_owned(),
103            version: version(service),
104            shards: service.shards.clone(),
105        };
106        if let Some(bytes) = reads::get_bytes(
107            &self.reader,
108            &at,
109            &discards::key(request.destination(), request.request_id()),
110        )? {
111            let receipt = Receipt::decode(&bytes)?;
112            receipt.replay(request)?;
113            self.confirm_replay(SERVICE, at.version)?;
114            return Ok(Some(receipt));
115        }
116        let name = branch_name(request.destination());
117        if self.pending_discards.contains(&name) {
118            return Err(Failure::new(
119                "discard_pending",
120                "branch removal requires receipt reconciliation",
121            ));
122        }
123        let Some(record) = self.refs.get(&name) else {
124            return Ok(None);
125        };
126        let at = Captured {
127            branch: request.destination().to_owned(),
128            version: version(record),
129            shards: record.shards.clone(),
130        };
131        let key = receipt_key(
132            request.destination(),
133            at.version.generation,
134            request.request_id(),
135        );
136        let Some(bytes) = reads::get_bytes(&self.reader, &at, &key)? else {
137            if matches!(request.change(), Change::Fork(_)) {
138                return Err(Failure::new(
139                    "branch_exists",
140                    "destination exists without this fork receipt",
141                ));
142            }
143            return Ok(None);
144        };
145        let receipt = Receipt::decode(&bytes)?;
146        receipt.replay(request)?;
147        if receipt.version().generation != at.version.generation
148            || receipt.version().sequence > at.version.sequence
149        {
150            return Err(Failure::new(
151                "receipt_corrupt",
152                "receipt publication version differs from its stored branch",
153            ));
154        }
155        self.confirm_replay(&name, at.version)?;
156        Ok(Some(receipt))
157    }
158
159    fn confirm_replay(
160        &mut self,
161        name: &str,
162        at: crate::serve::protocol::Version,
163    ) -> Result<(), Failure> {
164        self.refs
165            .confirm_version(name, at.generation, at.sequence)
166            .map_err(|error| Failure::new("durability_unknown", error.to_string()))?;
167        self.publish_record(name)
168    }
169}
170
171#[cfg(test)]
172#[path = "admission_tests.rs"]
173mod tests;
haematite_cli/serve/storage/admission.rs

haematite_cli/serve/storage/
admission.rs
