haematite_cli/serve/storage/
admission.rs1use 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#[derive(Debug)]
15pub enum Admission {
16 Replayed(Receipt),
17 Admitted(Captured),
18}
19
20impl Storage {
21 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;