harn_vm/
conditional_replace.rs

1//! Cross-process compare-and-replace for complete file payloads.
2
3use std::cell::RefCell;
4use std::fs::{self, File, OpenOptions};
5use std::io;
6use std::path::{Path, PathBuf};
7use std::time::Duration;
8
9use sha2::{Digest, Sha256};
10
11use crate::atomic_io::{
12    atomic_write_with_durability_unlocked, AtomicWriteDurability, AtomicWriteReceipt,
13};
14
15thread_local! {
16    static EXECUTION_LOCK_ROOT: RefCell<Option<PathBuf>> = const { RefCell::new(None) };
17}
18
19// The critical section performs one local read and one atomic replacement.
20// Thirty seconds leaves ample room for a slow disk without hiding a dead
21// holder indefinitely.
22const CONDITIONAL_REPLACE_LOCK_TIMEOUT: Duration = Duration::from_secs(30);
23
24/// Restores the prior execution-local replacement lock root on drop.
25#[derive(Debug)]
26#[must_use = "retain this guard for the execution that owns the lock root"]
27pub struct ScopedConditionalReplaceLockRoot {
28    previous: Option<PathBuf>,
29}
30
31/// Route compare-and-replace locks through a caller-owned root on this thread.
32/// Worker threads must install their own guard; the override is intended for
33/// isolated embedders and tests, not as ambient workflow state.
34pub fn scope_conditional_replace_lock_root(
35    root: impl AsRef<Path>,
36) -> ScopedConditionalReplaceLockRoot {
37    let previous = EXECUTION_LOCK_ROOT.with(|slot| slot.replace(Some(root.as_ref().to_path_buf())));
38    ScopedConditionalReplaceLockRoot { previous }
39}
40
41impl Drop for ScopedConditionalReplaceLockRoot {
42    fn drop(&mut self) {
43        EXECUTION_LOCK_ROOT.with(|slot| {
44            slot.replace(self.previous.take());
45        });
46    }
47}
48
49/// Preconditions and durability for one replacement.
50#[derive(Clone, Debug, Eq, PartialEq)]
51pub struct ConditionalReplaceOptions {
52    /// Digest observed by the caller, including the `sha256:` prefix.
53    pub expected_sha256: Option<String>,
54    /// Allow a missing destination to be created.
55    pub create: bool,
56    /// Allow an existing destination to change.
57    pub overwrite: bool,
58    /// Create a missing parent chain.
59    pub create_parents: bool,
60    /// Requested namespace or storage-flush durability.
61    pub durability: AtomicWriteDurability,
62}
63
64impl Default for ConditionalReplaceOptions {
65    fn default() -> Self {
66        Self {
67            expected_sha256: None,
68            create: true,
69            overwrite: true,
70            create_parents: true,
71            durability: AtomicWriteDurability::Flush,
72        }
73    }
74}
75
76/// Closed successful outcome for one replacement attempt.
77#[derive(Clone, Copy, Debug, Eq, PartialEq)]
78pub enum ConditionalReplaceStatus {
79    Created,
80    Replaced,
81    NoOp,
82    Stale,
83}
84
85impl ConditionalReplaceStatus {
86    pub fn as_str(self) -> &'static str {
87        match self {
88            Self::Created => "created",
89            Self::Replaced => "replaced",
90            Self::NoOp => "no_op",
91            Self::Stale => "stale",
92        }
93    }
94}
95
96/// Receipt for one compare-and-replace operation.
97#[derive(Clone, Debug, Eq, PartialEq)]
98pub struct ConditionalReplaceReceipt {
99    /// Closed operation outcome.
100    pub status: ConditionalReplaceStatus,
101    /// Whether the destination existed before this attempt.
102    pub before_exists: bool,
103    /// Digest observed while the replacement lock was held.
104    pub before_sha256: String,
105    /// Digest of the requested complete payload.
106    pub after_sha256: String,
107    /// Caller-supplied lease, when present.
108    pub expected_sha256: Option<String>,
109    /// Payload bytes written; zero for `NoOp` and `Stale`.
110    pub bytes_written: usize,
111    /// Whether the complete payload was flushed to the operating system.
112    pub file_synced: bool,
113    /// Whether persistence of the namespace replacement was confirmed.
114    pub namespace_synced: bool,
115}
116
117/// Replace a complete file payload under a canonical-path cross-process lock.
118pub fn conditional_replace(
119    path: &Path,
120    contents: &[u8],
121    options: &ConditionalReplaceOptions,
122) -> io::Result<ConditionalReplaceReceipt> {
123    conditional_replace_with_hook(path, contents, options, || {})
124}
125
126/// Variant used by hosts that must snapshot the pre-image immediately before
127/// mutation while the replacement lock is still held.
128pub fn conditional_replace_with_hook<F>(
129    path: &Path,
130    contents: &[u8],
131    options: &ConditionalReplaceOptions,
132    before_write: F,
133) -> io::Result<ConditionalReplaceReceipt>
134where
135    F: FnOnce(),
136{
137    conditional_replace_with_io(
138        path,
139        contents,
140        options,
141        |candidate| {
142            reject_symlink_destination(candidate)?;
143            fs::read(candidate)
144        },
145        |candidate, bytes, durability, create_parents| {
146            require_parent(candidate, create_parents)?;
147            // The compare-and-replace lock already covers the complete read,
148            // lease check, and write boundary.
149            atomic_write_with_durability_unlocked(candidate, bytes, durability)
150        },
151        before_write,
152    )
153}
154
155pub(crate) fn conditional_replace_with_io<R, W, F>(
156    path: &Path,
157    contents: &[u8],
158    options: &ConditionalReplaceOptions,
159    read: R,
160    write: W,
161    before_write: F,
162) -> io::Result<ConditionalReplaceReceipt>
163where
164    R: FnOnce(&Path) -> io::Result<Vec<u8>>,
165    W: FnOnce(&Path, &[u8], AtomicWriteDurability, bool) -> io::Result<AtomicWriteReceipt>,
166    F: FnOnce(),
167{
168    let _lock = acquire_lock(path)?;
169    let (before, before_exists) = match read(path) {
170        Ok(bytes) => (bytes, true),
171        Err(error) if error.kind() == io::ErrorKind::NotFound => (Vec::new(), false),
172        Err(error) => return Err(error),
173    };
174    let before_sha256 = sha256_label(&before);
175    let after_sha256 = sha256_label(contents);
176
177    if options
178        .expected_sha256
179        .as_deref()
180        .is_some_and(|expected| expected != before_sha256)
181    {
182        return Ok(ConditionalReplaceReceipt {
183            status: ConditionalReplaceStatus::Stale,
184            before_exists,
185            before_sha256,
186            after_sha256,
187            expected_sha256: options.expected_sha256.clone(),
188            bytes_written: 0,
189            file_synced: false,
190            namespace_synced: false,
191        });
192    }
193
194    if before_exists && before == contents {
195        return Ok(ConditionalReplaceReceipt {
196            status: ConditionalReplaceStatus::NoOp,
197            before_exists: true,
198            before_sha256,
199            after_sha256,
200            expected_sha256: options.expected_sha256.clone(),
201            bytes_written: 0,
202            file_synced: false,
203            namespace_synced: false,
204        });
205    }
206    if before_exists && !options.overwrite {
207        return Err(io::Error::new(
208            io::ErrorKind::AlreadyExists,
209            format!("'{}' exists and overwrite=false", path.display()),
210        ));
211    }
212    if !before_exists && !options.create {
213        return Err(io::Error::new(
214            io::ErrorKind::NotFound,
215            format!("'{}' does not exist and create=false", path.display()),
216        ));
217    }
218    before_write();
219    let durability = write(path, contents, options.durability, options.create_parents)?;
220    Ok(ConditionalReplaceReceipt {
221        status: if before_exists {
222            ConditionalReplaceStatus::Replaced
223        } else {
224            ConditionalReplaceStatus::Created
225        },
226        before_exists,
227        before_sha256,
228        after_sha256,
229        expected_sha256: options.expected_sha256.clone(),
230        bytes_written: contents.len(),
231        file_synced: durability.file_synced,
232        namespace_synced: durability.namespace_synced,
233    })
234}
235
236fn reject_symlink_destination(path: &Path) -> io::Result<()> {
237    match fs::symlink_metadata(path) {
238        Ok(metadata) if metadata.file_type().is_symlink() => Err(io::Error::new(
239            io::ErrorKind::InvalidInput,
240            format!(
241                "refusing to replace symlink destination '{}'",
242                path.display()
243            ),
244        )),
245        Ok(_) => Ok(()),
246        Err(error) if error.kind() == io::ErrorKind::NotFound => Ok(()),
247        Err(error) => Err(error),
248    }
249}
250
251pub(crate) fn require_parent(path: &Path, create_parents: bool) -> io::Result<()> {
252    if create_parents {
253        return Ok(());
254    }
255    if let Some(parent) = path.parent() {
256        if !parent.as_os_str().is_empty() && !parent.is_dir() {
257            return Err(io::Error::new(
258                io::ErrorKind::NotFound,
259                format!(
260                    "parent directory for '{}' does not exist (pass create_parents=true to create it)",
261                    path.display()
262                ),
263            ));
264        }
265    }
266    Ok(())
267}
268
269fn sha256_label(bytes: &[u8]) -> String {
270    format!("sha256:{}", hex::encode(Sha256::digest(bytes)))
271}
272
273fn lock_root() -> PathBuf {
274    if let Some(root) = EXECUTION_LOCK_ROOT.with(|slot| slot.borrow().clone()) {
275        return root;
276    }
277    let runtime_root = crate::stdlib::process::runtime_root_base();
278    crate::runtime_paths::state_root(&runtime_root).join("fs-cas-locks")
279}
280
281pub(crate) fn acquire_lock(path: &Path) -> io::Result<ConditionalReplaceLock> {
282    acquire_lock_with_timeout(path, CONDITIONAL_REPLACE_LOCK_TIMEOUT)
283}
284
285fn acquire_lock_with_timeout(path: &Path, timeout: Duration) -> io::Result<ConditionalReplaceLock> {
286    let root = lock_root();
287    fs::create_dir_all(&root)?;
288    let identity = canonical_lock_identity(path);
289    let name = format!(
290        "{}.lock",
291        hex::encode(Sha256::digest(lock_identity_bytes(&identity)))
292    );
293    let lock_path = root.join(name);
294    let file = OpenOptions::new()
295        .create(true)
296        .truncate(false)
297        .read(true)
298        .write(true)
299        .open(&lock_path)?;
300    harn_flock::lock_with_deadline(&file, &lock_path, harn_flock::LockMode::Exclusive, timeout)
301        .map_err(io::Error::other)?;
302    Ok(ConditionalReplaceLock { file })
303}
304
305fn lock_identity_bytes(identity: &Path) -> Vec<u8> {
306    #[cfg(any(target_os = "macos", target_os = "windows"))]
307    {
308        identity.to_string_lossy().to_lowercase().into_bytes()
309    }
310    #[cfg(not(any(target_os = "macos", target_os = "windows")))]
311    {
312        identity.as_os_str().as_encoded_bytes().to_vec()
313    }
314}
315
316fn canonical_lock_identity(path: &Path) -> PathBuf {
317    if let Ok(canonical) = fs::canonicalize(path) {
318        return canonical;
319    }
320    let absolute = if path.is_absolute() {
321        path.to_path_buf()
322    } else {
323        std::env::current_dir()
324            .unwrap_or_else(|_| PathBuf::from("."))
325            .join(path)
326    };
327    let mut ancestor = absolute.as_path();
328    let mut suffix = Vec::new();
329    while let Some(name) = ancestor.file_name() {
330        suffix.push(name.to_os_string());
331        let Some(parent) = ancestor.parent() else {
332            break;
333        };
334        if let Ok(canonical_parent) = fs::canonicalize(parent) {
335            let mut identity = canonical_parent;
336            for component in suffix.iter().rev() {
337                identity.push(component);
338            }
339            return identity;
340        }
341        ancestor = parent;
342    }
343    absolute
344}
345
346pub(crate) struct ConditionalReplaceLock {
347    file: File,
348}
349
350impl Drop for ConditionalReplaceLock {
351    fn drop(&mut self) {
352        let _ = self.file.unlock();
353    }
354}
355
356#[cfg(test)]
357mod tests {
358    use super::*;
359    use std::sync::{Arc, Barrier};
360
361    #[test]
362    fn stale_digest_never_writes() {
363        let dir = tempfile::tempdir().unwrap();
364        let _locks = scope_conditional_replace_lock_root(dir.path().join("locks"));
365        let path = dir.path().join("state.json");
366        fs::write(&path, b"current").unwrap();
367        let options = ConditionalReplaceOptions {
368            expected_sha256: Some(sha256_label(b"older")),
369            ..ConditionalReplaceOptions::default()
370        };
371        let receipt = conditional_replace(&path, b"new", &options).unwrap();
372        assert_eq!(receipt.status, ConditionalReplaceStatus::Stale);
373        assert_eq!(fs::read(&path).unwrap(), b"current");
374    }
375
376    #[test]
377    fn create_replace_and_no_op_are_distinct() {
378        let dir = tempfile::tempdir().unwrap();
379        let _locks = scope_conditional_replace_lock_root(dir.path().join("locks"));
380        let path = dir.path().join("state.json");
381        let options = ConditionalReplaceOptions::default();
382
383        let created = conditional_replace(&path, b"one", &options).unwrap();
384        assert_eq!(created.status, ConditionalReplaceStatus::Created);
385        let no_op = conditional_replace(&path, b"one", &options).unwrap();
386        assert_eq!(no_op.status, ConditionalReplaceStatus::NoOp);
387        let replaced = conditional_replace(&path, b"two", &options).unwrap();
388        assert_eq!(replaced.status, ConditionalReplaceStatus::Replaced);
389    }
390
391    #[test]
392    fn one_concurrent_writer_wins_an_observed_digest() {
393        let dir = tempfile::tempdir().unwrap();
394        let path = Arc::new(dir.path().join("state.json"));
395        let lock_root = Arc::new(dir.path().join("locks"));
396        fs::write(path.as_ref(), b"original").unwrap();
397        let expected = sha256_label(b"original");
398        let barrier = Arc::new(Barrier::new(17));
399        let mut workers = Vec::new();
400        for index in 0..16 {
401            let path = Arc::clone(&path);
402            let barrier = Arc::clone(&barrier);
403            let expected = expected.clone();
404            let lock_root = Arc::clone(&lock_root);
405            workers.push(std::thread::spawn(move || {
406                let _locks = scope_conditional_replace_lock_root(lock_root.as_ref());
407                let payload = format!("writer-{index}");
408                let options = ConditionalReplaceOptions {
409                    expected_sha256: Some(expected),
410                    ..ConditionalReplaceOptions::default()
411                };
412                barrier.wait();
413                conditional_replace(path.as_ref(), payload.as_bytes(), &options).unwrap()
414            }));
415        }
416        barrier.wait();
417        let receipts: Vec<_> = workers
418            .into_iter()
419            .map(|worker| worker.join().unwrap())
420            .collect();
421        assert_eq!(
422            receipts
423                .iter()
424                .filter(|receipt| receipt.status == ConditionalReplaceStatus::Replaced)
425                .count(),
426            1
427        );
428        assert_eq!(
429            receipts
430                .iter()
431                .filter(|receipt| receipt.status == ConditionalReplaceStatus::Stale)
432                .count(),
433            15
434        );
435    }
436
437    #[test]
438    fn canonical_path_aliases_share_a_lock_identity() {
439        let dir = tempfile::tempdir().unwrap();
440        fs::create_dir(dir.path().join("sub")).unwrap();
441        let path = dir.path().join("state.json");
442        let alias = dir.path().join("sub/../state.json");
443        fs::write(&path, b"original").unwrap();
444        assert_eq!(
445            canonical_lock_identity(&path),
446            canonical_lock_identity(&alias)
447        );
448    }
449
450    #[test]
451    fn held_replace_lock_reports_its_path() {
452        let dir = tempfile::tempdir().unwrap();
453        let _locks = scope_conditional_replace_lock_root(dir.path().join("locks"));
454        let path = dir.path().join("state.json");
455        let _holder = acquire_lock_with_timeout(&path, Duration::ZERO).unwrap();
456
457        let error = match acquire_lock_with_timeout(&path, Duration::ZERO) {
458            Ok(_) => panic!("a second lock must not pass the holder"),
459            Err(error) => error,
460        };
461
462        assert!(
463            error
464                .to_string()
465                .contains(&dir.path().join("locks").display().to_string()),
466            "{error}"
467        );
468        assert!(error.to_string().contains("timed out"), "{error}");
469    }
470
471    #[cfg(unix)]
472    #[test]
473    fn symlink_destinations_fail_closed() {
474        let dir = tempfile::tempdir().unwrap();
475        let _locks = scope_conditional_replace_lock_root(dir.path().join("locks"));
476        let target = dir.path().join("target.txt");
477        let alias = dir.path().join("alias.txt");
478        fs::write(&target, b"original").unwrap();
479        std::os::unix::fs::symlink(&target, &alias).unwrap();
480
481        let error =
482            conditional_replace(&alias, b"new", &ConditionalReplaceOptions::default()).unwrap_err();
483        assert_eq!(error.kind(), io::ErrorKind::InvalidInput);
484        assert_eq!(fs::read(&target).unwrap(), b"original");
485        assert!(fs::symlink_metadata(alias)
486            .unwrap()
487            .file_type()
488            .is_symlink());
489    }
490
491    #[test]
492    fn write_failure_preserves_the_preimage() {
493        let dir = tempfile::tempdir().unwrap();
494        let _locks = scope_conditional_replace_lock_root(dir.path().join("locks"));
495        let path = dir.path().join("state.json");
496        fs::write(&path, b"old").unwrap();
497        let hook_calls = std::cell::Cell::new(0);
498        let error = conditional_replace_with_io(
499            &path,
500            b"new",
501            &ConditionalReplaceOptions::default(),
502            |candidate| fs::read(candidate),
503            |_, _, _, _| Err(io::Error::other("injected write failure")),
504            || hook_calls.set(hook_calls.get() + 1),
505        )
506        .unwrap_err();
507        assert_eq!(error.to_string(), "injected write failure");
508        assert_eq!(hook_calls.get(), 1);
509        assert_eq!(fs::read(path).unwrap(), b"old");
510    }
511
512    #[test]
513    fn create_and_overwrite_policies_fail_closed() {
514        let dir = tempfile::tempdir().unwrap();
515        let _locks = scope_conditional_replace_lock_root(dir.path().join("locks"));
516        let path = dir.path().join("state.json");
517        let no_create = ConditionalReplaceOptions {
518            create: false,
519            ..ConditionalReplaceOptions::default()
520        };
521        assert_eq!(
522            conditional_replace(&path, b"new", &no_create)
523                .unwrap_err()
524                .kind(),
525            io::ErrorKind::NotFound
526        );
527        fs::write(&path, b"old").unwrap();
528        let no_overwrite = ConditionalReplaceOptions {
529            overwrite: false,
530            ..ConditionalReplaceOptions::default()
531        };
532        assert_eq!(
533            conditional_replace(&path, b"new", &no_overwrite)
534                .unwrap_err()
535                .kind(),
536            io::ErrorKind::AlreadyExists
537        );
538        assert_eq!(fs::read(path).unwrap(), b"old");
539    }
540}
harn_vm/conditional_replace.rs

harn_vm/
conditional_replace.rs
