1use std::collections::{BTreeMap, BTreeSet};
2use std::error::Error;
3use std::fmt;
4use std::fs::{self, File};
5use std::io::{self, Read, Write};
6use std::path::{Path, PathBuf};
7use std::sync::Arc;
8
9use cap_std::ambient_authority;
10use cap_std::fs::Dir;
11use vsh_store::{
12 BlobStore, BlobStoreError, CommitReservation, DataDirectory, DataDirectoryError,
13 TransactionStore, TransactionStoreError,
14};
15use vsh_types::{
16 BlobId, ContentVersion, DirectoryDigest, NodeKind, NodeState, PlatformFileId, TransactionId,
17 TransactionState, VPath,
18};
19use vsh_vfs::{BaseSnapshot, ReadObservation};
20
21use crate::host::{
22 HostError, SnapshotLimits, capture_snapshot, content_digest, create_new_file,
23 create_staged_symlink, directory_digest, open_coordination_file, open_or_create_real_dir,
24 open_real_dir, open_real_file, relative_path, relocated_state_matches, set_dir_mode,
25 set_file_mode, stamp_at, stamp_dir, stamp_file, state_matches, sync_dir, sync_installed_file,
26 validate_symlink_target, witness_matches,
27};
28use crate::journal::{
29 JOURNAL_FILE, Journal, JournalError, JournalState, PLAN_FILE, QUARANTINE_DIRECTORY,
30 STAGE_DIRECTORY, Witness, has_valid_commit_marker, read_journal, write_commit_marker,
31};
32use crate::plan::{
33 CommitPlan, CommitPlanError, Operation, PlanDecodeError, PreparedPlan, quarantine_name,
34 stage_link_name, stage_name,
35};
36
37const DIRECTORY_OWNER_MARKER: &str = ".vsh-runtime-owner";
38const DIRECTORY_OWNER_MAGIC: &[u8; 8] = b"VSHOWN01";
39const COMMIT_LOCK_FILE: &str = "commit.lock";
40
41fn paths_overlap(left: &Path, right: &Path) -> bool {
42 left.starts_with(right) || right.starts_with(left)
43}
44
45struct WorkspaceLockGuard<'a>(&'a File);
46
47impl<'a> WorkspaceLockGuard<'a> {
48 fn shared(file: &'a File) -> io::Result<Self> {
49 File::lock_shared(file)?;
50 Ok(Self(file))
51 }
52
53 fn exclusive(file: &'a File) -> io::Result<Self> {
54 File::lock(file)?;
55 Ok(Self(file))
56 }
57}
58
59impl Drop for WorkspaceLockGuard<'_> {
60 fn drop(&mut self) {
61 let _ = File::unlock(self.0);
62 }
63}
64
65#[derive(Clone, Copy)]
66struct OperationWitnesses {
67 completed: Option<Witness>,
68 source: Option<Witness>,
69 parent: Option<Witness>,
70}
71
72struct AppliedOperation {
73 witness: Witness,
74 created_directory: Option<(VPath, Dir)>,
75}
76
77#[derive(Clone, Copy, Debug, Eq, PartialEq)]
79pub struct CommitConfig {
80 pub max_operations: usize,
82 pub max_dependencies: usize,
84 pub max_path_bytes: usize,
86 pub max_plan_bytes: usize,
88 pub max_journal_bytes: usize,
90 pub max_conflicts: usize,
92}
93
94impl Default for CommitConfig {
95 fn default() -> Self {
96 Self {
97 max_operations: 100_000,
98 max_dependencies: 250_000,
99 max_path_bytes: 16 * 1024,
100 max_plan_bytes: 128 * 1024 * 1024,
101 max_journal_bytes: 64 * 1024 * 1024,
102 max_conflicts: 128,
103 }
104 }
105}
106
107#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
109#[non_exhaustive]
110pub enum FaultPoint {
111 PlanSynced,
113 StageSynced,
115 Revalidated,
117 CommitStatePersisted,
119 IntentSynced(u32),
121 OperationApplied(u32),
123 DoneSynced(u32),
125 OwnershipMarkersCleared,
127 Verified,
129 CommitMarkerSynced,
131 CommittedStatePersisted,
133}
134
135pub trait FaultInjector: Send + Sync {
137 fn should_fail(&self, point: FaultPoint) -> bool;
139}
140
141impl<F> FaultInjector for F
142where
143 F: Fn(FaultPoint) -> bool + Send + Sync,
144{
145 fn should_fail(&self, point: FaultPoint) -> bool {
146 self(point)
147 }
148}
149
150#[derive(Clone, Copy, Debug, Default)]
152pub struct NoFaults;
153
154impl FaultInjector for NoFaults {
155 fn should_fail(&self, _point: FaultPoint) -> bool {
156 false
157 }
158}
159
160#[derive(Clone, Debug, Eq, PartialEq)]
162#[non_exhaustive]
163pub enum RevalidationConflict {
164 Metadata {
166 path: VPath,
168 expected: Option<NodeState>,
170 actual: Option<NodeState>,
172 },
173 Content {
175 path: VPath,
177 expected: BlobId,
179 actual: BlobId,
181 },
182 Directory {
184 path: VPath,
186 expected: DirectoryDigest,
188 actual: DirectoryDigest,
190 },
191}
192
193#[derive(Clone, Debug, Eq, PartialEq)]
195pub struct CommitReceipt {
196 pub transaction: TransactionId,
198 pub operations: usize,
200 pub verified_paths: usize,
202 pub cleanup_pending: bool,
204}
205
206#[derive(Clone, Debug, Default, Eq, PartialEq)]
208pub struct RecoveryReport {
209 pub finalized_commits: usize,
211 pub rolled_back: usize,
213 pub cleaned: usize,
215 pub orphaned: usize,
217 pub conflicts: Vec<RecoveryConflict>,
219}
220
221#[derive(Clone, Debug, Eq, PartialEq)]
223pub struct RecoveryConflict {
224 pub transaction: TransactionId,
226 pub path: Option<VPath>,
228 pub reason: &'static str,
230}
231
232#[derive(Clone, Debug, Eq, PartialEq)]
234pub struct VerificationFailure {
235 pub path: VPath,
237 pub expected: Option<NodeState>,
239 pub actual: Option<NodeState>,
241}
242
243#[derive(Debug)]
245#[non_exhaustive]
246pub enum CommitError {
247 Plan(CommitPlanError),
249 Binding {
251 reserved_transaction: TransactionId,
253 plan_transaction: TransactionId,
255 },
256 BaseSnapshotBinding,
258 DependencyLimit {
260 observed: usize,
262 maximum: usize,
264 },
265 PlanSize {
267 observed: usize,
269 maximum: usize,
271 },
272 TransactionWorkspaceExists {
274 transaction: TransactionId,
276 },
277 UnsafeBlobStore {
279 workspace_root: PathBuf,
281 blobs_directory: PathBuf,
283 },
284 Host(HostError),
286 Store(TransactionStoreError),
288 Blob(BlobStoreError),
290 DataDirectory(DataDirectoryError),
292 Journal(JournalError),
294 PlanDecode(PlanDecodeError),
296 InternalIo {
298 operation: &'static str,
300 source: io::Error,
302 },
303 Stale {
305 conflicts: Vec<RevalidationConflict>,
307 },
308 Verification(Box<VerificationFailure>),
310 FaultInjected {
312 point: FaultPoint,
314 },
315 RecoveryRequired {
317 transaction: TransactionId,
319 cause: String,
321 },
322 RecoveryConflict(RecoveryConflict),
324 InvalidRecoveryState {
326 transaction: TransactionId,
328 state: TransactionState,
330 },
331}
332
333impl fmt::Display for CommitError {
334 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
335 match self {
336 Self::Plan(source) => fmt::Display::fmt(source, formatter),
337 Self::Binding {
338 reserved_transaction,
339 plan_transaction,
340 } => write!(
341 formatter,
342 "commit reservation {reserved_transaction} does not bind plan {plan_transaction}"
343 ),
344 Self::BaseSnapshotBinding => {
345 formatter.write_str("commit reservation and plan bind different base snapshots")
346 }
347 Self::DependencyLimit { observed, maximum } => write!(
348 formatter,
349 "commit has {observed} dependencies; maximum is {maximum}"
350 ),
351 Self::PlanSize { observed, maximum } => {
352 write!(
353 formatter,
354 "commit plan is {observed} bytes; maximum is {maximum}"
355 )
356 }
357 Self::TransactionWorkspaceExists { transaction } => write!(
358 formatter,
359 "transaction workspace already exists for {transaction}; recovery is required"
360 ),
361 Self::UnsafeBlobStore {
362 workspace_root,
363 blobs_directory,
364 } => write!(
365 formatter,
366 "blob store {} must be disjoint from workspace {}",
367 blobs_directory.display(),
368 workspace_root.display()
369 ),
370 Self::Host(source) => fmt::Display::fmt(source, formatter),
371 Self::Store(source) => fmt::Display::fmt(source, formatter),
372 Self::Blob(source) => fmt::Display::fmt(source, formatter),
373 Self::DataDirectory(source) => fmt::Display::fmt(source, formatter),
374 Self::Journal(source) => fmt::Display::fmt(source, formatter),
375 Self::PlanDecode(source) => fmt::Display::fmt(source, formatter),
376 Self::InternalIo { operation, source } => write!(formatter, "{operation}: {source}"),
377 Self::Stale { conflicts } => write!(
378 formatter,
379 "commit dependencies are stale ({} conflict(s))",
380 conflicts.len()
381 ),
382 Self::Verification(failure) => {
383 write!(
384 formatter,
385 "post-commit verification failed at {}",
386 failure.path
387 )
388 }
389 Self::FaultInjected { point } => write!(formatter, "injected fault at {point:?}"),
390 Self::RecoveryRequired { transaction, cause } => {
391 write!(
392 formatter,
393 "transaction {transaction} requires recovery: {cause}"
394 )
395 }
396 Self::RecoveryConflict(conflict) => write!(
397 formatter,
398 "transaction {} recovery conflict: {}",
399 conflict.transaction, conflict.reason
400 ),
401 Self::InvalidRecoveryState { transaction, state } => write!(
402 formatter,
403 "transaction {transaction} has invalid recovery state {state:?}"
404 ),
405 }
406 }
407}
408
409impl Error for CommitError {
410 fn source(&self) -> Option<&(dyn Error + 'static)> {
411 match self {
412 Self::Plan(source) => Some(source),
413 Self::Host(source) => Some(source),
414 Self::Store(source) => Some(source),
415 Self::Blob(source) => Some(source),
416 Self::DataDirectory(source) => Some(source),
417 Self::Journal(source) => Some(source),
418 Self::PlanDecode(source) => Some(source),
419 Self::InternalIo { source, .. } => Some(source),
420 Self::Binding { .. }
421 | Self::BaseSnapshotBinding
422 | Self::DependencyLimit { .. }
423 | Self::PlanSize { .. }
424 | Self::TransactionWorkspaceExists { .. }
425 | Self::UnsafeBlobStore { .. }
426 | Self::Stale { .. }
427 | Self::Verification(_)
428 | Self::FaultInjected { .. }
429 | Self::RecoveryRequired { .. }
430 | Self::RecoveryConflict(_)
431 | Self::InvalidRecoveryState { .. } => None,
432 }
433 }
434}
435
436impl From<CommitPlanError> for CommitError {
437 fn from(source: CommitPlanError) -> Self {
438 Self::Plan(source)
439 }
440}
441
442impl From<HostError> for CommitError {
443 fn from(source: HostError) -> Self {
444 Self::Host(source)
445 }
446}
447
448impl From<TransactionStoreError> for CommitError {
449 fn from(source: TransactionStoreError) -> Self {
450 Self::Store(source)
451 }
452}
453
454impl From<BlobStoreError> for CommitError {
455 fn from(source: BlobStoreError) -> Self {
456 Self::Blob(source)
457 }
458}
459
460impl From<DataDirectoryError> for CommitError {
461 fn from(source: DataDirectoryError) -> Self {
462 Self::DataDirectory(source)
463 }
464}
465
466impl From<JournalError> for CommitError {
467 fn from(source: JournalError) -> Self {
468 Self::Journal(source)
469 }
470}
471
472impl From<PlanDecodeError> for CommitError {
473 fn from(source: PlanDecodeError) -> Self {
474 Self::PlanDecode(source)
475 }
476}
477
478pub struct Committer {
480 workspace_root: Arc<PathBuf>,
481 root: Arc<Dir>,
482 root_file_id: PlatformFileId,
483 runtime: Arc<Dir>,
484 runtime_file_id: PlatformFileId,
485 transactions: Arc<Dir>,
486 coordination: Arc<File>,
487 blobs: BlobStore,
488 config: CommitConfig,
489}
490
491impl Committer {
492 pub fn open(
499 workspace_root: impl AsRef<Path>,
500 blobs: BlobStore,
501 config: CommitConfig,
502 ) -> Result<Self, CommitError> {
503 let workspace_root = fs::canonicalize(workspace_root.as_ref()).map_err(|source| {
504 CommitError::InternalIo {
505 operation: "canonicalize workspace capability",
506 source,
507 }
508 })?;
509 if paths_overlap(&workspace_root, blobs.blobs_dir()) {
510 return Err(CommitError::UnsafeBlobStore {
511 workspace_root,
512 blobs_directory: blobs.blobs_dir().to_path_buf(),
513 });
514 }
515 let (root, root_file_id, runtime, runtime_file_id, transactions, coordination) =
516 Self::open_workspace_components(&workspace_root)?;
517 let committer = Self {
518 workspace_root: Arc::new(workspace_root),
519 root: Arc::new(root),
520 root_file_id,
521 runtime: Arc::new(runtime),
522 runtime_file_id,
523 transactions: Arc::new(transactions),
524 coordination: Arc::new(coordination),
525 blobs,
526 config,
527 };
528 committer.validate_runtime_directory()?;
529 Ok(committer)
530 }
531
532 pub fn open_with_workspace_data(
540 workspace_root: impl AsRef<Path>,
541 config: CommitConfig,
542 ) -> Result<(Self, DataDirectory), CommitError> {
543 let workspace_root = fs::canonicalize(workspace_root.as_ref()).map_err(|source| {
544 CommitError::InternalIo {
545 operation: "canonicalize workspace capability",
546 source,
547 }
548 })?;
549 let (root, root_file_id, runtime, runtime_file_id, transactions, coordination) =
550 Self::open_workspace_components(&workspace_root)?;
551 let data_directory = DataDirectory::open_runtime_data(&runtime, &workspace_root)?;
552 let blobs = BlobStore::open_in(&data_directory)?;
553 sync_dir(&runtime).map_err(|source| CommitError::InternalIo {
554 operation: "sync VSH runtime data directory",
555 source,
556 })?;
557 sync_dir(&root).map_err(|source| CommitError::InternalIo {
558 operation: "sync workspace data parent",
559 source,
560 })?;
561 let committer = Self {
562 workspace_root: Arc::new(workspace_root),
563 root: Arc::new(root),
564 root_file_id,
565 runtime: Arc::new(runtime),
566 runtime_file_id,
567 transactions: Arc::new(transactions),
568 coordination: Arc::new(coordination),
569 blobs,
570 config,
571 };
572 committer.validate_runtime_directory()?;
573 Ok((committer, data_directory))
574 }
575
576 #[must_use]
578 pub fn artifact_store(&self) -> BlobStore {
579 self.blobs.clone()
580 }
581
582 fn open_workspace_components(
583 workspace_root: &Path,
584 ) -> Result<(Dir, PlatformFileId, Dir, PlatformFileId, Dir, File), CommitError> {
585 let root =
586 Dir::open_ambient_dir(workspace_root, ambient_authority()).map_err(|source| {
587 CommitError::InternalIo {
588 operation: "open workspace capability",
589 source,
590 }
591 })?;
592 let root_stamp = stamp_dir(&root, &VPath::root())?;
593 let runtime =
594 open_or_create_real_dir(&root, crate::host::RUNTIME_DIRECTORY).map_err(|source| {
595 CommitError::InternalIo {
596 operation: "open VSH runtime directory",
597 source,
598 }
599 })?;
600 let runtime_path = VPath::parse(crate::host::RUNTIME_DIRECTORY)
601 .expect("built-in runtime directory is a valid VPath");
602 let opened = stamp_dir(&runtime, &runtime_path)?;
603 let named = stamp_at(&root, &runtime_path)?.ok_or_else(|| {
604 HostError::io(
605 "verify VSH runtime directory",
606 &runtime_path,
607 io::Error::new(io::ErrorKind::NotFound, "runtime directory disappeared"),
608 )
609 })?;
610 if opened.kind != NodeKind::Directory
611 || named.kind != NodeKind::Directory
612 || opened.file_id != named.file_id
613 {
614 return Err(HostError::Unstable {
615 path: runtime_path,
616 before: Box::new(named),
617 after: Box::new(opened),
618 }
619 .into());
620 }
621 let transactions = open_or_create_real_dir(&runtime, crate::host::TRANSACTIONS_DIRECTORY)
622 .map_err(|source| CommitError::InternalIo {
623 operation: "open VSH transaction directory",
624 source,
625 })?;
626 let coordination = open_coordination_file(&runtime, COMMIT_LOCK_FILE)?;
627 sync_dir(&runtime).map_err(|source| CommitError::InternalIo {
628 operation: "sync VSH runtime directory",
629 source,
630 })?;
631 sync_dir(&root).map_err(|source| CommitError::InternalIo {
632 operation: "sync workspace root",
633 source,
634 })?;
635 Ok((
636 root,
637 root_stamp.file_id,
638 runtime,
639 opened.file_id,
640 transactions,
641 coordination,
642 ))
643 }
644
645 fn validate_workspace_directory(&self) -> Result<(), CommitError> {
646 let metadata = fs::symlink_metadata(self.workspace_root.as_path()).map_err(|source| {
647 CommitError::InternalIo {
648 operation: "inspect named workspace directory",
649 source,
650 }
651 })?;
652 if !metadata.is_dir() || metadata.file_type().is_symlink() {
653 return Err(CommitError::InternalIo {
654 operation: "verify pinned workspace directory",
655 source: io::Error::new(
656 io::ErrorKind::InvalidData,
657 "workspace path is no longer a real directory",
658 ),
659 });
660 }
661 let named = Dir::open_ambient_dir(self.workspace_root.as_path(), ambient_authority())
662 .map_err(|source| CommitError::InternalIo {
663 operation: "reopen named workspace directory",
664 source,
665 })?;
666 let stamp = stamp_dir(&named, &VPath::root())?;
667 let final_metadata =
668 fs::symlink_metadata(self.workspace_root.as_path()).map_err(|source| {
669 CommitError::InternalIo {
670 operation: "reinspect named workspace directory",
671 source,
672 }
673 })?;
674 if stamp.kind == NodeKind::Directory
675 && stamp.file_id == self.root_file_id
676 && final_metadata.is_dir()
677 && !final_metadata.file_type().is_symlink()
678 {
679 Ok(())
680 } else {
681 Err(CommitError::InternalIo {
682 operation: "verify pinned workspace directory",
683 source: io::Error::new(
684 io::ErrorKind::InvalidData,
685 "workspace directory identity changed",
686 ),
687 })
688 }
689 }
690
691 fn validate_runtime_directory(&self) -> Result<(), CommitError> {
692 self.validate_workspace_directory()?;
693 let runtime_path = VPath::parse(crate::host::RUNTIME_DIRECTORY)
694 .expect("built-in runtime directory is a valid VPath");
695 let opened = stamp_dir(&self.runtime, &runtime_path)?;
696 let named = stamp_at(&self.root, &runtime_path)?;
697 if opened.kind == NodeKind::Directory
698 && opened.file_id == self.runtime_file_id
699 && named.is_some_and(|named| {
700 named.kind == NodeKind::Directory && named.file_id == self.runtime_file_id
701 })
702 {
703 Ok(())
704 } else {
705 Err(CommitError::InternalIo {
706 operation: "verify pinned VSH runtime directory",
707 source: io::Error::new(
708 io::ErrorKind::InvalidData,
709 "protected runtime directory identity changed",
710 ),
711 })
712 }
713 }
714
715 #[must_use]
716 pub const fn config(&self) -> CommitConfig {
718 self.config
719 }
720
721 pub fn snapshot(&self, limits: SnapshotLimits) -> Result<BaseSnapshot, CommitError> {
727 let _guard = WorkspaceLockGuard::shared(&self.coordination).map_err(|source| {
728 CommitError::InternalIo {
729 operation: "acquire shared workspace lock",
730 source,
731 }
732 })?;
733 self.validate_runtime_directory()?;
734 let snapshot = capture_snapshot(&self.root, self.blobs.clone(), limits).map_err(Into::into);
735 self.validate_runtime_directory()?;
736 snapshot
737 }
738
739 pub fn revalidate(
745 &self,
746 plan: &CommitPlan<'_>,
747 ) -> Result<Vec<RevalidationConflict>, CommitError> {
748 self.validate_runtime_directory()?;
749 let dependency_count = plan.read_set().len().saturating_add(plan.write_set().len());
750 if dependency_count > self.config.max_dependencies {
751 return Err(CommitError::DependencyLimit {
752 observed: dependency_count,
753 maximum: self.config.max_dependencies,
754 });
755 }
756 let mut conflicts = Vec::new();
757 for (path, observation) in plan.read_set() {
758 self.revalidate_read(path, observation, &mut conflicts)?;
759 if conflicts.len() >= self.config.max_conflicts {
760 self.validate_runtime_directory()?;
761 return Ok(conflicts);
762 }
763 }
764 for (path, precondition) in plan.write_set() {
765 let (matches, actual) = state_matches(&self.root, path, precondition.expected)?;
766 if !matches {
767 conflicts.push(RevalidationConflict::Metadata {
768 path: path.clone(),
769 expected: precondition.expected,
770 actual,
771 });
772 if conflicts.len() >= self.config.max_conflicts {
773 break;
774 }
775 }
776 }
777 self.validate_runtime_directory()?;
778 Ok(conflicts)
779 }
780
781 pub fn commit<S: TransactionStore + ?Sized>(
788 &self,
789 store: &S,
790 reservation: CommitReservation,
791 plan: &CommitPlan<'_>,
792 ) -> Result<CommitReceipt, CommitError> {
793 self.commit_with_faults(store, reservation, plan, &NoFaults)
794 }
795
796 #[allow(clippy::needless_pass_by_value)]
805 pub fn commit_with_faults<S, F>(
806 &self,
807 store: &S,
808 reservation: CommitReservation,
809 plan: &CommitPlan<'_>,
810 faults: &F,
811 ) -> Result<CommitReceipt, CommitError>
812 where
813 S: TransactionStore + ?Sized,
814 F: FaultInjector + ?Sized,
815 {
816 let (transaction, prepared, encoded) =
817 self.prepare_reserved_plan(store, &reservation, plan)?;
818 let _guard = WorkspaceLockGuard::exclusive(&self.coordination).map_err(|source| {
819 Self::fail_reserved(
820 store,
821 transaction,
822 CommitError::InternalIo {
823 operation: "acquire exclusive workspace lock",
824 source,
825 },
826 )
827 })?;
828 self.validate_runtime_directory()
829 .map_err(|error| Self::fail_reserved(store, transaction, error))?;
830 store.compare_and_transition(
831 transaction,
832 TransactionState::Reserved,
833 TransactionState::Revalidating,
834 )?;
835
836 let transaction_name = transaction.to_string();
837 let transaction_dir = match self.create_transaction_workspace(&transaction_name) {
838 Ok(directory) => directory,
839 Err(error) => {
840 let _ = store.compare_and_transition(
841 transaction,
842 TransactionState::Revalidating,
843 TransactionState::Failed,
844 );
845 return Err(error);
846 }
847 };
848 let result = self.prepare_and_commit(
849 store,
850 transaction,
851 &transaction_dir,
852 plan,
853 &prepared,
854 &encoded,
855 faults,
856 );
857 drop(transaction_dir);
860 self.resolve_commit_result(store, transaction, &transaction_name, &prepared, result)
861 }
862
863 fn prepare_reserved_plan<S: TransactionStore + ?Sized>(
864 &self,
865 store: &S,
866 reservation: &CommitReservation,
867 plan: &CommitPlan<'_>,
868 ) -> Result<(TransactionId, PreparedPlan, Vec<u8>), CommitError> {
869 let transaction = plan.transaction();
870 let reserved_transaction = reservation.transaction();
871 if reserved_transaction != transaction {
872 return Err(Self::fail_reserved(
873 store,
874 reserved_transaction,
875 CommitError::Binding {
876 reserved_transaction,
877 plan_transaction: transaction,
878 },
879 ));
880 }
881 if reservation.base_snapshot() != plan.base_snapshot() {
882 return Err(Self::fail_reserved(
883 store,
884 reserved_transaction,
885 CommitError::BaseSnapshotBinding,
886 ));
887 }
888 let prepared =
889 PreparedPlan::prepare(plan, self.config.max_operations, self.config.max_path_bytes)
890 .map_err(|source| {
891 Self::fail_reserved(store, reserved_transaction, CommitError::from(source))
892 })?;
893 let encoded = prepared.encode().map_err(|source| {
894 Self::fail_reserved(store, reserved_transaction, CommitError::from(source))
895 })?;
896 if encoded.len() > self.config.max_plan_bytes {
897 return Err(Self::fail_reserved(
898 store,
899 reserved_transaction,
900 CommitError::PlanSize {
901 observed: encoded.len(),
902 maximum: self.config.max_plan_bytes,
903 },
904 ));
905 }
906 Ok((transaction, prepared, encoded))
907 }
908
909 fn resolve_commit_result<S: TransactionStore + ?Sized>(
910 &self,
911 store: &S,
912 transaction: TransactionId,
913 transaction_name: &str,
914 prepared: &PreparedPlan,
915 result: Result<CommitReceipt, CommitError>,
916 ) -> Result<CommitReceipt, CommitError> {
917 match result {
918 Ok(mut receipt) => {
919 receipt.cleanup_pending = self.cleanup_transaction(transaction_name).is_err();
920 Ok(receipt)
921 }
922 Err(error) => {
923 let current = store.get(transaction).ok().map(|record| record.state());
924 match current {
925 Some(TransactionState::Revalidating) => {
926 let _ = store.compare_and_transition(
927 transaction,
928 TransactionState::Revalidating,
929 TransactionState::Failed,
930 );
931 let _ = self.cleanup_transaction(transaction_name);
932 Err(error)
933 }
934 Some(TransactionState::Committing) => {
935 let _ = store.compare_and_transition(
936 transaction,
937 TransactionState::Committing,
938 TransactionState::RecoveryRequired,
939 );
940 Err(CommitError::RecoveryRequired {
941 transaction,
942 cause: error.to_string(),
943 })
944 }
945 Some(TransactionState::RecoveryRequired) => {
946 Err(CommitError::RecoveryRequired {
947 transaction,
948 cause: error.to_string(),
949 })
950 }
951 Some(TransactionState::Committed) => {
952 let cleanup_pending = self.cleanup_transaction(transaction_name).is_err();
953 Ok(CommitReceipt {
954 transaction,
955 operations: prepared.operations.len(),
956 verified_paths: prepared.final_states.len(),
957 cleanup_pending,
958 })
959 }
960 Some(TransactionState::Stale) => {
961 let _ = self.cleanup_transaction(transaction_name);
962 Err(error)
963 }
964 _ => Err(error),
965 }
966 }
967 }
968 }
969
970 fn fail_reserved<S: TransactionStore + ?Sized>(
971 store: &S,
972 transaction: TransactionId,
973 error: CommitError,
974 ) -> CommitError {
975 match store.compare_and_transition(
976 transaction,
977 TransactionState::Reserved,
978 TransactionState::Failed,
979 ) {
980 Ok(_) => error,
981 Err(source) => CommitError::Store(source),
982 }
983 }
984
985 #[allow(clippy::too_many_arguments, clippy::too_many_lines)]
986 fn prepare_and_commit<S, F>(
987 &self,
988 store: &S,
989 transaction: TransactionId,
990 transaction_dir: &Dir,
991 plan: &CommitPlan<'_>,
992 prepared: &PreparedPlan,
993 encoded: &[u8],
994 faults: &F,
995 ) -> Result<CommitReceipt, CommitError>
996 where
997 S: TransactionStore + ?Sized,
998 F: FaultInjector + ?Sized,
999 {
1000 Self::write_plan(transaction_dir, encoded)?;
1001 let mut journal =
1002 Journal::create(transaction_dir).map_err(|source| CommitError::InternalIo {
1003 operation: "create commit journal",
1004 source,
1005 })?;
1006 Self::check_fault(faults, FaultPoint::PlanSynced)?;
1007
1008 let stage =
1009 open_or_create_real_dir(transaction_dir, STAGE_DIRECTORY).map_err(|source| {
1010 CommitError::InternalIo {
1011 operation: "create commit staging directory",
1012 source,
1013 }
1014 })?;
1015 let quarantine =
1016 open_or_create_real_dir(transaction_dir, QUARANTINE_DIRECTORY).map_err(|source| {
1017 CommitError::InternalIo {
1018 operation: "create commit quarantine directory",
1019 source,
1020 }
1021 })?;
1022 self.stage_content(prepared, &stage)?;
1023 sync_dir(&stage).map_err(|source| CommitError::InternalIo {
1024 operation: "sync commit staging directory",
1025 source,
1026 })?;
1027 sync_dir(&quarantine).map_err(|source| CommitError::InternalIo {
1028 operation: "sync commit quarantine directory",
1029 source,
1030 })?;
1031 sync_dir(transaction_dir).map_err(|source| CommitError::InternalIo {
1032 operation: "sync transaction directory",
1033 source,
1034 })?;
1035 Self::check_fault(faults, FaultPoint::StageSynced)?;
1036
1037 let conflicts = self.revalidate(plan)?;
1038 if !conflicts.is_empty() {
1039 store.compare_and_transition(
1040 transaction,
1041 TransactionState::Revalidating,
1042 TransactionState::Stale,
1043 )?;
1044 return Err(CommitError::Stale { conflicts });
1045 }
1046 let mut pinned_parents = match self.pin_parent_directories(plan, prepared) {
1047 Ok(parents) => parents,
1048 Err(CommitError::Stale { conflicts }) => {
1049 store.compare_and_transition(
1050 transaction,
1051 TransactionState::Revalidating,
1052 TransactionState::Stale,
1053 )?;
1054 return Err(CommitError::Stale { conflicts });
1055 }
1056 Err(error) => return Err(error),
1057 };
1058 Self::check_fault(faults, FaultPoint::Revalidated)?;
1059 store.compare_and_transition(
1060 transaction,
1061 TransactionState::Revalidating,
1062 TransactionState::Committing,
1063 )?;
1064 Self::check_fault(faults, FaultPoint::CommitStatePersisted)?;
1065
1066 let mut completed_witnesses = Vec::with_capacity(prepared.operations.len());
1067 for (index, operation) in prepared.operations.iter().enumerate() {
1068 let index =
1069 u32::try_from(index).map_err(|_| CommitPlanError::OperationCountOverflow)?;
1070 let parent_path = operation
1071 .path()
1072 .parent()
1073 .expect("commit operations cannot target the workspace root");
1074 let parent =
1075 pinned_parents
1076 .get(&parent_path)
1077 .ok_or_else(|| CommitError::InternalIo {
1078 operation: "locate pinned commit parent",
1079 source: io::Error::new(
1080 io::ErrorKind::InvalidData,
1081 format!("commit parent {parent_path} was not pinned"),
1082 ),
1083 })?;
1084 let source_witness = Self::operation_source_witness(operation, &stage)?;
1085 let parent_witness = Witness::from(stamp_dir(parent, &parent_path)?);
1086 journal
1087 .intent(index, source_witness, parent_witness)
1088 .map_err(|source| CommitError::InternalIo {
1089 operation: "sync commit intent",
1090 source,
1091 })?;
1092 Self::check_fault(faults, FaultPoint::IntentSynced(index))?;
1093 let applied =
1094 Self::apply_operation(transaction, index, operation, parent, &stage, &quarantine)?;
1095 if let Some((path, directory)) = applied.created_directory {
1096 pinned_parents.insert(path, directory);
1097 }
1098 Self::check_fault(faults, FaultPoint::OperationApplied(index))?;
1099 journal
1100 .done(index, applied.witness)
1101 .map_err(|source| CommitError::InternalIo {
1102 operation: "sync commit completion",
1103 source,
1104 })?;
1105 completed_witnesses.push(applied.witness);
1106 Self::check_fault(faults, FaultPoint::DoneSynced(index))?;
1107 }
1108 for (index, operation) in prepared.operations.iter().enumerate().rev() {
1109 let index =
1110 u32::try_from(index).map_err(|_| CommitPlanError::OperationCountOverflow)?;
1111 Self::clear_operation_marker(
1112 transaction,
1113 index,
1114 operation,
1115 &pinned_parents,
1116 completed_witnesses[usize::try_from(index).expect("u32 fits usize")],
1117 )?;
1118 }
1119 Self::check_fault(faults, FaultPoint::OwnershipMarkersCleared)?;
1120 self.verify_final(prepared)?;
1121 self.validate_runtime_directory()?;
1122 Self::check_fault(faults, FaultPoint::Verified)?;
1123 write_commit_marker(transaction_dir, transaction).map_err(|source| {
1124 CommitError::InternalIo {
1125 operation: "sync commit-complete marker",
1126 source,
1127 }
1128 })?;
1129 Self::check_fault(faults, FaultPoint::CommitMarkerSynced)?;
1130 self.validate_runtime_directory()?;
1131 store.compare_and_transition(
1132 transaction,
1133 TransactionState::Committing,
1134 TransactionState::Committed,
1135 )?;
1136 Self::check_fault(faults, FaultPoint::CommittedStatePersisted)?;
1137 Ok(CommitReceipt {
1138 transaction,
1139 operations: prepared.operations.len(),
1140 verified_paths: prepared.final_states.len(),
1141 cleanup_pending: true,
1143 })
1144 }
1145
1146 fn pin_parent_directories(
1147 &self,
1148 plan: &CommitPlan<'_>,
1149 prepared: &PreparedPlan,
1150 ) -> Result<BTreeMap<VPath, Dir>, CommitError> {
1151 let parents = prepared
1152 .operations
1153 .iter()
1154 .map(|operation| {
1155 operation
1156 .path()
1157 .parent()
1158 .expect("commit operations cannot target the workspace root")
1159 })
1160 .collect::<BTreeSet<_>>();
1161 let mut pinned = BTreeMap::new();
1162 let mut conflicts = Vec::new();
1163 for parent in parents {
1164 let expected = plan
1165 .read_set()
1166 .get(&parent)
1167 .and_then(|observation| observation.metadata)
1168 .expect("validated commit plans have parent metadata dependencies");
1169 let Some(expected) = expected else {
1170 continue;
1171 };
1172 let directory = if parent.is_root() {
1173 self.root.try_clone().map_err(|source| {
1174 HostError::io("pin workspace root for commit", &parent, source)
1175 })?
1176 } else if let Ok(directory) = self.root.open_dir(relative_path(&parent)) {
1177 directory
1178 } else {
1179 conflicts.push(RevalidationConflict::Metadata {
1180 path: parent.clone(),
1181 expected: Some(expected),
1182 actual: stamp_at(&self.root, &parent)?.map(NodeState::from_stamp),
1183 });
1184 continue;
1185 };
1186 let actual = NodeState::from_stamp(stamp_dir(&directory, &parent)?);
1187 if actual != expected {
1188 conflicts.push(RevalidationConflict::Metadata {
1189 path: parent,
1190 expected: Some(expected),
1191 actual: Some(actual),
1192 });
1193 continue;
1194 }
1195 pinned.insert(parent, directory);
1196 }
1197 if conflicts.is_empty() {
1198 Ok(pinned)
1199 } else {
1200 Err(CommitError::Stale { conflicts })
1201 }
1202 }
1203
1204 fn create_transaction_workspace(&self, name: &str) -> Result<Dir, CommitError> {
1205 match self.transactions.create_dir(name) {
1206 Ok(()) => {}
1207 Err(source) if source.kind() == io::ErrorKind::AlreadyExists => {
1208 let transaction = parse_transaction_name(name)
1209 .unwrap_or_else(|| TransactionId::from_bytes([0; 32]));
1210 return Err(CommitError::TransactionWorkspaceExists { transaction });
1211 }
1212 Err(source) => {
1213 return Err(CommitError::InternalIo {
1214 operation: "create transaction workspace",
1215 source,
1216 });
1217 }
1218 }
1219 sync_dir(&self.transactions).map_err(|source| CommitError::InternalIo {
1220 operation: "sync transaction workspace parent",
1221 source,
1222 })?;
1223 open_real_dir(&self.transactions, name).map_err(|source| CommitError::InternalIo {
1224 operation: "open transaction workspace",
1225 source,
1226 })
1227 }
1228
1229 fn write_plan(transaction_dir: &Dir, bytes: &[u8]) -> Result<(), CommitError> {
1230 let mut file = create_new_file(transaction_dir, PLAN_FILE).map_err(|source| {
1231 CommitError::InternalIo {
1232 operation: "create durable commit plan",
1233 source,
1234 }
1235 })?;
1236 file.write_all(bytes)
1237 .map_err(|source| CommitError::InternalIo {
1238 operation: "write durable commit plan",
1239 source,
1240 })?;
1241 file.sync_all().map_err(|source| CommitError::InternalIo {
1242 operation: "sync durable commit plan",
1243 source,
1244 })?;
1245 sync_dir(transaction_dir).map_err(|source| CommitError::InternalIo {
1246 operation: "sync durable commit-plan directory",
1247 source,
1248 })
1249 }
1250
1251 fn stage_content(&self, plan: &PreparedPlan, stage: &Dir) -> Result<(), CommitError> {
1252 for operation in &plan.operations {
1253 let (after, slot) = match operation {
1254 Operation::InstallFile { after, slot, .. }
1255 | Operation::InstallSymlink { after, slot, .. } => (*after, *slot),
1256 Operation::Quarantine { .. }
1257 | Operation::CreateDirectory { .. }
1258 | Operation::SetDirectoryMode { .. } => continue,
1259 };
1260 let Some(ContentVersion::Blob(blob)) = after.content() else {
1261 return Err(CommitError::Plan(
1262 CommitPlanError::UnmaterializedAfterState {
1263 path: operation.path().clone(),
1264 },
1265 ));
1266 };
1267 let bytes = self.blobs.get(blob)?;
1268 if bytes.len() as u64 != after.size() {
1269 return Err(CommitError::Blob(BlobStoreError::Corrupt {
1270 path: self.blobs.blobs_dir().to_owned(),
1271 expected: blob,
1272 actual: BlobId::digest(&bytes),
1273 }));
1274 }
1275 let name = stage_name(slot);
1276 let mut file =
1277 create_new_file(stage, &name).map_err(|source| CommitError::InternalIo {
1278 operation: "create staged content",
1279 source,
1280 })?;
1281 file.write_all(&bytes)
1282 .map_err(|source| CommitError::InternalIo {
1283 operation: "write staged content",
1284 source,
1285 })?;
1286 if after.kind() == NodeKind::File {
1287 set_file_mode(&file, after.mode()).map_err(|source| CommitError::InternalIo {
1288 operation: "set staged file mode",
1289 source,
1290 })?;
1291 }
1292 file.sync_all().map_err(|source| CommitError::InternalIo {
1293 operation: "sync staged content",
1294 source,
1295 })?;
1296 if after.kind() == NodeKind::Symlink {
1297 let target = validate_symlink_target(operation.path(), &bytes)?;
1298 create_staged_symlink(
1299 stage,
1300 &stage_link_name(slot),
1301 &self.root,
1302 operation.path(),
1303 &target,
1304 )?;
1305 }
1306 }
1307 Ok(())
1308 }
1309
1310 fn operation_source_witness(
1311 operation: &Operation,
1312 stage: &Dir,
1313 ) -> Result<Option<Witness>, CommitError> {
1314 let Operation::InstallSymlink { path, slot, .. } = operation else {
1315 return Ok(None);
1316 };
1317 let staged_path = VPath::parse(&stage_link_name(*slot))
1318 .expect("staged symbolic-link name is a valid VPath");
1319 let stamp = stamp_at(stage, &staged_path)?.ok_or_else(|| {
1320 CommitError::Host(HostError::io(
1321 "inspect staged symbolic link",
1322 path,
1323 io::Error::new(io::ErrorKind::NotFound, "staged symbolic link is missing"),
1324 ))
1325 })?;
1326 Ok(Some(stamp.into()))
1327 }
1328
1329 fn revalidate_read(
1330 &self,
1331 path: &VPath,
1332 observation: &ReadObservation,
1333 conflicts: &mut Vec<RevalidationConflict>,
1334 ) -> Result<(), CommitError> {
1335 if let Some(expected) = observation.metadata {
1336 let (matches, actual) = state_matches(&self.root, path, expected)?;
1337 if !matches {
1338 conflicts.push(RevalidationConflict::Metadata {
1339 path: path.clone(),
1340 expected,
1341 actual,
1342 });
1343 return Ok(());
1344 }
1345 }
1346 if let Some(expected) = observation.content {
1347 let actual = content_digest(&self.root, path)?;
1348 if actual != expected {
1349 conflicts.push(RevalidationConflict::Content {
1350 path: path.clone(),
1351 expected,
1352 actual,
1353 });
1354 return Ok(());
1355 }
1356 }
1357 if let Some(expected) = observation.directory {
1358 let actual = directory_digest(&self.root, path, self.config.max_dependencies)?;
1359 if actual != expected {
1360 conflicts.push(RevalidationConflict::Directory {
1361 path: path.clone(),
1362 expected,
1363 actual,
1364 });
1365 }
1366 }
1367 Ok(())
1368 }
1369
1370 #[allow(clippy::too_many_lines)]
1371 fn apply_operation(
1372 transaction: TransactionId,
1373 index: u32,
1374 operation: &Operation,
1375 parent: &Dir,
1376 stage: &Dir,
1377 quarantine: &Dir,
1378 ) -> Result<AppliedOperation, CommitError> {
1379 let path = operation.path();
1380 let leaf = path
1381 .file_name()
1382 .expect("commit operations cannot target the workspace root");
1383 let leaf_path = VPath::parse(leaf).expect("a VPath leaf is a valid VPath");
1384 match operation {
1385 Operation::Quarantine {
1386 path,
1387 expected,
1388 slot,
1389 } => {
1390 let (matches, _) = state_matches(parent, &leaf_path, Some(*expected))?;
1391 if !matches {
1392 return Err(CommitError::Verification(Box::new(VerificationFailure {
1393 path: path.clone(),
1394 expected: Some(*expected),
1395 actual: stamp_at(parent, &leaf_path)?.map(NodeState::from_stamp),
1396 })));
1397 }
1398 let name = quarantine_name(*slot);
1399 parent
1400 .rename(leaf, quarantine, &name)
1401 .map_err(|source| HostError::io("move old node to quarantine", path, source))?;
1402 let quarantine_path =
1403 VPath::parse(&name).expect("quarantine slot is a valid VPath");
1404 if !relocated_state_matches(quarantine, &quarantine_path, *expected)? {
1405 return Err(CommitError::Verification(Box::new(VerificationFailure {
1406 path: path.clone(),
1407 expected: Some(*expected),
1408 actual: stamp_at(quarantine, &quarantine_path)?.map(NodeState::from_stamp),
1409 })));
1410 }
1411 sync_dir(parent).map_err(|source| {
1412 HostError::io("sync committed parent directory", path, source)
1413 })?;
1414 sync_dir(quarantine).map_err(|source| CommitError::InternalIo {
1415 operation: "sync quarantine directory",
1416 source,
1417 })?;
1418 let stamp = stamp_at(quarantine, &quarantine_path)?
1419 .expect("verified quarantined node remains present");
1420 Ok(AppliedOperation {
1421 witness: stamp.into(),
1422 created_directory: None,
1423 })
1424 }
1425 Operation::CreateDirectory { path, after } => {
1426 parent
1427 .create_dir(leaf)
1428 .map_err(|source| HostError::io("create committed directory", path, source))?;
1429 let directory = parent
1430 .open_dir(leaf)
1431 .map_err(|source| HostError::io("open committed directory", path, source))?;
1432 set_dir_mode(&directory, after.mode()).map_err(|source| {
1433 HostError::io("set committed directory mode", path, source)
1434 })?;
1435 sync_dir(&directory)
1436 .map_err(|source| HostError::io("sync committed directory", path, source))?;
1437 Self::write_directory_owner(&directory, transaction, index).map_err(|source| {
1438 HostError::io("write directory ownership marker", path, source)
1439 })?;
1440 sync_dir(parent).map_err(|source| {
1441 HostError::io("sync committed parent directory", path, source)
1442 })?;
1443 Ok(AppliedOperation {
1444 witness: stamp_dir(&directory, path)?.into(),
1445 created_directory: Some((path.clone(), directory)),
1446 })
1447 }
1448 Operation::InstallFile { path, after, slot } => {
1449 let name = stage_name(*slot);
1450 stage
1451 .hard_link(&name, parent, leaf)
1452 .map_err(|source| HostError::io("install committed file", path, source))?;
1453 let (matches, actual) = state_matches(parent, &leaf_path, Some(*after))?;
1454 if !matches {
1455 return Err(CommitError::Verification(Box::new(VerificationFailure {
1456 path: path.clone(),
1457 expected: Some(*after),
1458 actual,
1459 })));
1460 }
1461 let file = parent
1462 .open(leaf)
1463 .map_err(|source| HostError::io("open committed file", path, source))?;
1464 sync_installed_file(&file)
1465 .map_err(|source| HostError::io("sync committed file", path, source))?;
1466 sync_dir(parent).map_err(|source| {
1467 HostError::io("sync committed parent directory", path, source)
1468 })?;
1469 Ok(AppliedOperation {
1470 witness: stamp_file(&file, path)?.into(),
1471 created_directory: None,
1472 })
1473 }
1474 Operation::InstallSymlink { path, after, slot } => {
1475 stage
1476 .rename(stage_link_name(*slot), parent, leaf)
1477 .map_err(|source| {
1478 HostError::io("install committed symbolic link", path, source)
1479 })?;
1480 let (matches, actual) = state_matches(parent, &leaf_path, Some(*after))?;
1481 if !matches {
1482 return Err(CommitError::Verification(Box::new(VerificationFailure {
1483 path: path.clone(),
1484 expected: Some(*after),
1485 actual,
1486 })));
1487 }
1488 sync_dir(parent).map_err(|source| {
1489 HostError::io("sync committed parent directory", path, source)
1490 })?;
1491 Ok(AppliedOperation {
1492 witness: stamp_at(parent, &leaf_path)?
1493 .expect("verified committed symlink remains present")
1494 .into(),
1495 created_directory: None,
1496 })
1497 }
1498 Operation::SetDirectoryMode {
1499 path,
1500 expected,
1501 after_mode,
1502 } => {
1503 let (matches, actual) = state_matches(parent, &leaf_path, Some(*expected))?;
1504 if !matches {
1505 return Err(CommitError::Verification(Box::new(VerificationFailure {
1506 path: path.clone(),
1507 expected: Some(*expected),
1508 actual,
1509 })));
1510 }
1511 let directory = parent.open_dir(leaf).map_err(|source| {
1512 HostError::io("open directory for metadata commit", path, source)
1513 })?;
1514 set_dir_mode(&directory, *after_mode).map_err(|source| {
1515 HostError::io("set committed directory mode", path, source)
1516 })?;
1517 sync_dir(&directory).map_err(|source| {
1518 HostError::io("sync committed directory mode", path, source)
1519 })?;
1520 sync_dir(parent).map_err(|source| {
1521 HostError::io("sync committed parent directory", path, source)
1522 })?;
1523 Ok(AppliedOperation {
1524 witness: stamp_dir(&directory, path)?.into(),
1525 created_directory: None,
1526 })
1527 }
1528 }
1529 }
1530
1531 fn write_directory_owner(
1532 directory: &Dir,
1533 transaction: TransactionId,
1534 index: u32,
1535 ) -> Result<(), io::Error> {
1536 let mut file = create_new_file(directory, DIRECTORY_OWNER_MARKER)?;
1537 file.write_all(&directory_owner_payload(transaction, index))?;
1538 file.sync_all()?;
1539 sync_dir(directory)
1540 }
1541
1542 fn clear_operation_marker(
1543 transaction: TransactionId,
1544 index: u32,
1545 operation: &Operation,
1546 pinned_parents: &BTreeMap<VPath, Dir>,
1547 witness: Witness,
1548 ) -> Result<(), CommitError> {
1549 let Operation::CreateDirectory { path, .. } = operation else {
1550 return Ok(());
1551 };
1552 let directory = pinned_parents
1553 .get(path)
1554 .ok_or_else(|| CommitError::InternalIo {
1555 operation: "locate pinned created directory",
1556 source: io::Error::new(
1557 io::ErrorKind::InvalidData,
1558 format!("created directory {path} was not pinned"),
1559 ),
1560 })?;
1561 let stamp = stamp_dir(directory, path)?;
1562 if stamp.kind != witness.kind || stamp.file_id != witness.file_id {
1563 return Err(CommitError::InternalIo {
1564 operation: "created directory ownership witness mismatch",
1565 source: io::Error::new(io::ErrorKind::InvalidData, "ownership witness mismatch"),
1566 });
1567 }
1568 if !directory_owner_matches(directory, transaction, index)
1569 .map_err(|source| HostError::io("verify directory ownership marker", path, source))?
1570 {
1571 return Err(CommitError::InternalIo {
1572 operation: "directory ownership marker mismatch",
1573 source: io::Error::new(io::ErrorKind::InvalidData, "ownership marker mismatch"),
1574 });
1575 }
1576 directory
1577 .remove_file(DIRECTORY_OWNER_MARKER)
1578 .map_err(|source| HostError::io("remove directory ownership marker", path, source))?;
1579 sync_dir(directory).map_err(|source| {
1580 HostError::io("sync directory ownership marker removal", path, source)
1581 })?;
1582 Ok(())
1583 }
1584
1585 fn verify_final(&self, plan: &PreparedPlan) -> Result<(), CommitError> {
1586 for (path, expected) in &plan.final_states {
1587 let (matches, actual) = state_matches(&self.root, path, *expected)?;
1588 if !matches {
1589 return Err(CommitError::Verification(Box::new(VerificationFailure {
1590 path: path.clone(),
1591 expected: *expected,
1592 actual,
1593 })));
1594 }
1595 }
1596 Ok(())
1597 }
1598
1599 fn check_fault<F: FaultInjector + ?Sized>(
1600 faults: &F,
1601 point: FaultPoint,
1602 ) -> Result<(), CommitError> {
1603 if faults.should_fail(point) {
1604 Err(CommitError::FaultInjected { point })
1605 } else {
1606 Ok(())
1607 }
1608 }
1609
1610 fn cleanup_transaction(&self, name: &str) -> Result<(), CommitError> {
1611 match self.transactions.remove_dir_all(name) {
1612 Ok(()) => {}
1613 Err(source) if source.kind() == io::ErrorKind::NotFound => return Ok(()),
1614 Err(source) => {
1615 return Err(CommitError::InternalIo {
1616 operation: "remove transaction workspace",
1617 source,
1618 });
1619 }
1620 }
1621 sync_dir(&self.transactions).map_err(|source| CommitError::InternalIo {
1622 operation: "sync transaction cleanup",
1623 source,
1624 })
1625 }
1626
1627 #[allow(clippy::too_many_lines)]
1636 pub fn recover<S: TransactionStore + ?Sized>(
1637 &self,
1638 store: &S,
1639 ) -> Result<RecoveryReport, CommitError> {
1640 let _guard = WorkspaceLockGuard::exclusive(&self.coordination).map_err(|source| {
1641 CommitError::InternalIo {
1642 operation: "acquire recovery workspace lock",
1643 source,
1644 }
1645 })?;
1646 self.validate_runtime_directory()?;
1647 let mut names = Vec::new();
1648 let entries = self
1649 .transactions
1650 .entries()
1651 .map_err(|source| CommitError::InternalIo {
1652 operation: "enumerate recovery journals",
1653 source,
1654 })?;
1655 for entry in entries {
1656 let entry = entry.map_err(|source| CommitError::InternalIo {
1657 operation: "enumerate recovery journal",
1658 source,
1659 })?;
1660 if !entry
1661 .file_type()
1662 .map_err(|source| CommitError::InternalIo {
1663 operation: "inspect recovery journal type",
1664 source,
1665 })?
1666 .is_dir()
1667 {
1668 continue;
1669 }
1670 let name = entry
1671 .file_name()
1672 .into_string()
1673 .map_err(|_| CommitError::InternalIo {
1674 operation: "decode recovery transaction name",
1675 source: io::Error::new(
1676 io::ErrorKind::InvalidData,
1677 "non-UTF-8 transaction name",
1678 ),
1679 })?;
1680 names.push(name);
1681 }
1682 names.sort_unstable();
1683 let mut report = RecoveryReport::default();
1684 for name in names {
1685 let transaction_dir = open_real_dir(&self.transactions, &name).map_err(|source| {
1686 CommitError::InternalIo {
1687 operation: "open recovery transaction",
1688 source,
1689 }
1690 })?;
1691 let plan = self.read_prepared_plan(&transaction_dir)?;
1692 if name != plan.transaction.to_string() {
1693 return Err(CommitError::RecoveryConflict(RecoveryConflict {
1694 transaction: plan.transaction,
1695 path: None,
1696 reason: "transaction directory name does not match durable plan",
1697 }));
1698 }
1699 let marker = has_valid_commit_marker(&transaction_dir, plan.transaction)?;
1700 if marker {
1701 if self.verify_final(&plan).is_err() {
1702 report.conflicts.push(RecoveryConflict {
1703 transaction: plan.transaction,
1704 path: None,
1705 reason: "durable commit marker exists but final state no longer verifies",
1706 });
1707 continue;
1708 }
1709 match store.get(plan.transaction) {
1710 Ok(record) => match record.state() {
1711 TransactionState::Committing => {
1712 store.compare_and_transition(
1713 plan.transaction,
1714 TransactionState::Committing,
1715 TransactionState::Committed,
1716 )?;
1717 }
1718 TransactionState::RecoveryRequired => {
1719 store.compare_and_transition(
1720 plan.transaction,
1721 TransactionState::RecoveryRequired,
1722 TransactionState::Committed,
1723 )?;
1724 }
1725 TransactionState::Committed => {}
1726 state => {
1727 report.conflicts.push(RecoveryConflict {
1728 transaction: plan.transaction,
1729 path: None,
1730 reason: recovery_state_reason(state, true),
1731 });
1732 continue;
1733 }
1734 },
1735 Err(TransactionStoreError::NotFound { .. }) => {
1736 report.orphaned += 1;
1737 }
1738 Err(source) => return Err(source.into()),
1739 }
1740 report.finalized_commits += 1;
1741 drop(transaction_dir);
1742 self.cleanup_transaction(&name)?;
1743 report.cleaned += 1;
1744 continue;
1745 }
1746
1747 let journal = self.read_bounded_journal(&transaction_dir)?;
1748 let stage =
1749 open_or_create_real_dir(&transaction_dir, STAGE_DIRECTORY).map_err(|source| {
1750 CommitError::InternalIo {
1751 operation: "open recovery staging directory",
1752 source,
1753 }
1754 })?;
1755 let quarantine = open_or_create_real_dir(&transaction_dir, QUARANTINE_DIRECTORY)
1756 .map_err(|source| CommitError::InternalIo {
1757 operation: "open recovery quarantine directory",
1758 source,
1759 })?;
1760 let stored_state = match store.get(plan.transaction) {
1761 Ok(record) => Some(record.state()),
1762 Err(TransactionStoreError::NotFound { .. }) => {
1763 report.orphaned += 1;
1764 None
1765 }
1766 Err(source) => return Err(source.into()),
1767 };
1768 if stored_state == Some(TransactionState::Committed) {
1769 report.conflicts.push(RecoveryConflict {
1770 transaction: plan.transaction,
1771 path: None,
1772 reason: "store says committed but durable commit marker is missing",
1773 });
1774 continue;
1775 }
1776 if stored_state == Some(TransactionState::Committing) {
1777 store.compare_and_transition(
1778 plan.transaction,
1779 TransactionState::Committing,
1780 TransactionState::RecoveryRequired,
1781 )?;
1782 }
1783 match self.rollback(&plan, &journal, &stage, &quarantine) {
1784 Ok(()) => {}
1785 Err(conflict) => {
1786 report.conflicts.push(conflict);
1787 continue;
1788 }
1789 }
1790 if let Some(state) = stored_state {
1791 let current = if state == TransactionState::Committing {
1792 TransactionState::RecoveryRequired
1793 } else {
1794 state
1795 };
1796 match current {
1797 TransactionState::RecoveryRequired => {
1798 store.compare_and_transition(
1799 plan.transaction,
1800 TransactionState::RecoveryRequired,
1801 TransactionState::Failed,
1802 )?;
1803 }
1804 TransactionState::Revalidating | TransactionState::Reserved => {
1805 store.compare_and_transition(
1806 plan.transaction,
1807 current,
1808 TransactionState::Failed,
1809 )?;
1810 }
1811 TransactionState::Failed => {}
1812 other => {
1813 report.conflicts.push(RecoveryConflict {
1814 transaction: plan.transaction,
1815 path: None,
1816 reason: recovery_state_reason(other, false),
1817 });
1818 continue;
1819 }
1820 }
1821 }
1822 report.rolled_back += 1;
1823 drop(quarantine);
1824 drop(stage);
1825 drop(transaction_dir);
1826 self.cleanup_transaction(&name)?;
1827 report.cleaned += 1;
1828 }
1829 self.validate_runtime_directory()?;
1830 Ok(report)
1831 }
1832
1833 fn read_prepared_plan(&self, transaction_dir: &Dir) -> Result<PreparedPlan, CommitError> {
1834 let file = open_real_file(transaction_dir, PLAN_FILE).map_err(|source| {
1835 CommitError::InternalIo {
1836 operation: "open durable commit plan",
1837 source,
1838 }
1839 })?;
1840 let mut bytes = Vec::new();
1841 file.take((self.config.max_plan_bytes as u64).saturating_add(1))
1842 .read_to_end(&mut bytes)
1843 .map_err(|source| CommitError::InternalIo {
1844 operation: "read durable commit plan",
1845 source,
1846 })?;
1847 if bytes.len() > self.config.max_plan_bytes {
1848 return Err(CommitError::PlanSize {
1849 observed: bytes.len(),
1850 maximum: self.config.max_plan_bytes,
1851 });
1852 }
1853 PreparedPlan::decode(
1854 &bytes,
1855 self.config.max_operations,
1856 self.config.max_path_bytes,
1857 )
1858 .map_err(Into::into)
1859 }
1860
1861 fn read_bounded_journal(&self, transaction_dir: &Dir) -> Result<JournalState, CommitError> {
1862 let metadata = transaction_dir
1863 .symlink_metadata(JOURNAL_FILE)
1864 .map_err(|source| CommitError::InternalIo {
1865 operation: "inspect commit journal",
1866 source,
1867 })?;
1868 let journal_bytes = usize::try_from(metadata.len()).unwrap_or(usize::MAX);
1869 if journal_bytes > self.config.max_journal_bytes {
1870 return Err(CommitError::PlanSize {
1871 observed: journal_bytes,
1872 maximum: self.config.max_journal_bytes,
1873 });
1874 }
1875 read_journal(transaction_dir, self.config.max_journal_bytes).map_err(Into::into)
1876 }
1877
1878 fn rollback(
1879 &self,
1880 plan: &PreparedPlan,
1881 journal: &JournalState,
1882 stage: &Dir,
1883 quarantine: &Dir,
1884 ) -> Result<(), RecoveryConflict> {
1885 for (index, operation) in plan.operations.iter().enumerate().rev() {
1886 let index = u32::try_from(index).map_err(|_| RecoveryConflict {
1887 transaction: plan.transaction,
1888 path: None,
1889 reason: "operation index overflow during recovery",
1890 })?;
1891 if !journal.has_intent(index) {
1892 continue;
1893 }
1894 let witnesses = OperationWitnesses {
1895 completed: journal.witness(index),
1896 source: journal.intent_witness(index),
1897 parent: journal.parent_witness(index),
1898 };
1899 let parent =
1900 self.open_recovery_parent(plan.transaction, operation.path(), witnesses.parent)?;
1901 let applied = Self::infer_applied(
1902 plan.transaction,
1903 index,
1904 operation,
1905 witnesses,
1906 &parent,
1907 stage,
1908 quarantine,
1909 )?;
1910 if !applied {
1911 continue;
1912 }
1913 Self::undo_operation(
1914 plan.transaction,
1915 index,
1916 operation,
1917 witnesses,
1918 &parent,
1919 stage,
1920 quarantine,
1921 )?;
1922 }
1923 Ok(())
1924 }
1925
1926 fn open_recovery_parent(
1927 &self,
1928 transaction: TransactionId,
1929 path: &VPath,
1930 witness: Option<Witness>,
1931 ) -> Result<Dir, RecoveryConflict> {
1932 let parent_path = path.parent().ok_or_else(|| {
1933 recovery_conflict(
1934 transaction,
1935 Some(path.clone()),
1936 "recovery operation targets the workspace root",
1937 )
1938 })?;
1939 let expected = witness.ok_or_else(|| {
1940 recovery_conflict(
1941 transaction,
1942 Some(parent_path.clone()),
1943 "recovery intent lacks a parent-directory witness",
1944 )
1945 })?;
1946 let parent = if parent_path.is_root() {
1947 self.root.try_clone()
1948 } else {
1949 self.root.open_dir(relative_path(&parent_path))
1950 }
1951 .map_err(|_| {
1952 recovery_conflict(
1953 transaction,
1954 Some(parent_path.clone()),
1955 "cannot open witnessed recovery parent",
1956 )
1957 })?;
1958 let stamp = stamp_dir(&parent, &parent_path).map_err(|_| {
1959 recovery_conflict(
1960 transaction,
1961 Some(parent_path.clone()),
1962 "cannot inspect witnessed recovery parent",
1963 )
1964 })?;
1965 if stamp.kind != expected.kind || stamp.file_id != expected.file_id {
1966 return Err(recovery_conflict(
1967 transaction,
1968 Some(parent_path),
1969 "recovery parent identity changed",
1970 ));
1971 }
1972 Ok(parent)
1973 }
1974
1975 #[allow(clippy::too_many_lines)]
1976 fn infer_applied(
1977 transaction: TransactionId,
1978 index: u32,
1979 operation: &Operation,
1980 witnesses: OperationWitnesses,
1981 parent: &Dir,
1982 stage: &Dir,
1983 quarantine: &Dir,
1984 ) -> Result<bool, RecoveryConflict> {
1985 if witnesses.completed.is_some() {
1986 return Ok(true);
1987 }
1988 let path = operation.path();
1989 let leaf = path
1990 .file_name()
1991 .expect("recovery operations cannot target the workspace root");
1992 let leaf_path = VPath::parse(leaf).expect("a VPath leaf is a valid VPath");
1993 match operation {
1994 Operation::Quarantine {
1995 path,
1996 expected,
1997 slot,
1998 } => {
1999 let qpath = VPath::parse(&quarantine_name(*slot)).expect("valid quarantine slot");
2000 let original = stamp_at(parent, &leaf_path).map_err(|_| {
2001 recovery_conflict(
2002 transaction,
2003 Some(path.clone()),
2004 "cannot inspect incomplete quarantine operation",
2005 )
2006 })?;
2007 let quarantined = stamp_at(quarantine, &qpath).map_err(|_| {
2008 recovery_conflict(
2009 transaction,
2010 Some(path.clone()),
2011 "cannot inspect incomplete quarantine slot",
2012 )
2013 })?;
2014 match (original, quarantined) {
2015 (Some(_), None) => Ok(false),
2016 (None, Some(_)) => relocated_state_matches(quarantine, &qpath, *expected)
2017 .map_err(|_| {
2018 recovery_conflict(
2019 transaction,
2020 Some(path.clone()),
2021 "incomplete quarantine no longer matches the precondition",
2022 )
2023 }),
2024 _ => Err(recovery_conflict(
2025 transaction,
2026 Some(path.clone()),
2027 "incomplete quarantine has ambiguous source and destination state",
2028 )),
2029 }
2030 }
2031 Operation::CreateDirectory { path, .. } => {
2032 if stamp_at(parent, &leaf_path)
2033 .map_err(|_| {
2034 recovery_conflict(
2035 transaction,
2036 Some(path.clone()),
2037 "cannot inspect incomplete directory creation",
2038 )
2039 })?
2040 .is_none()
2041 {
2042 Ok(false)
2043 } else {
2044 let directory = parent.open_dir(leaf).map_err(|_| {
2045 recovery_conflict(
2046 transaction,
2047 Some(path.clone()),
2048 "cannot open incomplete created directory",
2049 )
2050 })?;
2051 directory_owner_matches(&directory, transaction, index)
2052 .map_err(|_| {
2053 recovery_conflict(
2054 transaction,
2055 Some(path.clone()),
2056 "cannot verify incomplete directory ownership marker",
2057 )
2058 })
2059 .and_then(|matches| {
2060 if matches {
2061 Ok(true)
2062 } else {
2063 Err(recovery_conflict(
2064 transaction,
2065 Some(path.clone()),
2066 "directory creation lacks a valid ownership marker",
2067 ))
2068 }
2069 })
2070 }
2071 }
2072 Operation::InstallFile { path, slot, .. } => {
2073 if stamp_at(parent, &leaf_path)
2074 .map_err(|_| {
2075 recovery_conflict(
2076 transaction,
2077 Some(path.clone()),
2078 "cannot inspect incomplete file install",
2079 )
2080 })?
2081 .is_none()
2082 {
2083 return Ok(false);
2084 }
2085 let stage_path = VPath::parse(&stage_name(*slot)).expect("valid stage slot");
2086 if same_file_pair(parent, &leaf_path, stage, &stage_path).map_err(|_| {
2087 recovery_conflict(
2088 transaction,
2089 Some(path.clone()),
2090 "cannot prove incomplete file install ownership",
2091 )
2092 })? {
2093 Ok(true)
2094 } else {
2095 Err(recovery_conflict(
2096 transaction,
2097 Some(path.clone()),
2098 "incomplete file install is not the staged inode",
2099 ))
2100 }
2101 }
2102 Operation::InstallSymlink { path, after, slot } => {
2103 let source_witness = witnesses.source.ok_or_else(|| {
2104 recovery_conflict(
2105 transaction,
2106 Some(path.clone()),
2107 "symlink intent lacks its staged ownership witness",
2108 )
2109 })?;
2110 let staged_path =
2111 VPath::parse(&stage_link_name(*slot)).expect("valid staged symlink slot");
2112 let destination = stamp_at(parent, &leaf_path).map_err(|_| {
2113 recovery_conflict(
2114 transaction,
2115 Some(path.clone()),
2116 "cannot inspect incomplete symlink install",
2117 )
2118 })?;
2119 let staged = stamp_at(stage, &staged_path).map_err(|_| {
2120 recovery_conflict(
2121 transaction,
2122 Some(path.clone()),
2123 "cannot inspect staged symlink ownership",
2124 )
2125 })?;
2126 match (destination, staged) {
2127 (None, Some(stamp)) if Witness::from(stamp) == source_witness => Ok(false),
2128 (Some(stamp), None) if Witness::from(stamp) == source_witness => {
2129 state_matches(parent, &leaf_path, Some(*after))
2130 .map_err(|_| {
2131 recovery_conflict(
2132 transaction,
2133 Some(path.clone()),
2134 "cannot verify incomplete symlink content",
2135 )
2136 })?
2137 .0
2138 .then_some(true)
2139 .ok_or_else(|| {
2140 recovery_conflict(
2141 transaction,
2142 Some(path.clone()),
2143 "incomplete symlink content changed",
2144 )
2145 })
2146 }
2147 (None, None) => Ok(false),
2148 _ => Err(recovery_conflict(
2149 transaction,
2150 Some(path.clone()),
2151 "incomplete symlink ownership is ambiguous",
2152 )),
2153 }
2154 }
2155 Operation::SetDirectoryMode {
2156 path,
2157 expected,
2158 after_mode,
2159 } => {
2160 let Some(stamp) = stamp_at(parent, &leaf_path).map_err(|_| {
2161 recovery_conflict(
2162 transaction,
2163 Some(path.clone()),
2164 "cannot inspect incomplete mode change",
2165 )
2166 })?
2167 else {
2168 return Err(recovery_conflict(
2169 transaction,
2170 Some(path.clone()),
2171 "directory disappeared during incomplete mode change",
2172 ));
2173 };
2174 let expected_id = match expected.content() {
2175 Some(ContentVersion::Stamp(expected_stamp)) => Some(expected_stamp.file_id),
2176 _ => None,
2177 };
2178 if expected_id.is_some_and(|id| id != stamp.file_id) {
2179 return Err(recovery_conflict(
2180 transaction,
2181 Some(path.clone()),
2182 "directory identity changed during incomplete mode change",
2183 ));
2184 }
2185 if stamp.mode == *after_mode {
2186 Ok(true)
2187 } else if stamp.mode == expected.mode() {
2188 Ok(false)
2189 } else {
2190 Err(recovery_conflict(
2191 transaction,
2192 Some(path.clone()),
2193 "directory has neither the old nor new mode",
2194 ))
2195 }
2196 }
2197 }
2198 }
2199
2200 #[allow(clippy::too_many_lines)]
2201 fn undo_operation(
2202 transaction: TransactionId,
2203 index: u32,
2204 operation: &Operation,
2205 witnesses: OperationWitnesses,
2206 parent: &Dir,
2207 stage: &Dir,
2208 quarantine: &Dir,
2209 ) -> Result<(), RecoveryConflict> {
2210 let path = operation.path();
2211 let leaf = path
2212 .file_name()
2213 .expect("recovery operations cannot target the workspace root");
2214 let leaf_path = VPath::parse(leaf).expect("a VPath leaf is a valid VPath");
2215 match operation {
2216 Operation::Quarantine {
2217 path,
2218 expected,
2219 slot,
2220 } => {
2221 let qname = quarantine_name(*slot);
2222 let qpath = VPath::parse(&qname).expect("valid quarantine slot");
2223 if let Some(witness) = witnesses.completed
2224 && !witness_matches(quarantine, &qpath, witness.kind, witness.file_id).map_err(
2225 |_| {
2226 recovery_conflict(
2227 transaction,
2228 Some(path.clone()),
2229 "cannot verify quarantined ownership",
2230 )
2231 },
2232 )?
2233 {
2234 return Err(recovery_conflict(
2235 transaction,
2236 Some(path.clone()),
2237 "quarantined node identity changed",
2238 ));
2239 }
2240 if !relocated_state_matches(quarantine, &qpath, *expected).map_err(|_| {
2241 recovery_conflict(
2242 transaction,
2243 Some(path.clone()),
2244 "cannot verify quarantined node",
2245 )
2246 })? {
2247 return Err(recovery_conflict(
2248 transaction,
2249 Some(path.clone()),
2250 "quarantined node no longer matches its precondition",
2251 ));
2252 }
2253 if stamp_at(parent, &leaf_path)
2254 .map_err(|_| {
2255 recovery_conflict(
2256 transaction,
2257 Some(path.clone()),
2258 "cannot inspect rollback destination",
2259 )
2260 })?
2261 .is_some()
2262 {
2263 return Err(recovery_conflict(
2264 transaction,
2265 Some(path.clone()),
2266 "rollback destination is occupied",
2267 ));
2268 }
2269 quarantine.rename(&qname, parent, leaf).map_err(|_| {
2270 recovery_conflict(
2271 transaction,
2272 Some(path.clone()),
2273 "cannot restore quarantined node",
2274 )
2275 })?;
2276 sync_dir(parent).map_err(|_| {
2277 recovery_conflict(transaction, Some(path.clone()), "cannot sync restored node")
2278 })?;
2279 sync_dir(quarantine).map_err(|_| {
2280 recovery_conflict(
2281 transaction,
2282 Some(path.clone()),
2283 "cannot sync quarantine restoration",
2284 )
2285 })?;
2286 }
2287 Operation::CreateDirectory { path, .. } => {
2288 if let Some(witness) = witnesses.completed {
2289 if !witness_matches(parent, &leaf_path, witness.kind, witness.file_id).map_err(
2290 |_| {
2291 recovery_conflict(
2292 transaction,
2293 Some(path.clone()),
2294 "cannot verify created directory ownership",
2295 )
2296 },
2297 )? {
2298 return Err(recovery_conflict(
2299 transaction,
2300 Some(path.clone()),
2301 "created directory identity changed",
2302 ));
2303 }
2304 } else {
2305 let directory = parent.open_dir(leaf).map_err(|_| {
2306 recovery_conflict(
2307 transaction,
2308 Some(path.clone()),
2309 "cannot open incomplete created directory",
2310 )
2311 })?;
2312 let marker_present = directory_owner_matches(&directory, transaction, index)
2313 .map_err(|_| {
2314 recovery_conflict(
2315 transaction,
2316 Some(path.clone()),
2317 "cannot verify created directory marker",
2318 )
2319 })?;
2320 if !marker_present {
2321 return Err(recovery_conflict(
2322 transaction,
2323 Some(path.clone()),
2324 "created directory lacks a durable ownership proof",
2325 ));
2326 }
2327 }
2328 let directory = parent.open_dir(leaf).map_err(|_| {
2329 recovery_conflict(
2330 transaction,
2331 Some(path.clone()),
2332 "cannot open created directory for rollback",
2333 )
2334 })?;
2335 match directory.remove_file(DIRECTORY_OWNER_MARKER) {
2336 Ok(()) => {}
2337 Err(source) if source.kind() == io::ErrorKind::NotFound => {}
2338 Err(_) => {
2339 return Err(recovery_conflict(
2340 transaction,
2341 Some(path.clone()),
2342 "cannot remove created-directory ownership marker",
2343 ));
2344 }
2345 }
2346 sync_dir(&directory).map_err(|_| {
2347 recovery_conflict(
2348 transaction,
2349 Some(path.clone()),
2350 "cannot sync created-directory marker removal",
2351 )
2352 })?;
2353 drop(directory);
2356 parent.remove_dir(leaf).map_err(|_| {
2357 recovery_conflict(
2358 transaction,
2359 Some(path.clone()),
2360 "created directory is not safely removable",
2361 )
2362 })?;
2363 sync_dir(parent).map_err(|_| {
2364 recovery_conflict(
2365 transaction,
2366 Some(path.clone()),
2367 "cannot sync directory rollback",
2368 )
2369 })?;
2370 }
2371 Operation::InstallFile { path, after, slot } => {
2372 if let Some(witness) = witnesses.completed {
2373 if !witness_matches(parent, &leaf_path, witness.kind, witness.file_id).map_err(
2374 |_| {
2375 recovery_conflict(
2376 transaction,
2377 Some(path.clone()),
2378 "cannot verify installed node ownership",
2379 )
2380 },
2381 )? {
2382 return Err(recovery_conflict(
2383 transaction,
2384 Some(path.clone()),
2385 "installed node identity changed",
2386 ));
2387 }
2388 } else {
2389 let stage_path = VPath::parse(&stage_name(*slot)).expect("valid stage slot");
2390 if !same_file_pair(parent, &leaf_path, stage, &stage_path).map_err(|_| {
2391 recovery_conflict(
2392 transaction,
2393 Some(path.clone()),
2394 "cannot verify incomplete installed-file ownership",
2395 )
2396 })? {
2397 return Err(recovery_conflict(
2398 transaction,
2399 Some(path.clone()),
2400 "incomplete installed file is not the staged inode",
2401 ));
2402 }
2403 }
2404 let (matches, _) =
2405 state_matches(parent, &leaf_path, Some(*after)).map_err(|_| {
2406 recovery_conflict(
2407 transaction,
2408 Some(path.clone()),
2409 "cannot verify installed node content",
2410 )
2411 })?;
2412 if !matches {
2413 return Err(recovery_conflict(
2414 transaction,
2415 Some(path.clone()),
2416 "installed node content changed before rollback",
2417 ));
2418 }
2419 parent.remove_file(leaf).map_err(|_| {
2420 recovery_conflict(
2421 transaction,
2422 Some(path.clone()),
2423 "cannot remove installed node",
2424 )
2425 })?;
2426 sync_dir(parent).map_err(|_| {
2427 recovery_conflict(
2428 transaction,
2429 Some(path.clone()),
2430 "cannot sync installed-node rollback",
2431 )
2432 })?;
2433 }
2434 Operation::InstallSymlink { path, after, .. } => {
2435 let witness = witnesses.completed.or(witnesses.source).ok_or_else(|| {
2436 recovery_conflict(
2437 transaction,
2438 Some(path.clone()),
2439 "installed symlink lacks a durable ownership witness",
2440 )
2441 })?;
2442 if !witness_matches(parent, &leaf_path, witness.kind, witness.file_id).map_err(
2443 |_| {
2444 recovery_conflict(
2445 transaction,
2446 Some(path.clone()),
2447 "cannot verify installed symlink ownership",
2448 )
2449 },
2450 )? {
2451 return Err(recovery_conflict(
2452 transaction,
2453 Some(path.clone()),
2454 "installed symlink identity changed",
2455 ));
2456 }
2457 let (matches, _) =
2458 state_matches(parent, &leaf_path, Some(*after)).map_err(|_| {
2459 recovery_conflict(
2460 transaction,
2461 Some(path.clone()),
2462 "cannot verify installed symlink content",
2463 )
2464 })?;
2465 if !matches {
2466 return Err(recovery_conflict(
2467 transaction,
2468 Some(path.clone()),
2469 "installed symlink changed before rollback",
2470 ));
2471 }
2472 parent.remove_file(leaf).map_err(|_| {
2473 recovery_conflict(
2474 transaction,
2475 Some(path.clone()),
2476 "cannot remove installed symlink",
2477 )
2478 })?;
2479 sync_dir(parent).map_err(|_| {
2480 recovery_conflict(
2481 transaction,
2482 Some(path.clone()),
2483 "cannot sync installed-symlink rollback",
2484 )
2485 })?;
2486 }
2487 Operation::SetDirectoryMode { path, expected, .. } => {
2488 if let Some(witness) = witnesses.completed
2489 && !witness_matches(parent, &leaf_path, witness.kind, witness.file_id).map_err(
2490 |_| {
2491 recovery_conflict(
2492 transaction,
2493 Some(path.clone()),
2494 "cannot verify mode-change ownership",
2495 )
2496 },
2497 )?
2498 {
2499 return Err(recovery_conflict(
2500 transaction,
2501 Some(path.clone()),
2502 "mode-changed directory identity changed",
2503 ));
2504 }
2505 let directory = parent.open_dir(leaf).map_err(|_| {
2506 recovery_conflict(
2507 transaction,
2508 Some(path.clone()),
2509 "cannot open directory for mode rollback",
2510 )
2511 })?;
2512 set_dir_mode(&directory, expected.mode()).map_err(|_| {
2513 recovery_conflict(
2514 transaction,
2515 Some(path.clone()),
2516 "cannot restore directory mode",
2517 )
2518 })?;
2519 sync_dir(&directory).map_err(|_| {
2520 recovery_conflict(
2521 transaction,
2522 Some(path.clone()),
2523 "cannot sync restored directory mode",
2524 )
2525 })?;
2526 }
2527 }
2528 Ok(())
2529 }
2530}
2531
2532fn same_file_pair(
2533 left_root: &Dir,
2534 left: &VPath,
2535 right_root: &Dir,
2536 right: &VPath,
2537) -> Result<bool, HostError> {
2538 let left = stamp_at(left_root, left)?;
2539 let right = stamp_at(right_root, right)?;
2540 Ok(
2541 matches!((left, right), (Some(left), Some(right)) if left.kind == right.kind && left.file_id == right.file_id),
2542 )
2543}
2544
2545fn directory_owner_payload(transaction: TransactionId, index: u32) -> Vec<u8> {
2546 let mut payload = Vec::with_capacity(44);
2547 payload.extend_from_slice(DIRECTORY_OWNER_MAGIC);
2548 payload.extend_from_slice(transaction.as_bytes());
2549 payload.extend_from_slice(&index.to_le_bytes());
2550 payload
2551}
2552
2553fn directory_owner_matches(
2554 directory: &Dir,
2555 transaction: TransactionId,
2556 index: u32,
2557) -> Result<bool, io::Error> {
2558 let file = match directory.open(DIRECTORY_OWNER_MARKER) {
2559 Ok(file) => file,
2560 Err(source) if source.kind() == io::ErrorKind::NotFound => return Ok(false),
2561 Err(source) => return Err(source),
2562 };
2563 let mut bytes = Vec::new();
2564 file.take(45).read_to_end(&mut bytes)?;
2565 Ok(bytes == directory_owner_payload(transaction, index))
2566}
2567
2568fn recovery_conflict(
2569 transaction: TransactionId,
2570 path: Option<VPath>,
2571 reason: &'static str,
2572) -> RecoveryConflict {
2573 RecoveryConflict {
2574 transaction,
2575 path,
2576 reason,
2577 }
2578}
2579
2580#[allow(clippy::match_same_arms)]
2581fn recovery_state_reason(state: TransactionState, marker: bool) -> &'static str {
2582 if marker {
2583 match state {
2584 TransactionState::Created => "commit marker exists while transaction is Created",
2585 TransactionState::Running => "commit marker exists while transaction is Running",
2586 TransactionState::VirtualComplete => {
2587 "commit marker exists while transaction is VirtualComplete"
2588 }
2589 TransactionState::Denied => "commit marker exists while transaction is Denied",
2590 TransactionState::AutoApproved => {
2591 "commit marker exists while transaction is AutoApproved"
2592 }
2593 TransactionState::PendingApproval => {
2594 "commit marker exists while transaction is PendingApproval"
2595 }
2596 TransactionState::Approved => "commit marker exists while transaction is Approved",
2597 TransactionState::Reserved => "commit marker exists while transaction is Reserved",
2598 TransactionState::Revalidating => {
2599 "commit marker exists while transaction is Revalidating"
2600 }
2601 TransactionState::Stale => "commit marker exists while transaction is Stale",
2602 TransactionState::Expired => "commit marker exists while transaction is Expired",
2603 TransactionState::Failed => "commit marker exists while transaction is Failed",
2604 TransactionState::Committing
2605 | TransactionState::Committed
2606 | TransactionState::RecoveryRequired => "valid marker recovery state",
2607 _ => "commit marker exists in an unknown transaction state",
2608 }
2609 } else {
2610 match state {
2611 TransactionState::Created => "recovery journal exists while transaction is Created",
2612 TransactionState::Running => "recovery journal exists while transaction is Running",
2613 TransactionState::VirtualComplete => {
2614 "recovery journal exists while transaction is VirtualComplete"
2615 }
2616 TransactionState::Denied => "recovery journal exists while transaction is Denied",
2617 TransactionState::AutoApproved => {
2618 "recovery journal exists while transaction is AutoApproved"
2619 }
2620 TransactionState::PendingApproval => {
2621 "recovery journal exists while transaction is PendingApproval"
2622 }
2623 TransactionState::Approved => "recovery journal exists while transaction is Approved",
2624 TransactionState::Stale => "recovery journal exists while transaction is Stale",
2625 TransactionState::Expired => "recovery journal exists while transaction is Expired",
2626 TransactionState::Committed => "committed transaction has no durable marker",
2627 TransactionState::Reserved
2628 | TransactionState::Revalidating
2629 | TransactionState::Committing
2630 | TransactionState::RecoveryRequired
2631 | TransactionState::Failed => "valid rollback recovery state",
2632 _ => "recovery journal exists in an unknown transaction state",
2633 }
2634 }
2635}
2636
2637fn parse_transaction_name(name: &str) -> Option<TransactionId> {
2638 if name.len() != 64 {
2639 return None;
2640 }
2641 let mut bytes = [0_u8; 32];
2642 for (index, chunk) in name.as_bytes().chunks_exact(2).enumerate() {
2643 let high = decode_hex(chunk[0])?;
2644 let low = decode_hex(chunk[1])?;
2645 bytes[index] = (high << 4) | low;
2646 }
2647 Some(TransactionId::from_bytes(bytes))
2648}
2649
2650const fn decode_hex(byte: u8) -> Option<u8> {
2651 match byte {
2652 b'0'..=b'9' => Some(byte - b'0'),
2653 b'a'..=b'f' => Some(byte - b'a' + 10),
2654 _ => None,
2655 }
2656}