1use serde::{Deserialize, Serialize};
4use std::path::{Path, PathBuf};
5use thiserror::Error;
6
7use acyclic_native_runtime::durable_rename;
8
9const LEGACY_NATIVE_EXCHANGE_JOURNAL_VERSION: u32 = 1;
10const IDENTITY_NATIVE_EXCHANGE_JOURNAL_VERSION: u32 = 2;
11const NATIVE_EXCHANGE_JOURNAL_VERSION: u32 = 3;
12
13#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize, Deserialize)]
15pub enum NativeExchangePhase {
16 Carrying,
18 Exchanging,
20}
21
22#[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
24pub struct NativeExchangeJournal {
25 #[serde(default = "legacy_journal_version")]
27 pub version: u32,
28 #[serde(default = "legacy_operation")]
30 pub operation: crate::IdempotencyKey,
31 #[serde(default)]
33 pub recovery: Vec<u8>,
34 pub live: PathBuf,
36 pub prepared: PathBuf,
38 pub carried: Vec<PathBuf>,
40 pub roots: [[u8; 16]; 2],
42 pub phase: NativeExchangePhase,
44}
45
46#[derive(Clone, Debug, Eq, PartialEq)]
48pub struct NativeExchangeOutcome {
49 pub published: bool,
51 pub displaced: Option<PathBuf>,
53 pub recovery: Vec<u8>,
55}
56
57pub fn prepare_native_exchange(
65 journal_path: &Path,
66 live: &Path,
67 prepared: &Path,
68 operation: crate::IdempotencyKey,
69 carried: Vec<PathBuf>,
70) -> Result<(), NativeExchangeError> {
71 prepare_native_exchange_with_recovery(
72 journal_path,
73 live,
74 prepared,
75 operation,
76 carried,
77 Vec::new(),
78 )
79}
80
81pub fn prepare_native_exchange_with_recovery(
83 journal_path: &Path,
84 live: &Path,
85 prepared: &Path,
86 operation: crate::IdempotencyKey,
87 carried: Vec<PathBuf>,
88 recovery: Vec<u8>,
89) -> Result<(), NativeExchangeError> {
90 validate_layout(live, prepared)?;
91 let journal = NativeExchangeJournal {
92 version: NATIVE_EXCHANGE_JOURNAL_VERSION,
93 operation,
94 recovery,
95 live: live.to_path_buf(),
96 prepared: prepared.to_path_buf(),
97 carried,
98 roots: [root_identity(live)?, root_identity(prepared)?],
99 phase: NativeExchangePhase::Carrying,
100 };
101 match read_journal(journal_path) {
102 Ok(existing) if same_prepared_request(&existing, &journal) => Ok(()),
103 Ok(_) => Err(NativeExchangeError::IncompatibleJournal),
104 Err(NativeExchangeError::Io(error)) if error.kind() == std::io::ErrorKind::NotFound => {
105 write_journal(journal_path, &journal)
106 }
107 Err(error) => Err(error),
108 }
109}
110
111#[derive(Debug, Error)]
113pub enum NativeExchangeError {
114 #[error("native exchange paths must be distinct sibling directories")]
116 InvalidLayout,
117 #[error("native exchange journal is incompatible")]
119 IncompatibleJournal,
120 #[error("native exchange journal version {0} is unsupported")]
122 UnsupportedJournalVersion(u32),
123 #[error("legacy native exchange outcome is ambiguous; recover it explicitly before retrying")]
125 AmbiguousLegacyJournal,
126 #[error("both copies of carried path exist: {0}")]
128 CarriedPathConflict(PathBuf),
129 #[error(transparent)]
131 Io(#[from] std::io::Error),
132 #[error(transparent)]
134 Json(#[from] serde_json::Error),
135}
136
137pub fn publish_native_exchange(
143 journal_path: &Path,
144 live: &Path,
145 prepared: &Path,
146 operation: crate::IdempotencyKey,
147 carried: Vec<PathBuf>,
148) -> Result<NativeExchangeOutcome, NativeExchangeError> {
149 validate_layout(live, prepared)?;
150 if journal_path.exists() {
151 let journal = read_journal(journal_path)?;
152 if journal.live != live || journal.prepared != prepared {
153 return Err(NativeExchangeError::IncompatibleJournal);
154 }
155 if journal.version >= IDENTITY_NATIVE_EXCHANGE_JOURNAL_VERSION
156 && (journal.operation != operation || journal.carried != carried)
157 {
158 return Err(NativeExchangeError::IncompatibleJournal);
159 }
160 if journal.version == LEGACY_NATIVE_EXCHANGE_JOURNAL_VERSION
161 && journal.phase == NativeExchangePhase::Exchanging
162 {
163 return Err(NativeExchangeError::AmbiguousLegacyJournal);
164 }
165 let same_operation = journal.version >= IDENTITY_NATIVE_EXCHANGE_JOURNAL_VERSION
166 && journal.operation == operation;
167 let recovered = recover_native_exchange(journal_path)?;
168 if recovered.published && same_operation {
169 return Ok(recovered);
170 }
171 }
172 prepare_native_exchange(journal_path, live, prepared, operation, carried)?;
173 let mut journal = read_journal(journal_path)?;
174 if let Err(error) = carry(live, prepared, &journal.carried) {
175 move_back(prepared, live, &journal.carried)?;
176 remove_journal(journal_path)?;
177 return Err(error);
178 }
179 journal.phase = NativeExchangePhase::Exchanging;
180 write_journal(journal_path, &journal)?;
181 if let Err(error) = exchange(live, prepared) {
182 return match recover_native_exchange(journal_path) {
183 Ok(outcome) if outcome.published => Ok(outcome),
184 Ok(_) => Err(error),
185 Err(recovery) => Err(recovery),
186 };
187 }
188 remove_journal(journal_path)?;
189 Ok(NativeExchangeOutcome {
190 published: true,
191 displaced: Some(prepared.to_path_buf()),
192 recovery: journal.recovery,
193 })
194}
195
196fn same_prepared_request(
197 existing: &NativeExchangeJournal,
198 requested: &NativeExchangeJournal,
199) -> bool {
200 existing.version >= IDENTITY_NATIVE_EXCHANGE_JOURNAL_VERSION
201 && existing.operation == requested.operation
202 && existing.recovery == requested.recovery
203 && existing.live == requested.live
204 && existing.prepared == requested.prepared
205 && existing.carried == requested.carried
206 && existing.roots == requested.roots
207 && existing.phase == requested.phase
208}
209
210pub fn exchange_native_entries(left: &Path, right: &Path) -> Result<(), NativeExchangeError> {
217 if left == right || left.parent().is_none() || right.parent().is_none() {
218 return Err(NativeExchangeError::InvalidLayout);
219 }
220 if !entry_exists(left)? || !entry_exists(right)? {
221 return Err(NativeExchangeError::InvalidLayout);
222 }
223 exchange(left, right)?;
224 sync_parent(left)?;
225 if left.parent() != right.parent() {
226 sync_parent(right)?;
227 }
228 Ok(())
229}
230
231pub(crate) fn recover_native_entry_exchange(live: &Path) -> Result<(), NativeExchangeError> {
235 #[cfg(windows)]
236 {
237 let scratch = scratch(live)?;
238 let witness = entry_exchange_witness(live)?;
239 if !entry_exists(&witness)? {
240 if entry_exists(&scratch)? {
241 return Err(NativeExchangeError::IncompatibleJournal);
242 }
243 return Ok(());
244 }
245 validate_entry_exchange_witness(&witness)?;
246 if entry_exists(&scratch)? && entry_exists(live)? {
247 remove_entry(&scratch)?;
248 sync_parent(live)?;
249 } else if entry_exists(&scratch)? {
250 durable_rename(
251 &scratch,
252 live,
253 acyclic_native_runtime::RenameMode::NoReplace,
254 )?;
255 }
256 remove_entry_exchange_witness(&witness)?;
257 }
258 #[cfg(not(windows))]
259 let _ = live;
260 Ok(())
261}
262
263pub fn recover_native_exchange(
265 journal_path: &Path,
266) -> Result<NativeExchangeOutcome, NativeExchangeError> {
267 let journal = read_journal(journal_path)?;
268 validate_layout(&journal.live, &journal.prepared)?;
269 match journal.phase {
270 NativeExchangePhase::Carrying => {
271 move_back(&journal.prepared, &journal.live, &journal.carried)?;
272 remove_journal(journal_path)?;
273 Ok(NativeExchangeOutcome {
274 published: false,
275 displaced: None,
276 recovery: journal.recovery,
277 })
278 }
279 NativeExchangePhase::Exchanging => recover_exchange(journal_path, &journal),
280 }
281}
282
283fn read_journal(path: &Path) -> Result<NativeExchangeJournal, NativeExchangeError> {
284 let journal: NativeExchangeJournal = serde_json::from_slice(&std::fs::read(path)?)?;
285 match journal.version {
286 LEGACY_NATIVE_EXCHANGE_JOURNAL_VERSION
287 | IDENTITY_NATIVE_EXCHANGE_JOURNAL_VERSION
288 | NATIVE_EXCHANGE_JOURNAL_VERSION => Ok(journal),
289 version => Err(NativeExchangeError::UnsupportedJournalVersion(version)),
290 }
291}
292
293const fn legacy_journal_version() -> u32 {
294 LEGACY_NATIVE_EXCHANGE_JOURNAL_VERSION
295}
296
297const fn legacy_operation() -> crate::IdempotencyKey {
298 crate::IdempotencyKey::from_bytes([0; 16])
299}
300
301fn validate_layout(live: &Path, prepared: &Path) -> Result<(), NativeExchangeError> {
302 if live == prepared || live.parent().is_none() || live.parent() != prepared.parent() {
303 return Err(NativeExchangeError::InvalidLayout);
304 }
305 Ok(())
306}
307
308fn carry(from: &Path, into: &Path, relative: &[PathBuf]) -> Result<(), NativeExchangeError> {
309 for path in relative {
310 if path.is_absolute()
311 || path
312 .components()
313 .any(|part| matches!(part, std::path::Component::ParentDir))
314 {
315 return Err(NativeExchangeError::IncompatibleJournal);
316 }
317 let source = from.join(path);
318 if !entry_exists(&source)? {
319 continue;
320 }
321 let destination = into.join(path);
322 if let Some(parent) = destination.parent() {
323 std::fs::create_dir_all(parent)?;
324 }
325 remove_entry(&destination)?;
326 std::fs::rename(source, destination)?;
327 }
328 Ok(())
329}
330
331fn move_back(from: &Path, into: &Path, relative: &[PathBuf]) -> Result<(), NativeExchangeError> {
332 for path in relative {
333 let source = from.join(path);
334 if !entry_exists(&source)? {
335 continue;
336 }
337 let destination = into.join(path);
338 if entry_exists(&destination)? {
339 return Err(NativeExchangeError::CarriedPathConflict(path.clone()));
340 }
341 if let Some(parent) = destination.parent() {
342 std::fs::create_dir_all(parent)?;
343 }
344 std::fs::rename(source, destination)?;
345 }
346 Ok(())
347}
348
349fn write_journal(path: &Path, journal: &NativeExchangeJournal) -> Result<(), NativeExchangeError> {
350 use std::io::Write;
351 let temporary = path.with_extension("next");
352 let mut file = std::fs::File::create(&temporary)?;
353 file.write_all(&serde_json::to_vec(journal)?)?;
354 file.sync_all()?;
355 drop(file);
356 durable_rename(
357 &temporary,
358 path,
359 acyclic_native_runtime::RenameMode::Replace,
360 )?;
361 Ok(())
362}
363
364fn remove_journal(path: &Path) -> Result<(), NativeExchangeError> {
365 match std::fs::remove_file(path) {
366 Ok(()) => sync_parent(path).map_err(Into::into),
367 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
368 Err(error) => Err(error.into()),
369 }
370}
371
372fn entry_exists(path: &Path) -> Result<bool, std::io::Error> {
373 match std::fs::symlink_metadata(path) {
374 Ok(_) => Ok(true),
375 Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(false),
376 Err(error) => Err(error),
377 }
378}
379
380fn root_identity(path: &Path) -> Result<[u8; 16], NativeExchangeError> {
381 let root = cap_std::fs::Dir::open_ambient_dir(path, cap_std::ambient_authority())?;
382 Ok(crate::NativeRootIdentity::from_metadata(&root.dir_metadata()?)?.to_bytes())
383}
384
385fn remove_entry(path: &Path) -> Result<(), std::io::Error> {
386 let metadata = match std::fs::symlink_metadata(path) {
387 Ok(metadata) => metadata,
388 Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(()),
389 Err(error) => return Err(error),
390 };
391 if metadata.is_dir() && !metadata.file_type().is_symlink() {
392 std::fs::remove_dir_all(path)
393 } else {
394 std::fs::remove_file(path)
395 }
396}
397
398fn sync_parent(path: &Path) -> std::io::Result<()> {
399 let parent = path
400 .parent()
401 .ok_or_else(|| std::io::Error::other("path has no parent"))?;
402 acyclic_native_runtime::sync_parent(parent, acyclic_native_runtime::Durability::Full)
403}
404
405#[cfg(windows)]
406fn scratch(live: &Path) -> Result<PathBuf, NativeExchangeError> {
407 let parent = live.parent().ok_or(NativeExchangeError::InvalidLayout)?;
408 let name = live.file_name().ok_or(NativeExchangeError::InvalidLayout)?;
409 Ok(parent.join(format!(".{}.acyclic-exchange", name.to_string_lossy())))
410}
411
412#[cfg(windows)]
413const ENTRY_EXCHANGE_WITNESS: &[u8] = b"acyclic-native-entry-exchange-v1\n";
414
415#[cfg(windows)]
416fn entry_exchange_witness(live: &Path) -> Result<PathBuf, NativeExchangeError> {
417 let parent = live.parent().ok_or(NativeExchangeError::InvalidLayout)?;
418 let name = live.file_name().ok_or(NativeExchangeError::InvalidLayout)?;
419 Ok(parent.join(format!(
420 ".{}.acyclic-exchange-witness",
421 name.to_string_lossy()
422 )))
423}
424
425#[cfg(windows)]
426fn validate_entry_exchange_witness(path: &Path) -> Result<(), NativeExchangeError> {
427 if std::fs::read(path)? == ENTRY_EXCHANGE_WITNESS {
428 Ok(())
429 } else {
430 Err(NativeExchangeError::IncompatibleJournal)
431 }
432}
433
434#[cfg(windows)]
435fn create_entry_exchange_witness(path: &Path) -> Result<(), NativeExchangeError> {
436 use std::io::Write as _;
437
438 let mut file = std::fs::OpenOptions::new()
439 .create_new(true)
440 .write(true)
441 .open(path)?;
442 file.write_all(ENTRY_EXCHANGE_WITNESS)?;
443 file.sync_all()?;
444 sync_parent(path)?;
445 Ok(())
446}
447
448#[cfg(windows)]
449fn remove_entry_exchange_witness(path: &Path) -> Result<(), NativeExchangeError> {
450 remove_entry(path)?;
451 sync_parent(path)?;
452 Ok(())
453}
454
455#[cfg(windows)]
456fn exchange(live: &Path, prepared: &Path) -> Result<(), NativeExchangeError> {
457 recover_native_entry_exchange(live)?;
458 let scratch = scratch(live)?;
459 let witness = entry_exchange_witness(live)?;
460 create_entry_exchange_witness(&witness)?;
461 durable_rename(
462 live,
463 &scratch,
464 acyclic_native_runtime::RenameMode::NoReplace,
465 )?;
466 if let Err(error) = durable_rename(
467 prepared,
468 live,
469 acyclic_native_runtime::RenameMode::NoReplace,
470 ) {
471 let _ = durable_rename(
472 &scratch,
473 live,
474 acyclic_native_runtime::RenameMode::NoReplace,
475 );
476 if entry_exists(live)? && !entry_exists(&scratch)? {
477 remove_entry_exchange_witness(&witness)?;
478 }
479 return Err(error.into());
480 }
481 durable_rename(
482 &scratch,
483 prepared,
484 acyclic_native_runtime::RenameMode::NoReplace,
485 )?;
486 remove_entry_exchange_witness(&witness)?;
487 Ok(())
488}
489
490#[cfg(windows)]
491fn recover_exchange(
492 path: &Path,
493 journal: &NativeExchangeJournal,
494) -> Result<NativeExchangeOutcome, NativeExchangeError> {
495 let scratch = scratch(&journal.live)?;
496 let live = entry_exists(&journal.live)?
497 .then(|| root_identity(&journal.live))
498 .transpose()?;
499 let prepared = entry_exists(&journal.prepared)?
500 .then(|| root_identity(&journal.prepared))
501 .transpose()?;
502 let scratch_identity = entry_exists(&scratch)?
503 .then(|| root_identity(&scratch))
504 .transpose()?;
505 let original = journal.roots[0];
506 let replacement = journal.roots[1];
507 let displaced = if live == Some(original)
508 && prepared == Some(replacement)
509 && scratch_identity.is_none()
510 {
511 move_back(&journal.prepared, &journal.live, &journal.carried)?;
512 None
513 } else if live.is_none() && prepared == Some(replacement) && scratch_identity == Some(original)
514 {
515 move_back(&journal.prepared, &scratch, &journal.carried)?;
516 durable_rename(
517 &scratch,
518 &journal.live,
519 acyclic_native_runtime::RenameMode::NoReplace,
520 )?;
521 None
522 } else if live == Some(replacement) && prepared.is_none() && scratch_identity == Some(original)
523 {
524 durable_rename(
525 &scratch,
526 &journal.prepared,
527 acyclic_native_runtime::RenameMode::NoReplace,
528 )?;
529 Some(journal.prepared.clone())
530 } else if live == Some(replacement) && prepared == Some(original) && scratch_identity.is_none()
531 {
532 Some(journal.prepared.clone())
533 } else if live == Some(replacement) && scratch_identity.is_none() {
534 None
535 } else {
536 return Err(NativeExchangeError::IncompatibleJournal);
537 };
538 remove_journal(path)?;
539 Ok(NativeExchangeOutcome {
540 published: live == Some(replacement),
541 displaced,
542 recovery: journal.recovery.clone(),
543 })
544}
545
546#[cfg(any(target_os = "linux", target_os = "macos"))]
547fn recover_exchange(
548 path: &Path,
549 journal: &NativeExchangeJournal,
550) -> Result<NativeExchangeOutcome, NativeExchangeError> {
551 let live = entry_exists(&journal.live)?
552 .then(|| root_identity(&journal.live))
553 .transpose()?;
554 let prepared = entry_exists(&journal.prepared)?
555 .then(|| root_identity(&journal.prepared))
556 .transpose()?;
557 let (published, displaced) = match (live, prepared) {
558 (Some(live), Some(prepared)) if [live, prepared] == journal.roots => {
559 move_back(&journal.prepared, &journal.live, &journal.carried)?;
560 (false, None)
561 }
562 (Some(live), Some(prepared))
563 if [live, prepared] == [journal.roots[1], journal.roots[0]] =>
564 {
565 (true, Some(journal.prepared.clone()))
566 }
567 (Some(live), _) if live == journal.roots[1] => (true, None),
568 _ => return Err(NativeExchangeError::IncompatibleJournal),
569 };
570 remove_journal(path)?;
571 Ok(NativeExchangeOutcome {
572 published,
573 displaced,
574 recovery: journal.recovery.clone(),
575 })
576}
577
578#[cfg(target_os = "macos")]
579#[allow(unsafe_code, reason = "renamex_np receives two live C paths")]
580fn exchange(live: &Path, prepared: &Path) -> Result<(), NativeExchangeError> {
581 use std::os::unix::ffi::OsStrExt;
582 let live = std::ffi::CString::new(live.as_os_str().as_bytes())
583 .map_err(|_| NativeExchangeError::InvalidLayout)?;
584 let prepared = std::ffi::CString::new(prepared.as_os_str().as_bytes())
585 .map_err(|_| NativeExchangeError::InvalidLayout)?;
586 if unsafe { libc::renamex_np(live.as_ptr(), prepared.as_ptr(), libc::RENAME_SWAP) } == 0 {
587 Ok(())
588 } else {
589 Err(std::io::Error::last_os_error().into())
590 }
591}
592
593#[cfg(target_os = "linux")]
594#[allow(unsafe_code, reason = "renameat2 receives two live C paths")]
595fn exchange(live: &Path, prepared: &Path) -> Result<(), NativeExchangeError> {
596 use std::os::unix::ffi::OsStrExt;
597 let live = std::ffi::CString::new(live.as_os_str().as_bytes())
598 .map_err(|_| NativeExchangeError::InvalidLayout)?;
599 let prepared = std::ffi::CString::new(prepared.as_os_str().as_bytes())
600 .map_err(|_| NativeExchangeError::InvalidLayout)?;
601 let result = unsafe {
602 libc::syscall(
603 libc::SYS_renameat2,
604 libc::AT_FDCWD,
605 live.as_ptr(),
606 libc::AT_FDCWD,
607 prepared.as_ptr(),
608 libc::RENAME_EXCHANGE,
609 )
610 };
611 if result == 0 {
612 Ok(())
613 } else {
614 Err(std::io::Error::last_os_error().into())
615 }
616}
617
618#[cfg(test)]
619#[allow(clippy::expect_used)]
620mod tests {
621 use super::*;
622
623 #[cfg(windows)]
624 #[test]
625 fn entry_exchange_recovery_restores_a_missing_live_entry() {
626 let temporary = tempfile::tempdir().expect("temporary directory");
627 let live = temporary.path().join("live");
628 let scratch = scratch(&live).expect("scratch path");
629 let witness = entry_exchange_witness(&live).expect("witness path");
630 create_entry_exchange_witness(&witness).expect("exchange witness");
631 std::fs::write(&scratch, b"before").expect("scratch entry");
632
633 recover_native_entry_exchange(&live).expect("recover exchange");
634
635 assert_eq!(std::fs::read(&live).expect("restored entry"), b"before");
636 assert!(!scratch.exists());
637 }
638
639 #[cfg(windows)]
640 #[test]
641 fn entry_exchange_recovery_keeps_a_published_live_entry() {
642 let temporary = tempfile::tempdir().expect("temporary directory");
643 let live = temporary.path().join("live");
644 let scratch = scratch(&live).expect("scratch path");
645 let witness = entry_exchange_witness(&live).expect("witness path");
646 create_entry_exchange_witness(&witness).expect("exchange witness");
647 std::fs::write(&live, b"after").expect("published entry");
648 std::fs::write(&scratch, b"before").expect("scratch entry");
649
650 recover_native_entry_exchange(&live).expect("recover exchange");
651
652 assert_eq!(std::fs::read(&live).expect("published entry"), b"after");
653 assert!(!scratch.exists());
654 }
655
656 #[cfg(windows)]
657 #[test]
658 fn entry_exchange_recovery_preserves_an_unowned_scratch_collision() {
659 let temporary = tempfile::tempdir().expect("temporary directory");
660 let live = temporary.path().join("live");
661 let scratch = scratch(&live).expect("scratch path");
662 std::fs::write(&live, b"live").expect("live entry");
663 std::fs::write(&scratch, b"unrelated").expect("scratch collision");
664
665 assert!(matches!(
666 recover_native_entry_exchange(&live),
667 Err(NativeExchangeError::IncompatibleJournal)
668 ));
669 assert_eq!(std::fs::read(&live).expect("live entry"), b"live");
670 assert_eq!(
671 std::fs::read(&scratch).expect("scratch collision"),
672 b"unrelated"
673 );
674 }
675
676 #[test]
677 fn recovery_recognizes_an_exchange_completed_before_journal_removal() {
678 let temporary = tempfile::tempdir().expect("temporary directory");
679 let live = temporary.path().join("live");
680 let prepared = temporary.path().join("prepared");
681 std::fs::create_dir(&live).expect("live tree");
682 std::fs::create_dir(&prepared).expect("prepared tree");
683 std::fs::write(live.join("old"), b"old").expect("old file");
684 std::fs::write(prepared.join("new"), b"new").expect("new file");
685 let journal_path = temporary.path().join("exchange.json");
686 let journal = NativeExchangeJournal {
687 version: NATIVE_EXCHANGE_JOURNAL_VERSION,
688 operation: crate::IdempotencyKey::from_bytes([1; 16]),
689 recovery: vec![7; 16],
690 live: live.clone(),
691 prepared: prepared.clone(),
692 carried: Vec::new(),
693 roots: [
694 root_identity(&live).expect("live identity"),
695 root_identity(&prepared).expect("prepared identity"),
696 ],
697 phase: NativeExchangePhase::Exchanging,
698 };
699 write_journal(&journal_path, &journal).expect("journal");
700 exchange(&live, &prepared).expect("exchange");
701
702 let outcome = publish_native_exchange(
703 &journal_path,
704 &live,
705 &prepared,
706 journal.operation,
707 Vec::new(),
708 )
709 .expect("recover publication retry");
710 assert!(outcome.published);
711 assert_eq!(outcome.displaced.as_deref(), Some(prepared.as_path()));
712 assert_eq!(outcome.recovery, vec![7; 16]);
713 assert_eq!(std::fs::read(live.join("new")).expect("new"), b"new");
714 assert_eq!(std::fs::read(prepared.join("old")).expect("old"), b"old");
715 assert!(!journal_path.exists());
716 }
717
718 #[test]
719 fn prepared_exchange_is_idempotent_and_rejects_other_requests() {
720 let temporary = tempfile::tempdir().expect("temporary directory");
721 let live = temporary.path().join("live");
722 let prepared = temporary.path().join("prepared");
723 std::fs::create_dir(&live).expect("live tree");
724 std::fs::create_dir(&prepared).expect("prepared tree");
725 let journal_path = temporary.path().join("exchange.json");
726 let operation = crate::IdempotencyKey::from_bytes([12; 16]);
727 let carried = vec![PathBuf::from("private")];
728
729 prepare_native_exchange_with_recovery(
730 &journal_path,
731 &live,
732 &prepared,
733 operation,
734 carried.clone(),
735 vec![21; 32],
736 )
737 .expect("prepare exchange");
738 prepare_native_exchange_with_recovery(
739 &journal_path,
740 &live,
741 &prepared,
742 operation,
743 carried.clone(),
744 vec![21; 32],
745 )
746 .expect("retry preparation");
747
748 let journal = read_journal(&journal_path).expect("read journal");
749 assert_eq!(journal.operation, operation);
750 assert_eq!(journal.carried, carried);
751 assert_eq!(journal.recovery, vec![21; 32]);
752 assert_eq!(journal.phase, NativeExchangePhase::Carrying);
753 assert!(matches!(
754 prepare_native_exchange(
755 &journal_path,
756 &live,
757 &prepared,
758 crate::IdempotencyKey::from_bytes([13; 16]),
759 Vec::new(),
760 ),
761 Err(NativeExchangeError::IncompatibleJournal)
762 ));
763 assert!(matches!(
764 publish_native_exchange(
765 &journal_path,
766 &live,
767 &prepared,
768 crate::IdempotencyKey::from_bytes([13; 16]),
769 Vec::new(),
770 ),
771 Err(NativeExchangeError::IncompatibleJournal)
772 ));
773 assert_eq!(
774 read_journal(&journal_path)
775 .expect("preserved journal")
776 .operation,
777 operation
778 );
779 }
780
781 #[test]
782 fn prior_identity_journal_retries_without_reencoding() {
783 let temporary = tempfile::tempdir().expect("temporary directory");
784 let live = temporary.path().join("live");
785 let prepared = temporary.path().join("prepared");
786 std::fs::create_dir(&live).expect("live tree");
787 std::fs::create_dir(&prepared).expect("prepared tree");
788 let journal_path = temporary.path().join("exchange.json");
789 let operation = crate::IdempotencyKey::from_bytes([14; 16]);
790 let journal = NativeExchangeJournal {
791 version: IDENTITY_NATIVE_EXCHANGE_JOURNAL_VERSION,
792 operation,
793 recovery: Vec::new(),
794 live: live.clone(),
795 prepared: prepared.clone(),
796 carried: Vec::new(),
797 roots: [
798 root_identity(&live).expect("live identity"),
799 root_identity(&prepared).expect("prepared identity"),
800 ],
801 phase: NativeExchangePhase::Carrying,
802 };
803 write_journal(&journal_path, &journal).expect("journal");
804
805 prepare_native_exchange(&journal_path, &live, &prepared, operation, Vec::new())
806 .expect("compatible retry");
807 assert_eq!(
808 read_journal(&journal_path).expect("journal").version,
809 IDENTITY_NATIVE_EXCHANGE_JOURNAL_VERSION
810 );
811 }
812
813 #[test]
814 fn publication_rejects_a_journal_for_different_roots() {
815 let temporary = tempfile::tempdir().expect("temporary directory");
816 let first_live = temporary.path().join("first-live");
817 let first_prepared = temporary.path().join("first-prepared");
818 let second_live = temporary.path().join("second-live");
819 let second_prepared = temporary.path().join("second-prepared");
820 for path in [&first_live, &first_prepared, &second_live, &second_prepared] {
821 std::fs::create_dir(path).expect("exchange tree");
822 }
823 std::fs::write(second_live.join("old"), b"old").expect("old file");
824 std::fs::write(second_prepared.join("new"), b"new").expect("new file");
825 let journal_path = temporary.path().join("exchange.json");
826 write_journal(
827 &journal_path,
828 &NativeExchangeJournal {
829 version: NATIVE_EXCHANGE_JOURNAL_VERSION,
830 operation: crate::IdempotencyKey::from_bytes([5; 16]),
831 recovery: Vec::new(),
832 live: first_live.clone(),
833 prepared: first_prepared.clone(),
834 carried: Vec::new(),
835 roots: [
836 root_identity(&first_live).expect("live identity"),
837 root_identity(&first_prepared).expect("prepared identity"),
838 ],
839 phase: NativeExchangePhase::Carrying,
840 },
841 )
842 .expect("journal");
843
844 assert!(matches!(
845 publish_native_exchange(
846 &journal_path,
847 &second_live,
848 &second_prepared,
849 crate::IdempotencyKey::from_bytes([6; 16]),
850 Vec::new(),
851 ),
852 Err(NativeExchangeError::IncompatibleJournal)
853 ));
854 assert_eq!(std::fs::read(second_live.join("old")).expect("old"), b"old");
855 assert_eq!(
856 std::fs::read(second_prepared.join("new")).expect("new"),
857 b"new"
858 );
859 assert!(journal_path.exists());
860 }
861
862 #[test]
863 fn legacy_journal_without_version_or_operation_recovers() {
864 let temporary = tempfile::tempdir().expect("temporary directory");
865 let live = temporary.path().join("live");
866 let prepared = temporary.path().join("prepared");
867 std::fs::create_dir(&live).expect("live tree");
868 std::fs::create_dir(&prepared).expect("prepared tree");
869 let journal_path = temporary.path().join("exchange.json");
870 let mut encoded = serde_json::to_value(NativeExchangeJournal {
871 version: NATIVE_EXCHANGE_JOURNAL_VERSION,
872 operation: crate::IdempotencyKey::from_bytes([9; 16]),
873 recovery: Vec::new(),
874 live: live.clone(),
875 prepared: prepared.clone(),
876 carried: Vec::new(),
877 roots: [
878 root_identity(&live).expect("live identity"),
879 root_identity(&prepared).expect("prepared identity"),
880 ],
881 phase: NativeExchangePhase::Carrying,
882 })
883 .expect("encode journal");
884 let object = encoded.as_object_mut().expect("journal object");
885 object.remove("version");
886 object.remove("operation");
887 std::fs::write(
888 &journal_path,
889 serde_json::to_vec(&encoded).expect("encode legacy journal"),
890 )
891 .expect("write legacy journal");
892
893 let outcome = recover_native_exchange(&journal_path).expect("recover legacy journal");
894 assert!(!outcome.published);
895 assert!(outcome.displaced.is_none());
896 assert!(live.exists());
897 assert!(prepared.exists());
898 assert!(!journal_path.exists());
899 }
900
901 #[test]
902 fn future_journal_version_fails_closed_without_mutation() {
903 let temporary = tempfile::tempdir().expect("temporary directory");
904 let live = temporary.path().join("live");
905 let prepared = temporary.path().join("prepared");
906 std::fs::create_dir(&live).expect("live tree");
907 std::fs::create_dir(&prepared).expect("prepared tree");
908 std::fs::write(live.join("old"), b"old").expect("old file");
909 std::fs::write(prepared.join("new"), b"new").expect("new file");
910 let journal_path = temporary.path().join("exchange.json");
911 let journal = NativeExchangeJournal {
912 version: NATIVE_EXCHANGE_JOURNAL_VERSION + 1,
913 operation: crate::IdempotencyKey::from_bytes([10; 16]),
914 recovery: Vec::new(),
915 live: live.clone(),
916 prepared: prepared.clone(),
917 carried: Vec::new(),
918 roots: [
919 root_identity(&live).expect("live identity"),
920 root_identity(&prepared).expect("prepared identity"),
921 ],
922 phase: NativeExchangePhase::Carrying,
923 };
924 std::fs::write(
925 &journal_path,
926 serde_json::to_vec(&journal).expect("encode future journal"),
927 )
928 .expect("write future journal");
929
930 assert!(matches!(
931 publish_native_exchange(
932 &journal_path,
933 &live,
934 &prepared,
935 crate::IdempotencyKey::from_bytes([11; 16]),
936 Vec::new(),
937 ),
938 Err(NativeExchangeError::UnsupportedJournalVersion(version))
939 if version == NATIVE_EXCHANGE_JOURNAL_VERSION + 1
940 ));
941 assert_eq!(std::fs::read(live.join("old")).expect("old"), b"old");
942 assert_eq!(std::fs::read(prepared.join("new")).expect("new"), b"new");
943 assert!(journal_path.exists());
944 }
945
946 #[cfg(windows)]
947 #[test]
948 fn windows_recovery_rolls_back_after_live_moves_to_scratch() {
949 let temporary = tempfile::tempdir().expect("temporary directory");
950 let live = temporary.path().join("live");
951 let prepared = temporary.path().join("prepared");
952 std::fs::create_dir_all(live.join("private")).expect("live tree");
953 std::fs::create_dir(&prepared).expect("prepared tree");
954 std::fs::write(live.join("old"), b"old").expect("old file");
955 std::fs::write(live.join("private/key"), b"secret").expect("carried file");
956 std::fs::write(prepared.join("new"), b"new").expect("new file");
957 let journal_path = temporary.path().join("exchange.json");
958 let mut journal = NativeExchangeJournal {
959 version: NATIVE_EXCHANGE_JOURNAL_VERSION,
960 operation: crate::IdempotencyKey::from_bytes([7; 16]),
961 recovery: Vec::new(),
962 live: live.clone(),
963 prepared: prepared.clone(),
964 carried: vec![PathBuf::from("private")],
965 roots: [
966 root_identity(&live).expect("live identity"),
967 root_identity(&prepared).expect("prepared identity"),
968 ],
969 phase: NativeExchangePhase::Carrying,
970 };
971 write_journal(&journal_path, &journal).expect("carrying journal");
972 carry(&live, &prepared, &journal.carried).expect("carry exclusion");
973 journal.phase = NativeExchangePhase::Exchanging;
974 write_journal(&journal_path, &journal).expect("exchange journal");
975 let scratch = scratch(&live).expect("scratch path");
976 durable_rename(
977 &live,
978 &scratch,
979 acyclic_native_runtime::RenameMode::NoReplace,
980 )
981 .expect("move live to scratch");
982
983 let outcome = recover_native_exchange(&journal_path).expect("roll back exchange");
984 assert!(!outcome.published);
985 assert!(outcome.displaced.is_none());
986 assert_eq!(std::fs::read(live.join("old")).expect("old"), b"old");
987 assert_eq!(
988 std::fs::read(live.join("private/key")).expect("carried"),
989 b"secret"
990 );
991 assert_eq!(std::fs::read(prepared.join("new")).expect("new"), b"new");
992 assert!(!scratch.exists());
993 assert!(!journal_path.exists());
994 }
995
996 #[cfg(windows)]
997 #[test]
998 fn windows_recovery_finishes_after_replacement_moves_to_live() {
999 let temporary = tempfile::tempdir().expect("temporary directory");
1000 let live = temporary.path().join("live");
1001 let prepared = temporary.path().join("prepared");
1002 std::fs::create_dir_all(live.join("private")).expect("live tree");
1003 std::fs::create_dir(&prepared).expect("prepared tree");
1004 std::fs::write(live.join("old"), b"old").expect("old file");
1005 std::fs::write(live.join("private/key"), b"secret").expect("carried file");
1006 std::fs::write(prepared.join("new"), b"new").expect("new file");
1007 let journal_path = temporary.path().join("exchange.json");
1008 let mut journal = NativeExchangeJournal {
1009 version: NATIVE_EXCHANGE_JOURNAL_VERSION,
1010 operation: crate::IdempotencyKey::from_bytes([8; 16]),
1011 recovery: Vec::new(),
1012 live: live.clone(),
1013 prepared: prepared.clone(),
1014 carried: vec![PathBuf::from("private")],
1015 roots: [
1016 root_identity(&live).expect("live identity"),
1017 root_identity(&prepared).expect("prepared identity"),
1018 ],
1019 phase: NativeExchangePhase::Carrying,
1020 };
1021 write_journal(&journal_path, &journal).expect("carrying journal");
1022 carry(&live, &prepared, &journal.carried).expect("carry exclusion");
1023 journal.phase = NativeExchangePhase::Exchanging;
1024 write_journal(&journal_path, &journal).expect("exchange journal");
1025 let scratch = scratch(&live).expect("scratch path");
1026 durable_rename(
1027 &live,
1028 &scratch,
1029 acyclic_native_runtime::RenameMode::NoReplace,
1030 )
1031 .expect("move live to scratch");
1032 durable_rename(
1033 &prepared,
1034 &live,
1035 acyclic_native_runtime::RenameMode::NoReplace,
1036 )
1037 .expect("publish replacement");
1038
1039 let outcome = recover_native_exchange(&journal_path).expect("finish exchange");
1040 assert!(outcome.published);
1041 assert_eq!(outcome.displaced.as_deref(), Some(prepared.as_path()));
1042 assert_eq!(std::fs::read(live.join("new")).expect("new"), b"new");
1043 assert_eq!(
1044 std::fs::read(live.join("private/key")).expect("carried"),
1045 b"secret"
1046 );
1047 assert_eq!(std::fs::read(prepared.join("old")).expect("old"), b"old");
1048 assert!(!scratch.exists());
1049 assert!(!journal_path.exists());
1050 }
1051
1052 #[test]
1053 fn exchange_publishes_whole_tree_and_carries_nested_exclusion() {
1054 let temporary = tempfile::tempdir().expect("temporary directory");
1055 let live = temporary.path().join("live");
1056 let prepared = temporary.path().join("prepared");
1057 std::fs::create_dir_all(live.join("private/nested")).expect("live tree");
1058 std::fs::create_dir_all(&prepared).expect("prepared tree");
1059 std::fs::write(live.join("old"), b"old").expect("old file");
1060 std::fs::write(live.join("private/nested/key"), b"secret").expect("excluded file");
1061 std::fs::write(prepared.join("new"), b"new").expect("new file");
1062 let journal = temporary.path().join("exchange.json");
1063 let outcome = publish_native_exchange(
1064 &journal,
1065 &live,
1066 &prepared,
1067 crate::IdempotencyKey::from_bytes([2; 16]),
1068 vec![PathBuf::from("private/nested")],
1069 )
1070 .expect("publish");
1071 assert!(outcome.published);
1072 assert_eq!(std::fs::read(live.join("new")).expect("new"), b"new");
1073 assert_eq!(
1074 std::fs::read(live.join("private/nested/key")).expect("excluded"),
1075 b"secret"
1076 );
1077 assert_eq!(std::fs::read(prepared.join("old")).expect("old"), b"old");
1078 assert!(!journal.exists());
1079 }
1080
1081 #[test]
1082 fn stale_completed_journal_does_not_suppress_the_next_publication() {
1083 let temporary = tempfile::tempdir().expect("temporary directory");
1084 let live = temporary.path().join("live");
1085 let prepared = temporary.path().join("prepared");
1086 std::fs::create_dir(&live).expect("live tree");
1087 std::fs::create_dir(&prepared).expect("prepared tree");
1088 std::fs::write(live.join("first"), b"first").expect("first live file");
1089 std::fs::write(prepared.join("second"), b"second").expect("first prepared file");
1090 let journal_path = temporary.path().join("exchange.json");
1091 let stale = NativeExchangeJournal {
1092 version: NATIVE_EXCHANGE_JOURNAL_VERSION,
1093 operation: crate::IdempotencyKey::from_bytes([3; 16]),
1094 recovery: Vec::new(),
1095 live: live.clone(),
1096 prepared: prepared.clone(),
1097 carried: Vec::new(),
1098 roots: [
1099 root_identity(&live).expect("live identity"),
1100 root_identity(&prepared).expect("prepared identity"),
1101 ],
1102 phase: NativeExchangePhase::Exchanging,
1103 };
1104 write_journal(&journal_path, &stale).expect("stale journal");
1105 exchange(&live, &prepared).expect("completed first exchange");
1106
1107 let recovered = recover_native_exchange(&journal_path).expect("recover completed exchange");
1108 assert!(recovered.published);
1109 remove_entry(&prepared).expect("remove displaced tree");
1110 std::fs::create_dir(&prepared).expect("next prepared tree");
1111 std::fs::write(prepared.join("third"), b"third").expect("next prepared file");
1112
1113 let outcome = publish_native_exchange(
1114 &journal_path,
1115 &live,
1116 &prepared,
1117 crate::IdempotencyKey::from_bytes([4; 16]),
1118 Vec::new(),
1119 )
1120 .expect("publish after stale completed journal");
1121 assert!(outcome.published);
1122 assert_eq!(std::fs::read(live.join("third")).expect("third"), b"third");
1123 assert_eq!(
1124 std::fs::read(prepared.join("second")).expect("second"),
1125 b"second"
1126 );
1127 assert!(!journal_path.exists());
1128 }
1129
1130 #[test]
1131 fn completed_legacy_journal_requires_explicit_recovery_before_new_publication() {
1132 let temporary = tempfile::tempdir().expect("temporary directory");
1133 let live = temporary.path().join("live");
1134 let prepared = temporary.path().join("prepared");
1135 std::fs::create_dir(&live).expect("live tree");
1136 std::fs::create_dir(&prepared).expect("prepared tree");
1137 std::fs::write(live.join("first"), b"first").expect("first live file");
1138 std::fs::write(prepared.join("second"), b"second").expect("first prepared file");
1139 let journal_path = temporary.path().join("exchange.json");
1140 let mut encoded = serde_json::to_value(NativeExchangeJournal {
1141 version: NATIVE_EXCHANGE_JOURNAL_VERSION,
1142 operation: crate::IdempotencyKey::from_bytes([12; 16]),
1143 recovery: Vec::new(),
1144 live: live.clone(),
1145 prepared: prepared.clone(),
1146 carried: Vec::new(),
1147 roots: [
1148 root_identity(&live).expect("live identity"),
1149 root_identity(&prepared).expect("prepared identity"),
1150 ],
1151 phase: NativeExchangePhase::Exchanging,
1152 })
1153 .expect("encode journal");
1154 let object = encoded.as_object_mut().expect("journal object");
1155 object.remove("version");
1156 object.remove("operation");
1157 std::fs::write(
1158 &journal_path,
1159 serde_json::to_vec(&encoded).expect("encode legacy journal"),
1160 )
1161 .expect("write legacy journal");
1162 exchange(&live, &prepared).expect("completed first exchange");
1163 std::fs::remove_file(prepared.join("first")).expect("replace displaced contents");
1164 std::fs::write(prepared.join("third"), b"third").expect("next prepared file");
1165
1166 let operation = crate::IdempotencyKey::from_bytes([13; 16]);
1167 assert!(matches!(
1168 publish_native_exchange(&journal_path, &live, &prepared, operation, Vec::new(),),
1169 Err(NativeExchangeError::AmbiguousLegacyJournal)
1170 ));
1171 assert_eq!(
1172 std::fs::read(live.join("second")).expect("second"),
1173 b"second"
1174 );
1175 assert_eq!(
1176 std::fs::read(prepared.join("third")).expect("third"),
1177 b"third"
1178 );
1179 assert!(journal_path.exists());
1180
1181 let recovered = recover_native_exchange(&journal_path).expect("recover legacy exchange");
1182 assert!(recovered.published);
1183 assert_eq!(recovered.displaced.as_deref(), Some(prepared.as_path()));
1184 remove_entry(&prepared).expect("remove recovered displaced tree");
1185 std::fs::create_dir(&prepared).expect("rebuild prepared tree");
1186 std::fs::write(prepared.join("third"), b"third").expect("rebuilt prepared file");
1187 let outcome =
1188 publish_native_exchange(&journal_path, &live, &prepared, operation, Vec::new())
1189 .expect("publish after explicit legacy recovery");
1190 assert!(outcome.published);
1191 assert_eq!(std::fs::read(live.join("third")).expect("third"), b"third");
1192 assert_eq!(
1193 std::fs::read(prepared.join("second")).expect("second"),
1194 b"second"
1195 );
1196 assert!(!journal_path.exists());
1197 }
1198
1199 #[test]
1200 fn entry_exchange_preserves_both_complete_entries() {
1201 let temporary = tempfile::tempdir().expect("temporary directory");
1202 let left = temporary.path().join("left");
1203 let right = temporary.path().join("right");
1204 std::fs::create_dir(&left).expect("left tree");
1205 std::fs::create_dir(&right).expect("right tree");
1206 std::fs::write(left.join("value"), b"left").expect("left value");
1207 std::fs::write(right.join("value"), b"right").expect("right value");
1208 exchange_native_entries(&left, &right).expect("exchange entries");
1209 assert_eq!(std::fs::read(left.join("value")).expect("left"), b"right");
1210 assert_eq!(std::fs::read(right.join("value")).expect("right"), b"left");
1211 }
1212}