reflex/
atomic_write.rs

1//! Crash-safe file replacement and the workspace index lock.
2//!
3//! Index binaries (`content.bin`, `trigrams.bin`) used to be opened at their
4//! final path with `truncate(true)` and streamed into. A crash, a killed
5//! process, or a second indexer starting mid-write left a short file behind,
6//! and the next reader failed with `content.bin is too small - appears to be
7//! corrupted`. Every writer now goes through this module:
8//!
9//! 1. write to `<final>.tmp` in the same directory,
10//! 2. `sync_all`,
11//! 3. [`atomic_replace`] renames the temp file over the final path.
12//!
13//! Readers therefore only ever see the previous complete file or the new
14//! complete file. Because the rename lands in the same directory, it is
15//! atomic on every platform we ship to.
16//!
17//! [`IndexLock`] is an OS advisory lock on `.reflex/index.lock` held for the
18//! whole `Indexer::index` run. The OS releases it when the process dies, so
19//! there is no stale-PID bookkeeping.
20
21use std::fs::{self, File, OpenOptions};
22use std::io;
23use std::path::{Path, PathBuf};
24use std::time::{Duration, Instant};
25
26use anyhow::{Context, Result};
27
28use crate::errors::ReflexError;
29
30/// Suffix appended to a final path to build its in-progress temp path.
31pub const TMP_SUFFIX: &str = ".tmp";
32
33/// File name of the workspace index lock inside the cache directory.
34pub const INDEX_LOCK_FILE: &str = "index.lock";
35
36/// Temp path for `final_path`: same directory, same file name plus `.tmp`.
37///
38/// Same directory matters: `fs::rename` is only atomic within one filesystem.
39pub fn tmp_path_for(final_path: &Path) -> PathBuf {
40    let mut name = final_path
41        .file_name()
42        .map(|n| n.to_os_string())
43        .unwrap_or_default();
44    name.push(TMP_SUFFIX);
45    final_path.with_file_name(name)
46}
47
48/// Rename `tmp` over `final_path`.
49///
50/// On Windows a rename can fail with a sharing violation while another
51/// process still maps the old file (the background symbol indexer or an MCP
52/// query mid-flight). We retry with backoff, then fall back to copying the
53/// bytes over the final path so the index still lands. The fallback is not
54/// atomic, so it logs a warning.
55pub fn atomic_replace(tmp: &Path, final_path: &Path) -> io::Result<()> {
56    let mut last_err = match fs::rename(tmp, final_path) {
57        Ok(()) => return Ok(()),
58        Err(e) => e,
59    };
60
61    if cfg!(windows) {
62        for delay_ms in [20u64, 40, 80, 160, 320] {
63            std::thread::sleep(Duration::from_millis(delay_ms));
64            match fs::rename(tmp, final_path) {
65                Ok(()) => return Ok(()),
66                Err(e) => last_err = e,
67            }
68        }
69        log::warn!(
70            "atomic rename of {} failed after retries ({}); falling back to non-atomic copy",
71            final_path.display(),
72            last_err
73        );
74        fs::copy(tmp, final_path)?;
75        let _ = fs::remove_file(tmp);
76        return Ok(());
77    }
78
79    Err(last_err)
80}
81
82/// Remove leftover `*.tmp` files in `dir` (a crashed indexer leaves them).
83///
84/// Errors are logged, never fatal: a stray temp file only wastes disk.
85pub fn remove_stale_tmp(dir: &Path) {
86    let entries = match fs::read_dir(dir) {
87        Ok(e) => e,
88        Err(_) => return,
89    };
90    for entry in entries.flatten() {
91        let path = entry.path();
92        let is_tmp = path
93            .file_name()
94            .and_then(|n| n.to_str())
95            .map(|n| n.ends_with(TMP_SUFFIX))
96            .unwrap_or(false);
97        if is_tmp && path.is_file() {
98            match fs::remove_file(&path) {
99                Ok(()) => log::info!("Removed stale temp file {}", path.display()),
100                Err(e) => log::warn!("Could not remove stale temp file {}: {}", path.display(), e),
101            }
102        }
103    }
104}
105
106/// RAII guard for the workspace index lock.
107///
108/// Dropping the guard (or the process exiting) releases the lock.
109#[derive(Debug)]
110pub struct IndexLock {
111    file: File,
112    path: PathBuf,
113}
114
115impl IndexLock {
116    /// Path of the lock file inside `cache_dir`.
117    pub fn lock_path(cache_dir: &Path) -> PathBuf {
118        cache_dir.join(INDEX_LOCK_FILE)
119    }
120
121    /// Try to take the lock without waiting.
122    ///
123    /// Returns `Ok(None)` when another process (or thread) holds it.
124    pub fn try_acquire(cache_dir: &Path) -> Result<Option<IndexLock>> {
125        fs::create_dir_all(cache_dir)
126            .with_context(|| format!("Failed to create {}", cache_dir.display()))?;
127        let path = Self::lock_path(cache_dir);
128        let file = OpenOptions::new()
129            .create(true)
130            .read(true)
131            .write(true)
132            .truncate(false)
133            .open(&path)
134            .with_context(|| format!("Failed to open {}", path.display()))?;
135        match file.try_lock() {
136            Ok(()) => Ok(Some(IndexLock { file, path })),
137            Err(std::fs::TryLockError::WouldBlock) => Ok(None),
138            Err(std::fs::TryLockError::Error(e)) => {
139                Err(e).with_context(|| format!("Failed to lock {}", path.display()))
140            }
141        }
142    }
143
144    /// Take the lock, polling every 100 ms until `timeout` elapses.
145    ///
146    /// Returns [`ReflexError::IndexLocked`] on timeout.
147    pub fn acquire_with_timeout(cache_dir: &Path, timeout: Duration) -> Result<IndexLock> {
148        let start = Instant::now();
149        loop {
150            if let Some(lock) = Self::try_acquire(cache_dir)? {
151                return Ok(lock);
152            }
153            if start.elapsed() >= timeout {
154                return Err(ReflexError::IndexLocked(
155                    Self::lock_path(cache_dir).display().to_string(),
156                )
157                .into());
158            }
159            std::thread::sleep(Duration::from_millis(100));
160        }
161    }
162
163    /// The lock file path this guard holds.
164    pub fn path(&self) -> &Path {
165        &self.path
166    }
167}
168
169impl Drop for IndexLock {
170    fn drop(&mut self) {
171        // Explicit unlock so the file handle's lifetime does not matter on
172        // platforms where the lock is tied to the descriptor.
173        let _ = self.file.unlock();
174    }
175}
176
177#[cfg(test)]
178mod tests {
179    use super::*;
180    use std::io::Write;
181    use tempfile::TempDir;
182
183    #[test]
184    fn tmp_path_is_sibling_with_suffix() {
185        let p = Path::new("/a/b/content.bin");
186        assert_eq!(tmp_path_for(p), PathBuf::from("/a/b/content.bin.tmp"));
187    }
188
189    #[test]
190    fn atomic_replace_moves_bytes_over_final() {
191        let dir = TempDir::new().unwrap();
192        let final_path = dir.path().join("f.bin");
193        fs::write(&final_path, b"old").unwrap();
194        let tmp = tmp_path_for(&final_path);
195        fs::write(&tmp, b"new-bytes").unwrap();
196        atomic_replace(&tmp, &final_path).unwrap();
197        assert_eq!(fs::read(&final_path).unwrap(), b"new-bytes");
198        assert!(!tmp.exists());
199    }
200
201    #[test]
202    fn remove_stale_tmp_only_touches_tmp_files() {
203        let dir = TempDir::new().unwrap();
204        fs::write(dir.path().join("content.bin"), b"keep").unwrap();
205        fs::write(dir.path().join("content.bin.tmp"), b"stale").unwrap();
206        remove_stale_tmp(dir.path());
207        assert!(dir.path().join("content.bin").exists());
208        assert!(!dir.path().join("content.bin.tmp").exists());
209    }
210
211    #[test]
212    fn second_acquire_in_other_process_scope_is_none_then_released() {
213        // Two handles in one process still exclude each other for
214        // `try_lock` on Linux/macOS/Windows (the lock is per open file
215        // description), which is what the indexer relies on.
216        let dir = TempDir::new().unwrap();
217        let first = IndexLock::try_acquire(dir.path()).unwrap();
218        assert!(first.is_some());
219        let second = IndexLock::try_acquire(dir.path()).unwrap();
220        assert!(second.is_none(), "lock must be exclusive while held");
221        drop(first);
222        let third = IndexLock::try_acquire(dir.path()).unwrap();
223        assert!(third.is_some(), "lock must be released on drop");
224    }
225
226    #[test]
227    fn acquire_with_timeout_reports_index_locked() {
228        let dir = TempDir::new().unwrap();
229        let _held = IndexLock::try_acquire(dir.path()).unwrap().unwrap();
230        let err = IndexLock::acquire_with_timeout(dir.path(), Duration::from_millis(250))
231            .expect_err("must time out");
232        let re = err
233            .downcast_ref::<ReflexError>()
234            .expect("typed ReflexError");
235        assert_eq!(re.kind(), "IndexLocked");
236        assert!(re.to_string().contains(INDEX_LOCK_FILE));
237    }
238
239    #[test]
240    fn lock_file_is_not_truncated_or_required_to_be_empty() {
241        let dir = TempDir::new().unwrap();
242        let path = IndexLock::lock_path(dir.path());
243        let mut f = File::create(&path).unwrap();
244        f.write_all(b"12345").unwrap();
245        drop(f);
246        let lock = IndexLock::try_acquire(dir.path()).unwrap().unwrap();
247        assert_eq!(lock.path(), path);
248    }
249}
reflex/atomic_write.rs

reflex/
atomic_write.rs
