1#[cfg(test)]
2use self::fs::test_failpoint;
3use self::fs::{
4 atomic_write, ensure_directory, io_error, owner_only_open, path_exists, read_optional_receipt,
5 read_optional_record, read_required_receipt, read_required_record,
6};
7use super::{
8 RuntimeActionKind, RuntimeOperationLease, RuntimeRequestKind, RuntimeRequestReceipt,
9 RuntimeRequestState, RuntimeStateReservation, RuntimeStateStore, RuntimeUnitRecord,
10};
11use crate::contract::{
12 RuntimeActionRequest, RuntimeApplyRequest, RuntimeExecRequest, RuntimeExecResult,
13 RuntimeObservation, RuntimeRemoval, RuntimeUnitState,
14};
15use crate::{RuntimeError, RuntimeResult};
16use async_trait::async_trait;
17use fs2::FileExt;
18use sha2::{Digest, Sha256};
19use std::fs::File;
20use std::path::PathBuf;
21
22mod fs;
23
24#[derive(Debug, Clone)]
25pub struct FileRuntimeStateStore {
26 root: PathBuf,
27}
28
29impl FileRuntimeStateStore {
30 pub fn new(root: impl Into<PathBuf>) -> Self {
31 Self { root: root.into() }
32 }
33
34 fn acquire_operation_lease_sync(&self, unit_id: &str) -> RuntimeResult<FileOperationLease> {
35 validate_unit_id(unit_id)?;
36 ensure_directory(&self.root)?;
37 let operations = self.root.join("operations");
38 ensure_directory(&operations)?;
39 let path = operations.join(format!("{}.lock", storage_key(unit_id)));
40 let file = owner_only_open(&path, "Runtime operation lease")?;
41 file.lock_exclusive()
42 .map_err(io_error("lock Runtime operation"))?;
43 Ok(FileOperationLease(file))
44 }
45
46 fn reserve_apply_sync(
47 &self,
48 request: RuntimeApplyRequest,
49 now_ms: u64,
50 ) -> RuntimeResult<RuntimeStateReservation> {
51 request.validate().map_err(RuntimeError::InvalidRequest)?;
52 let _lock = self.lock(&request.spec.unit_id)?;
53 let record_path = self.record_path(&request.spec.unit_id, true)?;
54 let mut receipt_path =
55 self.request_path(&request.spec.unit_id, &request.request_id, false)?;
56 let existing = read_optional_receipt(&receipt_path)?;
57 if let Some(receipt) = &existing {
58 let digest = request.digest().map_err(RuntimeError::InvalidRequest)?;
59 ensure_same_request(receipt, RuntimeRequestKind::Apply, &digest)?;
60 ensure_receipt_target(receipt, &request.spec.unit_id, request.spec.generation)?;
61 }
62
63 let stored_record = read_optional_record(&record_path)?;
64 if stored_record.is_none()
65 && existing
66 .as_ref()
67 .is_some_and(|receipt| receipt.state == RuntimeRequestState::Completed)
68 {
69 return Err(RuntimeError::Protocol(
70 "completed Runtime apply receipt has no unit record".into(),
71 ));
72 }
73 let mut record_changed = stored_record.is_none();
74 let mut record = match stored_record {
75 Some(record) => record,
76 None => RuntimeUnitRecord::new(&request, now_ms).map_err(RuntimeError::Protocol)?,
77 };
78
79 if let Some(receipt) = existing
80 .as_ref()
81 .filter(|receipt| receipt.state == RuntimeRequestState::Completed)
82 {
83 if reconcile_completed_observation(&mut record, receipt)? {
84 atomic_write(&record_path, &record, "state record")?;
85 }
86 return Ok(reservation(record, receipt.clone()));
87 }
88
89 let current_generation = record.spec.generation;
90 if request.spec.generation < current_generation {
91 return Err(RuntimeError::StaleGeneration {
92 unit_id: request.spec.unit_id,
93 requested: request.spec.generation,
94 current: current_generation,
95 });
96 }
97
98 let receipt_is_new = existing.is_none();
99 let mut receipt = existing.unwrap_or(
100 RuntimeRequestReceipt::pending_apply(&request).map_err(RuntimeError::Protocol)?,
101 );
102
103 if request.spec.generation == current_generation {
104 if request
105 .spec
106 .digest()
107 .map_err(RuntimeError::InvalidRequest)?
108 != record.spec.digest().map_err(RuntimeError::Protocol)?
109 || record.removed_at_ms.is_some()
110 {
111 return Err(RuntimeError::GenerationConflict {
112 unit_id: request.spec.unit_id,
113 generation: request.spec.generation,
114 });
115 }
116 if !matches!(
117 record.observation.state,
118 RuntimeUnitState::Accepted | RuntimeUnitState::Unknown
119 ) {
120 receipt.complete_with_observation(record.observation.clone());
121 receipt.validate().map_err(RuntimeError::Protocol)?;
122 }
123 } else {
124 record.spec = request.spec.clone();
125 record.observation = RuntimeObservation::accepted(&request.spec, now_ms)
126 .map_err(RuntimeError::Protocol)?;
127 record.removed_at_ms = None;
128 record_changed = true;
129 }
130
131 if receipt_is_new || receipt.state == RuntimeRequestState::Completed {
132 if receipt_is_new {
133 receipt_path =
134 self.request_path(&request.spec.unit_id, &request.request_id, true)?;
135 }
136 atomic_write(&receipt_path, &receipt, "request receipt")?;
137 }
138 record.validate().map_err(RuntimeError::Protocol)?;
139 if record_changed {
140 atomic_write(&record_path, &record, "state record")?;
141 }
142 Ok(reservation(record, receipt))
143 }
144
145 fn reserve_action_sync(
146 &self,
147 kind: RuntimeActionKind,
148 request: RuntimeActionRequest,
149 now_ms: u64,
150 ) -> RuntimeResult<RuntimeStateReservation> {
151 request.validate().map_err(RuntimeError::InvalidRequest)?;
152 let _lock = self.lock(&request.unit_id)?;
153 let record_path = self.record_path(&request.unit_id, false)?;
154 let mut record = read_required_record(&record_path, &request.unit_id)?;
155 let mut receipt_path = self.request_path(&request.unit_id, &request.request_id, false)?;
156 if let Some(receipt) = read_optional_receipt(&receipt_path)? {
157 let digest = request.digest().map_err(RuntimeError::InvalidRequest)?;
158 ensure_same_request(&receipt, kind.into(), &digest)?;
159 ensure_receipt_target(&receipt, &request.unit_id, request.generation)?;
160 if receipt.state == RuntimeRequestState::Completed {
161 let changed = match receipt.kind {
162 RuntimeRequestKind::Stop => {
163 reconcile_completed_observation(&mut record, &receipt)?
164 }
165 RuntimeRequestKind::Remove => {
166 reconcile_completed_removal(&mut record, &receipt)?
167 }
168 _ => false,
169 };
170 if changed {
171 atomic_write(&record_path, &record, "state record")?;
172 }
173 } else {
174 ensure_current_generation(&record, &request.unit_id, request.generation)?;
175 if record.removed_at_ms.is_some() {
176 return Err(RuntimeError::NotFound {
177 unit_id: request.unit_id,
178 });
179 }
180 }
181 return Ok(reservation(record, receipt));
182 }
183
184 ensure_current_generation(&record, &request.unit_id, request.generation)?;
185 let mut receipt = RuntimeRequestReceipt::pending_action(kind, &request)
186 .map_err(RuntimeError::Protocol)?;
187 match kind {
188 RuntimeActionKind::Stop => {
189 if record.removed_at_ms.is_some() {
190 return Err(RuntimeError::NotFound {
191 unit_id: request.unit_id,
192 });
193 }
194 if record.observation.state.is_terminal() {
195 receipt.complete_with_observation(record.observation.clone());
196 }
197 }
198 RuntimeActionKind::Remove => {
199 if record.removed_at_ms.is_some() {
200 receipt.complete_with_removal(RuntimeRemoval {
201 schema: RuntimeRemoval::SCHEMA.into(),
202 request_id: request.request_id.clone(),
203 unit_id: request.unit_id.clone(),
204 generation: request.generation,
205 removed_at_ms: now_ms,
206 already_absent: true,
207 });
208 }
209 }
210 }
211 receipt.validate().map_err(RuntimeError::Protocol)?;
212 receipt_path = self.request_path(&request.unit_id, &request.request_id, true)?;
213 atomic_write(&receipt_path, &receipt, "request receipt")?;
214 Ok(reservation(record, receipt))
215 }
216
217 fn reserve_exec_sync(
218 &self,
219 request: RuntimeExecRequest,
220 started_at_ms: u64,
221 ) -> RuntimeResult<RuntimeStateReservation> {
222 request.validate().map_err(RuntimeError::InvalidRequest)?;
223 let _lock = self.lock(&request.unit_id)?;
224 let record_path = self.record_path(&request.unit_id, false)?;
225 let mut record = read_required_record(&record_path, &request.unit_id)?;
226 let mut receipt_path = self.request_path(&request.unit_id, &request.request_id, false)?;
227 if let Some(receipt) = read_optional_receipt(&receipt_path)? {
228 let digest = request.digest().map_err(RuntimeError::InvalidRequest)?;
229 ensure_same_request(&receipt, RuntimeRequestKind::Exec, &digest)?;
230 ensure_receipt_target(&receipt, &request.unit_id, request.generation)?;
231 if receipt.state == RuntimeRequestState::Completed
232 && reconcile_completed_exec(&mut record, &receipt)?
233 {
234 atomic_write(&record_path, &record, "state record")?;
235 } else if receipt.state == RuntimeRequestState::Pending {
236 ensure_current_generation(&record, &request.unit_id, request.generation)?;
237 if record.removed_at_ms.is_some() {
238 return Err(RuntimeError::NotFound {
239 unit_id: request.unit_id,
240 });
241 }
242 }
243 return Ok(reservation(record, receipt));
244 }
245
246 ensure_current_generation(&record, &request.unit_id, request.generation)?;
247 if record.removed_at_ms.is_some() {
248 return Err(RuntimeError::NotFound {
249 unit_id: request.unit_id,
250 });
251 }
252 if record.observation.state != RuntimeUnitState::Running {
253 return Err(RuntimeError::InvalidRequest(format!(
254 "Runtime exec requires a running unit; {:?} is {:?}",
255 request.unit_id, record.observation.state
256 )));
257 }
258 let receipt = RuntimeRequestReceipt::pending_exec(&request, started_at_ms)
259 .map_err(RuntimeError::Protocol)?;
260 receipt.validate().map_err(RuntimeError::Protocol)?;
261 receipt_path = self.request_path(&request.unit_id, &request.request_id, true)?;
262 atomic_write(&receipt_path, &receipt, "request receipt")?;
263 Ok(reservation(record, receipt))
264 }
265
266 fn load_sync(&self, unit_id: &str) -> RuntimeResult<RuntimeUnitRecord> {
267 validate_unit_id(unit_id)?;
268 let _lock = self.lock(unit_id)?;
269 let path = self.record_path(unit_id, false)?;
270 read_required_record(&path, unit_id)
271 }
272
273 fn load_request_sync(
274 &self,
275 unit_id: &str,
276 request_id: &str,
277 ) -> RuntimeResult<RuntimeRequestReceipt> {
278 validate_unit_id(unit_id)?;
279 validate_request_id(request_id)?;
280 let _lock = self.lock(unit_id)?;
281 let path = self.request_path(unit_id, request_id, false)?;
282 let receipt =
283 read_optional_receipt(&path)?.ok_or_else(|| RuntimeError::RequestNotFound {
284 unit_id: unit_id.into(),
285 request_id: request_id.into(),
286 })?;
287 if receipt.unit_id != unit_id || receipt.request_id != request_id {
288 return Err(RuntimeError::Protocol(
289 "Runtime request receipt storage key mismatch".into(),
290 ));
291 }
292 Ok(receipt)
293 }
294
295 fn update_observation_sync(
296 &self,
297 request_id: Option<String>,
298 observation: RuntimeObservation,
299 ) -> RuntimeResult<RuntimeUnitRecord> {
300 observation.validate().map_err(RuntimeError::Protocol)?;
301 let _lock = self.lock(&observation.unit_id)?;
302 let record_path = self.record_path(&observation.unit_id, false)?;
303 let mut record = read_required_record(&record_path, &observation.unit_id)?;
304 if record.removed_at_ms.is_some() {
305 return Err(RuntimeError::Protocol(
306 "cannot update an explicitly removed Runtime unit".into(),
307 ));
308 }
309 validate_transition(&record.observation, &observation, &record.spec)?;
310
311 if let Some(request_id) = request_id {
312 let receipt_path = self.request_path(&observation.unit_id, &request_id, false)?;
313 let mut receipt =
314 read_required_receipt(&receipt_path, &observation.unit_id, &request_id)?;
315 if !matches!(
316 receipt.kind,
317 RuntimeRequestKind::Apply | RuntimeRequestKind::Stop
318 ) {
319 return Err(RuntimeError::Protocol(
320 "request kind cannot complete with an observation".into(),
321 ));
322 }
323 ensure_receipt_target(&receipt, &observation.unit_id, observation.generation)?;
324 if receipt.state == RuntimeRequestState::Completed {
325 if receipt.observation.as_ref() != Some(&observation) {
326 return Err(RuntimeError::Protocol(
327 "completed Runtime request result changed".into(),
328 ));
329 }
330 } else {
331 receipt.complete_with_observation(observation.clone());
332 receipt.validate().map_err(RuntimeError::Protocol)?;
333 atomic_write(&receipt_path, &receipt, "request receipt")?;
334 #[cfg(test)]
335 test_failpoint("state.complete-observation.after-receipt-publish");
336 }
337 }
338
339 record.observation = observation;
340 record.validate().map_err(RuntimeError::Protocol)?;
341 atomic_write(&record_path, &record, "state record")?;
342 Ok(record)
343 }
344
345 fn complete_removal_sync(&self, removal: RuntimeRemoval) -> RuntimeResult<RuntimeUnitRecord> {
346 removal.validate().map_err(RuntimeError::Protocol)?;
347 let _lock = self.lock(&removal.unit_id)?;
348 let record_path = self.record_path(&removal.unit_id, false)?;
349 let mut record = read_required_record(&record_path, &removal.unit_id)?;
350 if removal.generation != record.spec.generation {
351 return Err(RuntimeError::Protocol(
352 "Runtime removal generation does not match stored unit".into(),
353 ));
354 }
355 let receipt_path = self.request_path(&removal.unit_id, &removal.request_id, false)?;
356 let mut receipt =
357 read_required_receipt(&receipt_path, &removal.unit_id, &removal.request_id)?;
358 if receipt.kind != RuntimeRequestKind::Remove {
359 return Err(RuntimeError::Protocol(
360 "non-remove request completed with removal result".into(),
361 ));
362 }
363 ensure_receipt_target(&receipt, &removal.unit_id, removal.generation)?;
364 if receipt.state == RuntimeRequestState::Completed {
365 if receipt.removal.as_ref() != Some(&removal) {
366 return Err(RuntimeError::Protocol(
367 "completed Runtime removal result changed".into(),
368 ));
369 }
370 } else {
371 receipt.complete_with_removal(removal.clone());
372 receipt.validate().map_err(RuntimeError::Protocol)?;
373 atomic_write(&receipt_path, &receipt, "request receipt")?;
374 #[cfg(test)]
375 test_failpoint("state.complete-removal.after-receipt-publish");
376 }
377 record.removed_at_ms = Some(removal.removed_at_ms);
378 record.validate().map_err(RuntimeError::Protocol)?;
379 atomic_write(&record_path, &record, "state record")?;
380 Ok(record)
381 }
382
383 fn complete_exec_sync(&self, result: RuntimeExecResult) -> RuntimeResult<RuntimeUnitRecord> {
384 result.validate().map_err(RuntimeError::Protocol)?;
385 let unit_id = result.observation.unit_id.clone();
386 let _lock = self.lock(&unit_id)?;
387 let record_path = self.record_path(&unit_id, false)?;
388 let mut record = read_required_record(&record_path, &unit_id)?;
389 if record.removed_at_ms.is_some() {
390 return Err(RuntimeError::Protocol(
391 "cannot complete exec for an explicitly removed Runtime unit".into(),
392 ));
393 }
394 result
395 .observation
396 .validate_against(&record.spec)
397 .map_err(RuntimeError::Protocol)?;
398 validate_transition(&record.observation, &result.observation, &record.spec)?;
399 let receipt_path = self.request_path(&unit_id, &result.request_id, false)?;
400 let mut receipt = read_required_receipt(&receipt_path, &unit_id, &result.request_id)?;
401 if receipt.kind != RuntimeRequestKind::Exec {
402 return Err(RuntimeError::Protocol(
403 "non-exec request completed with exec result".into(),
404 ));
405 }
406 if receipt.state == RuntimeRequestState::Completed {
407 if receipt.exec_result.as_ref() != Some(&result) {
408 return Err(RuntimeError::Protocol(
409 "completed Runtime exec result changed".into(),
410 ));
411 }
412 } else {
413 receipt.complete_with_exec_result(result.clone());
414 receipt.validate().map_err(RuntimeError::Protocol)?;
415 atomic_write(&receipt_path, &receipt, "request receipt")?;
416 #[cfg(test)]
417 test_failpoint("state.complete-exec.after-receipt-publish");
418 }
419 record.observation = result.observation;
420 record.validate().map_err(RuntimeError::Protocol)?;
421 atomic_write(&record_path, &record, "state record")?;
422 Ok(record)
423 }
424
425 fn lock(&self, unit_id: &str) -> RuntimeResult<StateLock> {
426 validate_unit_id(unit_id)?;
427 ensure_directory(&self.root)?;
428 let locks = self.root.join("locks");
429 ensure_directory(&locks)?;
430 let file = owner_only_open(
431 &locks.join(format!("{}.lock", storage_key(unit_id))),
432 "Runtime state lock",
433 )?;
434 file.lock_exclusive()
435 .map_err(io_error("lock Runtime unit"))?;
436 Ok(StateLock(file))
437 }
438
439 fn unit_directory(&self, unit_id: &str, create: bool) -> RuntimeResult<PathBuf> {
440 validate_unit_id(unit_id)?;
441 let units = self.root.join("units");
442 if create {
443 ensure_directory(&self.root)?;
444 ensure_directory(&units)?;
445 } else if path_exists(&units)? {
446 ensure_directory(&units)?;
447 }
448 let key = storage_key(unit_id);
449 let legacy = units.join(format!("{key}.json"));
450 if path_exists(&legacy)? {
451 return Err(RuntimeError::Protocol(format!(
452 "legacy Runtime unit record {} requires explicit migration",
453 legacy.display()
454 )));
455 }
456 let unit = units.join(key);
457 if create || path_exists(&unit)? {
458 ensure_directory(&unit)?;
459 }
460 Ok(unit)
461 }
462
463 fn record_path(&self, unit_id: &str, create: bool) -> RuntimeResult<PathBuf> {
464 Ok(self.unit_directory(unit_id, create)?.join("record.json"))
465 }
466
467 fn request_path(
468 &self,
469 unit_id: &str,
470 request_id: &str,
471 create: bool,
472 ) -> RuntimeResult<PathBuf> {
473 validate_request_id(request_id)?;
474 let requests = self.unit_directory(unit_id, create)?.join("requests");
475 if create || path_exists(&requests)? {
476 ensure_directory(&requests)?;
477 }
478 Ok(requests.join(format!("{}.json", storage_key(request_id))))
479 }
480}
481
482#[async_trait]
483impl RuntimeStateStore for FileRuntimeStateStore {
484 async fn acquire_operation_lease(
485 &self,
486 unit_id: &str,
487 ) -> RuntimeResult<Box<dyn RuntimeOperationLease>> {
488 let store = self.clone();
489 let unit_id = unit_id.to_owned();
490 let lease =
491 tokio::task::spawn_blocking(move || store.acquire_operation_lease_sync(&unit_id))
492 .await
493 .map_err(task_error)??;
494 Ok(Box::new(lease))
495 }
496
497 async fn reserve_apply(
498 &self,
499 request: &RuntimeApplyRequest,
500 now_ms: u64,
501 ) -> RuntimeResult<RuntimeStateReservation> {
502 let store = self.clone();
503 let request = request.clone();
504 tokio::task::spawn_blocking(move || store.reserve_apply_sync(request, now_ms))
505 .await
506 .map_err(task_error)?
507 }
508
509 async fn reserve_action(
510 &self,
511 kind: RuntimeActionKind,
512 request: &RuntimeActionRequest,
513 now_ms: u64,
514 ) -> RuntimeResult<RuntimeStateReservation> {
515 let store = self.clone();
516 let request = request.clone();
517 tokio::task::spawn_blocking(move || store.reserve_action_sync(kind, request, now_ms))
518 .await
519 .map_err(task_error)?
520 }
521
522 async fn reserve_exec(
523 &self,
524 request: &RuntimeExecRequest,
525 now_ms: u64,
526 ) -> RuntimeResult<RuntimeStateReservation> {
527 let store = self.clone();
528 let request = request.clone();
529 tokio::task::spawn_blocking(move || store.reserve_exec_sync(request, now_ms))
530 .await
531 .map_err(task_error)?
532 }
533
534 async fn load(&self, unit_id: &str) -> RuntimeResult<RuntimeUnitRecord> {
535 let store = self.clone();
536 let unit_id = unit_id.to_owned();
537 tokio::task::spawn_blocking(move || store.load_sync(&unit_id))
538 .await
539 .map_err(task_error)?
540 }
541
542 async fn load_request(
543 &self,
544 unit_id: &str,
545 request_id: &str,
546 ) -> RuntimeResult<RuntimeRequestReceipt> {
547 let store = self.clone();
548 let unit_id = unit_id.to_owned();
549 let request_id = request_id.to_owned();
550 tokio::task::spawn_blocking(move || store.load_request_sync(&unit_id, &request_id))
551 .await
552 .map_err(task_error)?
553 }
554
555 async fn update_observation(
556 &self,
557 request_id: Option<&str>,
558 observation: &RuntimeObservation,
559 ) -> RuntimeResult<RuntimeUnitRecord> {
560 let store = self.clone();
561 let request_id = request_id.map(str::to_owned);
562 let observation = observation.clone();
563 tokio::task::spawn_blocking(move || store.update_observation_sync(request_id, observation))
564 .await
565 .map_err(task_error)?
566 }
567
568 async fn complete_removal(&self, removal: &RuntimeRemoval) -> RuntimeResult<RuntimeUnitRecord> {
569 let store = self.clone();
570 let removal = removal.clone();
571 tokio::task::spawn_blocking(move || store.complete_removal_sync(removal))
572 .await
573 .map_err(task_error)?
574 }
575
576 async fn complete_exec(&self, result: &RuntimeExecResult) -> RuntimeResult<RuntimeUnitRecord> {
577 let store = self.clone();
578 let result = result.clone();
579 tokio::task::spawn_blocking(move || store.complete_exec_sync(result))
580 .await
581 .map_err(task_error)?
582 }
583}
584
585fn reservation(
586 record: RuntimeUnitRecord,
587 receipt: RuntimeRequestReceipt,
588) -> RuntimeStateReservation {
589 RuntimeStateReservation {
590 dispatch: receipt.state == RuntimeRequestState::Pending,
591 record,
592 receipt,
593 }
594}
595
596fn ensure_same_request(
597 receipt: &RuntimeRequestReceipt,
598 kind: RuntimeRequestKind,
599 digest: &str,
600) -> RuntimeResult<()> {
601 if receipt.kind != kind || receipt.request_digest != digest {
602 return Err(RuntimeError::RequestConflict {
603 request_id: receipt.request_id.clone(),
604 });
605 }
606 Ok(())
607}
608
609fn ensure_receipt_target(
610 receipt: &RuntimeRequestReceipt,
611 unit_id: &str,
612 generation: u64,
613) -> RuntimeResult<()> {
614 if receipt.unit_id != unit_id || receipt.generation != generation {
615 return Err(RuntimeError::RequestConflict {
616 request_id: receipt.request_id.clone(),
617 });
618 }
619 Ok(())
620}
621
622fn ensure_current_generation(
623 record: &RuntimeUnitRecord,
624 unit_id: &str,
625 requested: u64,
626) -> RuntimeResult<()> {
627 if requested < record.spec.generation {
628 return Err(RuntimeError::StaleGeneration {
629 unit_id: unit_id.into(),
630 requested,
631 current: record.spec.generation,
632 });
633 }
634 if requested != record.spec.generation {
635 return Err(RuntimeError::GenerationConflict {
636 unit_id: unit_id.into(),
637 generation: requested,
638 });
639 }
640 Ok(())
641}
642
643fn reconcile_completed_observation(
644 record: &mut RuntimeUnitRecord,
645 receipt: &RuntimeRequestReceipt,
646) -> RuntimeResult<bool> {
647 let Some(observation) = &receipt.observation else {
648 return Ok(false);
649 };
650 if record.removed_at_ms.is_some()
651 || record.spec.generation != receipt.generation
652 || record.spec.unit_id != receipt.unit_id
653 || record.observation == *observation
654 || record.observation.observed_at_ms > observation.observed_at_ms
655 {
656 return Ok(false);
657 }
658 if validate_transition(&record.observation, observation, &record.spec).is_ok() {
659 record.observation = observation.clone();
660 record.validate().map_err(RuntimeError::Protocol)?;
661 return Ok(true);
662 }
663 Ok(false)
664}
665
666fn reconcile_completed_removal(
667 record: &mut RuntimeUnitRecord,
668 receipt: &RuntimeRequestReceipt,
669) -> RuntimeResult<bool> {
670 let Some(removal) = &receipt.removal else {
671 return Ok(false);
672 };
673 if record.spec.unit_id == receipt.unit_id
674 && record.spec.generation == receipt.generation
675 && record.removed_at_ms.is_none()
676 {
677 record.removed_at_ms = Some(removal.removed_at_ms);
678 record.validate().map_err(RuntimeError::Protocol)?;
679 return Ok(true);
680 }
681 Ok(false)
682}
683
684fn reconcile_completed_exec(
685 record: &mut RuntimeUnitRecord,
686 receipt: &RuntimeRequestReceipt,
687) -> RuntimeResult<bool> {
688 let Some(result) = &receipt.exec_result else {
689 return Ok(false);
690 };
691 if record.removed_at_ms.is_some()
692 || record.spec.generation != receipt.generation
693 || record.spec.unit_id != receipt.unit_id
694 || record.observation == result.observation
695 || record.observation.observed_at_ms > result.observation.observed_at_ms
696 {
697 return Ok(false);
698 }
699 if validate_transition(&record.observation, &result.observation, &record.spec).is_ok() {
700 record.observation = result.observation.clone();
701 record.validate().map_err(RuntimeError::Protocol)?;
702 return Ok(true);
703 }
704 Ok(false)
705}
706
707pub(crate) fn validate_transition(
708 current: &RuntimeObservation,
709 next: &RuntimeObservation,
710 spec: &crate::contract::RuntimeUnitSpec,
711) -> RuntimeResult<()> {
712 current
713 .validate_against(spec)
714 .map_err(RuntimeError::Protocol)?;
715 next.validate_against(spec)
716 .map_err(RuntimeError::Protocol)?;
717 if current.state != RuntimeUnitState::Unknown
718 && current.provider_resource_id.is_some()
719 && current.provider_resource_id != next.provider_resource_id
720 {
721 return Err(RuntimeError::Protocol(
722 "Runtime update changes provider resource identity".into(),
723 ));
724 }
725 if next.observed_at_ms < current.observed_at_ms {
726 return Err(RuntimeError::Protocol(
727 "Runtime observation time moved backwards".into(),
728 ));
729 }
730 if current.state.is_terminal() {
731 if current != next {
732 return Err(RuntimeError::Protocol(
733 "terminal Runtime observation is immutable".into(),
734 ));
735 }
736 return Ok(());
737 }
738 let allowed = current.state == next.state
739 || matches!(
740 (current.state, next.state),
741 (RuntimeUnitState::Accepted, RuntimeUnitState::Preparing)
742 | (RuntimeUnitState::Accepted, RuntimeUnitState::Starting)
743 | (RuntimeUnitState::Accepted, RuntimeUnitState::Running)
744 | (RuntimeUnitState::Accepted, RuntimeUnitState::Succeeded)
745 | (RuntimeUnitState::Accepted, RuntimeUnitState::Failed)
746 | (RuntimeUnitState::Accepted, RuntimeUnitState::Stopped)
747 | (RuntimeUnitState::Accepted, RuntimeUnitState::Unknown)
748 | (RuntimeUnitState::Preparing, RuntimeUnitState::Starting)
749 | (RuntimeUnitState::Preparing, RuntimeUnitState::Running)
750 | (RuntimeUnitState::Preparing, RuntimeUnitState::Succeeded)
751 | (RuntimeUnitState::Preparing, RuntimeUnitState::Stopping)
752 | (RuntimeUnitState::Preparing, RuntimeUnitState::Stopped)
753 | (RuntimeUnitState::Preparing, RuntimeUnitState::Failed)
754 | (RuntimeUnitState::Preparing, RuntimeUnitState::Unknown)
755 | (RuntimeUnitState::Starting, RuntimeUnitState::Running)
756 | (RuntimeUnitState::Starting, RuntimeUnitState::Succeeded)
757 | (RuntimeUnitState::Starting, RuntimeUnitState::Stopping)
758 | (RuntimeUnitState::Starting, RuntimeUnitState::Stopped)
759 | (RuntimeUnitState::Starting, RuntimeUnitState::Failed)
760 | (RuntimeUnitState::Starting, RuntimeUnitState::Unknown)
761 | (RuntimeUnitState::Running, RuntimeUnitState::Stopping)
762 | (RuntimeUnitState::Running, RuntimeUnitState::Stopped)
763 | (RuntimeUnitState::Running, RuntimeUnitState::Succeeded)
764 | (RuntimeUnitState::Running, RuntimeUnitState::Failed)
765 | (RuntimeUnitState::Running, RuntimeUnitState::Unknown)
766 | (RuntimeUnitState::Stopping, RuntimeUnitState::Stopped)
767 | (RuntimeUnitState::Stopping, RuntimeUnitState::Failed)
768 | (RuntimeUnitState::Stopping, RuntimeUnitState::Unknown)
769 | (RuntimeUnitState::Unknown, RuntimeUnitState::Preparing)
770 | (RuntimeUnitState::Unknown, RuntimeUnitState::Starting)
771 | (RuntimeUnitState::Unknown, RuntimeUnitState::Running)
772 | (RuntimeUnitState::Unknown, RuntimeUnitState::Stopping)
773 | (RuntimeUnitState::Unknown, RuntimeUnitState::Stopped)
774 | (RuntimeUnitState::Unknown, RuntimeUnitState::Succeeded)
775 | (RuntimeUnitState::Unknown, RuntimeUnitState::Failed)
776 );
777 if !allowed {
778 return Err(RuntimeError::Protocol(format!(
779 "invalid Runtime transition {:?} -> {:?}",
780 current.state, next.state
781 )));
782 }
783 Ok(())
784}
785
786struct StateLock(File);
787
788impl Drop for StateLock {
789 fn drop(&mut self) {
790 let _ = FileExt::unlock(&self.0);
791 }
792}
793
794struct FileOperationLease(File);
795
796impl RuntimeOperationLease for FileOperationLease {}
797
798impl Drop for FileOperationLease {
799 fn drop(&mut self) {
800 let _ = FileExt::unlock(&self.0);
801 }
802}
803
804fn storage_key(value: &str) -> String {
805 format!("{:x}", Sha256::digest(value.as_bytes()))
806}
807
808fn validate_unit_id(unit_id: &str) -> RuntimeResult<()> {
809 crate::contract::validate_id("unit_id", unit_id, 512).map_err(RuntimeError::InvalidRequest)
810}
811
812fn validate_request_id(request_id: &str) -> RuntimeResult<()> {
813 crate::contract::validate_id("request_id", request_id, 512)
814 .map_err(RuntimeError::InvalidRequest)
815}
816
817fn task_error(error: tokio::task::JoinError) -> RuntimeError {
818 RuntimeError::Transport(format!("Runtime state task failed: {error}"))
819}
820
821#[cfg(test)]
822mod tests;