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        // Partial trigram batches of an indexer that died between two batches.
104        if path.is_dir() && path.file_name().and_then(|n| n.to_str()) == Some("trigram_temp") {
105            match fs::remove_dir_all(&path) {
106                Ok(()) => log::info!("Removed stale partial index directory {}", path.display()),
107                Err(e) => log::warn!(
108                    "Could not remove stale partial index directory {}: {}",
109                    path.display(),
110                    e
111                ),
112            }
113        }
114    }
115}
116
117/// RAII guard for the workspace index lock.
118///
119/// Dropping the guard (or the process exiting) releases the lock.
120#[derive(Debug)]
121pub struct IndexLock {
122    file: File,
123    path: PathBuf,
124}
125
126impl IndexLock {
127    /// Path of the lock file inside `cache_dir`.
128    pub fn lock_path(cache_dir: &Path) -> PathBuf {
129        cache_dir.join(INDEX_LOCK_FILE)
130    }
131
132    /// Try to take the lock without waiting.
133    ///
134    /// Returns `Ok(None)` when another process (or thread) holds it.
135    pub fn try_acquire(cache_dir: &Path) -> Result<Option<IndexLock>> {
136        fs::create_dir_all(cache_dir)
137            .with_context(|| format!("Failed to create {}", cache_dir.display()))?;
138        let path = Self::lock_path(cache_dir);
139        let file = OpenOptions::new()
140            .create(true)
141            .read(true)
142            .write(true)
143            .truncate(false)
144            .open(&path)
145            .with_context(|| format!("Failed to open {}", path.display()))?;
146        match file.try_lock() {
147            Ok(()) => Ok(Some(IndexLock { file, path })),
148            Err(std::fs::TryLockError::WouldBlock) => Ok(None),
149            Err(std::fs::TryLockError::Error(e)) => {
150                Err(e).with_context(|| format!("Failed to lock {}", path.display()))
151            }
152        }
153    }
154
155    /// Take the lock, polling every 100 ms until `timeout` elapses.
156    ///
157    /// Returns [`ReflexError::IndexLocked`] on timeout.
158    pub fn acquire_with_timeout(cache_dir: &Path, timeout: Duration) -> Result<IndexLock> {
159        let start = Instant::now();
160        loop {
161            if let Some(lock) = Self::try_acquire(cache_dir)? {
162                return Ok(lock);
163            }
164            if start.elapsed() >= timeout {
165                return Err(ReflexError::IndexLocked(
166                    Self::lock_path(cache_dir).display().to_string(),
167                )
168                .into());
169            }
170            std::thread::sleep(Duration::from_millis(100));
171        }
172    }
173
174    /// The lock file path this guard holds.
175    pub fn path(&self) -> &Path {
176        &self.path
177    }
178}
179
180impl Drop for IndexLock {
181    fn drop(&mut self) {
182        // Explicit unlock so the file handle's lifetime does not matter on
183        // platforms where the lock is tied to the descriptor.
184        let _ = self.file.unlock();
185    }
186}
187
188#[cfg(test)]
189mod tests {
190    use super::*;
191    use std::io::Write;
192    use tempfile::TempDir;
193
194    #[test]
195    fn tmp_path_is_sibling_with_suffix() {
196        let p = Path::new("/a/b/content.bin");
197        assert_eq!(tmp_path_for(p), PathBuf::from("/a/b/content.bin.tmp"));
198    }
199
200    #[test]
201    fn atomic_replace_moves_bytes_over_final() {
202        let dir = TempDir::new().unwrap();
203        let final_path = dir.path().join("f.bin");
204        fs::write(&final_path, b"old").unwrap();
205        let tmp = tmp_path_for(&final_path);
206        fs::write(&tmp, b"new-bytes").unwrap();
207        atomic_replace(&tmp, &final_path).unwrap();
208        assert_eq!(fs::read(&final_path).unwrap(), b"new-bytes");
209        assert!(!tmp.exists());
210    }
211
212    #[test]
213    fn remove_stale_tmp_only_touches_tmp_files() {
214        let dir = TempDir::new().unwrap();
215        fs::write(dir.path().join("content.bin"), b"keep").unwrap();
216        fs::write(dir.path().join("content.bin.tmp"), b"stale").unwrap();
217        remove_stale_tmp(dir.path());
218        assert!(dir.path().join("content.bin").exists());
219        assert!(!dir.path().join("content.bin.tmp").exists());
220    }
221
222    #[test]
223    fn second_acquire_in_other_process_scope_is_none_then_released() {
224        // Two handles in one process still exclude each other for
225        // `try_lock` on Linux/macOS/Windows (the lock is per open file
226        // description), which is what the indexer relies on.
227        let dir = TempDir::new().unwrap();
228        let first = IndexLock::try_acquire(dir.path()).unwrap();
229        assert!(first.is_some());
230        let second = IndexLock::try_acquire(dir.path()).unwrap();
231        assert!(second.is_none(), "lock must be exclusive while held");
232        drop(first);
233        let third = IndexLock::try_acquire(dir.path()).unwrap();
234        assert!(third.is_some(), "lock must be released on drop");
235    }
236
237    #[test]
238    fn acquire_with_timeout_reports_index_locked() {
239        let dir = TempDir::new().unwrap();
240        let _held = IndexLock::try_acquire(dir.path()).unwrap().unwrap();
241        let err = IndexLock::acquire_with_timeout(dir.path(), Duration::from_millis(250))
242            .expect_err("must time out");
243        let re = err
244            .downcast_ref::<ReflexError>()
245            .expect("typed ReflexError");
246        assert_eq!(re.kind(), "IndexLocked");
247        assert!(re.to_string().contains(INDEX_LOCK_FILE));
248    }
249
250    #[test]
251    fn lock_file_is_not_truncated_or_required_to_be_empty() {
252        let dir = TempDir::new().unwrap();
253        let path = IndexLock::lock_path(dir.path());
254        let mut f = File::create(&path).unwrap();
255        f.write_all(b"12345").unwrap();
256        drop(f);
257        let lock = IndexLock::try_acquire(dir.path()).unwrap().unwrap();
258        assert_eq!(lock.path(), path);
259    }
260}
reflex/atomic_write.rs

reflex/
atomic_write.rs
