1use std::{
2 collections::HashMap,
3 fs::{File, OpenOptions},
4 io::{Read, Seek, SeekFrom, Write},
5 path::Path,
6 sync::{Arc, Mutex},
7};
8
9pub use kcode_k1_launch_node_codec::{
10 Authority, GroupId, NodeId, TargetId, TargetName, TxId, UserId,
11};
12use kcode_k1_launch_node_codec::{
13 PROJECTION_HEADER, ProjectionDecode, ProjectionRecord, SetAction, decode_projection_record,
14};
15use kcode_k1_peering::K1Peering;
16use kcode_k1_txn_ordering::{K1TxnOrdering, Subsystem as OrderingSubsystem, SubsystemId};
17
18struct Pending {
19 payload: Vec<u8>,
20 evidence: Option<TxId>,
21}
22
23struct Inner {
24 file: File,
25 bindings: HashMap<TargetId, NodeId>,
26 checkpoint: Option<TxId>,
27 pending: HashMap<[u8; 16], Pending>,
28 failed: bool,
29}
30
31struct Shared {
32 inner: Mutex<Inner>,
33}
34
35struct Driver {
36 shared: Arc<Shared>,
37}
38
39pub struct LaunchNodes {
40 shared: Arc<Shared>,
41 peering: Arc<K1Peering>,
42}
43
44impl LaunchNodes {
45 pub fn open(
46 path: &Path,
47 ordering: Arc<K1TxnOrdering>,
48 peering: Arc<K1Peering>,
49 ) -> Result<Self, String> {
50 let mut projection = open_projection(path)?;
51 if let Some(checkpoint) = projection.checkpoint.to_owned() {
52 match ordering.get_txn(checkpoint) {
53 Ok(Some(_)) => {}
54 Ok(None) => {
55 reset_projection(&mut projection.file, path)?;
56 projection.bindings.clear();
57 projection.checkpoint = None;
58 }
59 Err(error) => return Err(format!("validate launch nodes checkpoint: {error}")),
60 }
61 }
62 let after = projection.checkpoint.to_owned();
63 let shared = Arc::new(Shared {
64 inner: Mutex::new(Inner {
65 file: projection.file,
66 bindings: projection.bindings,
67 checkpoint: projection.checkpoint,
68 pending: HashMap::new(),
69 failed: false,
70 }),
71 });
72 ordering.register_subsystem(
73 subsystem()?,
74 after,
75 Arc::new(Driver {
76 shared: shared.clone(),
77 }),
78 )?;
79 Ok(Self { shared, peering })
80 }
81
82 pub fn get(&self, target: &TargetId) -> Result<Option<NodeId>, String> {
83 let inner = self
84 .shared
85 .inner
86 .lock()
87 .map_err(|_| "k1-launch-nodes lock poisoned".to_owned())?;
88 if inner.failed {
89 return Err("k1-launch-nodes facade is unavailable".into());
90 }
91 Ok(inner.bindings.get(target).cloned())
92 }
93
94 pub fn set(&self, target: TargetId, node: NodeId) -> Result<(), String> {
95 let (operation_id, payload) = loop {
96 let mut operation_id = [0_u8; 16];
97 getrandom::fill(&mut operation_id)
98 .map_err(|error| format!("generate launch nodes operation ID: {error}"))?;
99 let payload = SetAction::new(operation_id, target.clone(), node).encode();
100 let mut inner = self
101 .shared
102 .inner
103 .lock()
104 .map_err(|_| "k1-launch-nodes lock poisoned".to_owned())?;
105 if inner.failed {
106 return Err("k1-launch-nodes facade is unavailable".into());
107 }
108 if inner.pending.contains_key(&operation_id) {
109 continue;
110 }
111 inner.pending.insert(
112 operation_id,
113 Pending {
114 payload: payload.clone(),
115 evidence: None,
116 },
117 );
118 break (operation_id, payload);
119 };
120 let submission = self.peering.submit_txn(subsystem()?, &payload);
121 let mut inner = self
122 .shared
123 .inner
124 .lock()
125 .map_err(|_| "k1-launch-nodes lock poisoned".to_owned())?;
126 if inner.failed {
127 return Err("k1-launch-nodes facade is unavailable".into());
128 }
129 let Some(pending) = inner.pending.remove(&operation_id) else {
130 return Err(fault(
131 &mut inner,
132 "k1-launch-nodes callback evidence is missing",
133 ));
134 };
135 match (submission, pending.evidence) {
136 (Ok(submitted), Some(callback)) if submitted == callback => Ok(()),
137 (Err(_), Some(_)) => Ok(()),
138 (Err(error), None) => Err(format!("submit k1-launch-nodes transaction: {error}")),
139 (Ok(_), None) => Err(fault(
140 &mut inner,
141 "k1-launch-nodes callback evidence is missing",
142 )),
143 (Ok(_), Some(_)) => Err(fault(
144 &mut inner,
145 "k1-launch-nodes callback transaction does not match submission",
146 )),
147 }
148 }
149}
150
151impl OrderingSubsystem for Driver {
152 fn submit_txn(&self, id: TxId, payload: &[u8]) -> Result<(), String> {
153 let action = match SetAction::decode(payload) {
154 Ok(action) => action,
155 Err(error) => {
156 return self
157 .shared
158 .fail(format!("decode k1-launch-nodes Set action: {error}"));
159 }
160 };
161 let operation_id = action.operation_id().to_owned();
162 let canonical_payload = action.encode();
163 let record =
164 ProjectionRecord::new(id, action.target().to_owned(), action.node().to_owned())
165 .encode();
166 let mut inner = self
167 .shared
168 .inner
169 .lock()
170 .map_err(|_| "k1-launch-nodes lock poisoned".to_owned())?;
171 if inner.failed {
172 return Err("k1-launch-nodes facade is unavailable".into());
173 }
174 if inner.checkpoint == Some(id) {
175 return Err(fault(
176 &mut inner,
177 "k1-launch-nodes callback transaction is duplicated",
178 ));
179 }
180 if let Some(pending) = inner.pending.get(&operation_id) {
181 if pending.payload != canonical_payload {
182 return Err(fault(
183 &mut inner,
184 "k1-launch-nodes callback action does not match reservation",
185 ));
186 }
187 if pending.evidence.is_some() {
188 return Err(fault(
189 &mut inner,
190 "k1-launch-nodes callback evidence is duplicated",
191 ));
192 }
193 }
194 let append = inner
195 .file
196 .seek(SeekFrom::End(0))
197 .and_then(|_| inner.file.write_all(&record))
198 .and_then(|()| inner.file.sync_data());
199 if let Err(error) = append {
200 return Err(fault(
201 &mut inner,
202 format!("append k1-launch-nodes projection record: {error}"),
203 ));
204 }
205 inner
206 .bindings
207 .insert(action.target().to_owned(), action.node().to_owned());
208 inner.checkpoint = Some(id);
209 if let Some(pending) = inner.pending.get_mut(&operation_id) {
210 pending.evidence = Some(id);
211 }
212 Ok(())
213 }
214
215 fn reorg(&self) -> Result<(), String> {
216 let mut inner = self
217 .shared
218 .inner
219 .lock()
220 .map_err(|_| "k1-launch-nodes lock poisoned".to_owned())?;
221 inner.failed = true;
222 inner.pending.clear();
223 Ok(())
224 }
225}
226
227impl Shared {
228 fn fail(&self, message: String) -> Result<(), String> {
229 let mut inner = self
230 .inner
231 .lock()
232 .map_err(|_| "k1-launch-nodes lock poisoned".to_owned())?;
233 Err(fault(&mut inner, message))
234 }
235}
236
237struct Projection {
238 file: File,
239 bindings: HashMap<TargetId, NodeId>,
240 checkpoint: Option<TxId>,
241}
242
243fn open_projection(path: &Path) -> Result<Projection, String> {
244 let mut file = OpenOptions::new()
245 .create(true)
246 .truncate(false)
247 .read(true)
248 .write(true)
249 .open(path)
250 .map_err(|error| format!("open launch nodes projection: {error}"))?;
251 if file
252 .metadata()
253 .map_err(|error| format!("inspect launch nodes projection: {error}"))?
254 .len()
255 == 0
256 {
257 reset_projection(&mut file, path)?;
258 }
259 file.seek(SeekFrom::Start(0))
260 .map_err(|error| format!("seek launch nodes projection: {error}"))?;
261 let mut bytes = Vec::new();
262 file.read_to_end(&mut bytes)
263 .map_err(|error| format!("read launch nodes projection: {error}"))?;
264 if bytes.len() < PROJECTION_HEADER.len()
265 || &bytes[..PROJECTION_HEADER.len()] != PROJECTION_HEADER
266 {
267 return Err("unsupported launch nodes projection format; no migration is available".into());
268 }
269 let mut bindings = HashMap::new();
270 let mut checkpoint = None;
271 let mut offset = PROJECTION_HEADER.len();
272 let mut corrupt = false;
273 while offset < bytes.len() {
274 match decode_projection_record(&bytes[offset..]) {
275 Ok(ProjectionDecode::Complete { record, consumed }) => {
276 bindings.insert(record.target().to_owned(), record.node().to_owned());
277 let callback: [u8; 12] = bytes[offset..offset + 12]
278 .try_into()
279 .map_err(|_| "launch nodes projection callback is unavailable".to_owned())?;
280 checkpoint = Some(TxId::from_bytes(callback));
281 offset += consumed;
282 }
283 Ok(ProjectionDecode::Incomplete) => {
284 repair_tail(&mut file, offset)?;
285 break;
286 }
287 Err(_) => {
288 corrupt = true;
289 break;
290 }
291 }
292 }
293 if corrupt {
294 reset_projection(&mut file, path)?;
295 bindings.clear();
296 checkpoint = None;
297 }
298 file.seek(SeekFrom::End(0))
299 .map_err(|error| format!("seek launch nodes append position: {error}"))?;
300 Ok(Projection {
301 file,
302 bindings,
303 checkpoint,
304 })
305}
306
307fn reset_projection(file: &mut File, path: &Path) -> Result<(), String> {
308 file.set_len(0)
309 .and_then(|()| file.seek(SeekFrom::Start(0)).map(|_| ()))
310 .and_then(|()| file.write_all(PROJECTION_HEADER))
311 .and_then(|()| file.sync_data())
312 .map_err(|error| format!("reset launch nodes projection: {error}"))?;
313 sync_parent(path)
314}
315
316fn repair_tail(file: &mut File, offset: usize) -> Result<(), String> {
317 file.set_len(offset as u64)
318 .and_then(|()| file.sync_data())
319 .map_err(|error| format!("repair launch nodes projection tail: {error}"))
320}
321
322fn sync_parent(path: &Path) -> Result<(), String> {
323 let parent = path
324 .parent()
325 .filter(|parent| !parent.as_os_str().is_empty())
326 .unwrap_or_else(|| Path::new("."));
327 File::open(parent)
328 .and_then(|directory| directory.sync_all())
329 .map_err(|error| format!("synchronize launch nodes projection directory: {error}"))
330}
331
332fn subsystem() -> Result<SubsystemId, String> {
333 SubsystemId::from_str("k1-launch-nodes")
334}
335
336fn fault(inner: &mut Inner, message: impl Into<String>) -> String {
337 inner.failed = true;
338 inner.pending.clear();
339 message.into()
340}
341
342#[cfg(test)]
343mod tests;