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

harn_vm/
conditional_replace.rs
