winx_code_agent/utils/
mmap.rs

1//! Race-safe bounded file reads.
2//!
3//! The module name is retained for source compatibility, but project files are
4//! intentionally no longer memory-mapped. A repository file can be truncated
5//! by another editor or agent at any time; copying through `Read` keeps those
6//! races inside Rust's safe I/O model.
7
8use std::cmp::min;
9use std::fs::{File, Metadata};
10use std::io::{Read, Seek, SeekFrom};
11#[cfg(not(unix))]
12use std::time::SystemTime;
13use std::{path::Path, path::PathBuf, sync::Arc};
14
15use rayon::prelude::*;
16#[cfg(unix)]
17use std::os::unix::fs::MetadataExt as _;
18use tracing::debug;
19
20use crate::errors::{Result, WinxError};
21
22/// Legacy threshold retained for callers that used the old tuning constants.
23pub const DIRECT_READ_THRESHOLD: u64 = 10_000_000;
24/// Historical single-map ceiling, now the maximum `ShareableMap` snapshot size.
25pub const MAX_MMAP_SIZE: u64 = 1_000_000_000;
26/// Legacy segmented-map ceiling retained for API compatibility.
27pub const MAX_SEGMENTED_MMAP_SIZE: u64 = 4_000_000_000;
28/// Legacy map segment size retained for API compatibility.
29pub const SEGMENT_SIZE: u64 = 256_000_000;
30
31const READ_CHUNK_SIZE: usize = 256 * 1024;
32const MAX_PREALLOC_BYTES: usize = 64 * 1024 * 1024;
33const MAX_STABILITY_ATTEMPTS: usize = 3;
34
35#[derive(Clone, Debug, Eq, PartialEq)]
36struct FileVersion {
37    len: u64,
38    #[cfg(unix)]
39    device: u64,
40    #[cfg(unix)]
41    inode: u64,
42    #[cfg(unix)]
43    modified_seconds: i64,
44    #[cfg(unix)]
45    modified_nanoseconds: i64,
46    #[cfg(unix)]
47    changed_seconds: i64,
48    #[cfg(unix)]
49    changed_nanoseconds: i64,
50    #[cfg(not(unix))]
51    modified: Option<SystemTime>,
52}
53
54impl From<&Metadata> for FileVersion {
55    fn from(metadata: &Metadata) -> Self {
56        Self {
57            len: metadata.len(),
58            #[cfg(unix)]
59            device: metadata.dev(),
60            #[cfg(unix)]
61            inode: metadata.ino(),
62            #[cfg(unix)]
63            modified_seconds: metadata.mtime(),
64            #[cfg(unix)]
65            modified_nanoseconds: metadata.mtime_nsec(),
66            #[cfg(unix)]
67            changed_seconds: metadata.ctime(),
68            #[cfg(unix)]
69            changed_nanoseconds: metadata.ctime_nsec(),
70            #[cfg(not(unix))]
71            modified: metadata.modified().ok(),
72        }
73    }
74}
75
76fn access_error(path: &Path, operation: &str, error: impl std::fmt::Display) -> WinxError {
77    WinxError::FileAccessError {
78        path: path.to_path_buf(),
79        message: format!("{operation}: {error}"),
80    }
81}
82
83fn open_with_version(path: &Path, max_file_size: u64) -> Result<(File, FileVersion)> {
84    let file = File::open(path).map_err(|error| match error.kind() {
85        std::io::ErrorKind::NotFound => WinxError::FileNotFound { path: path.to_path_buf() },
86        std::io::ErrorKind::PermissionDenied => {
87            WinxError::FileOperationDenied { path: path.to_path_buf(), message: error.to_string() }
88        }
89        _ => access_error(path, "Error opening file", error),
90    })?;
91    let metadata = file
92        .metadata()
93        .map_err(|error| access_error(path, "Failed to get file metadata", error))?;
94    let version = FileVersion::from(&metadata);
95    if version.len > max_file_size {
96        return Err(WinxError::FileTooLarge {
97            path: path.to_path_buf(),
98            size: version.len,
99            max_size: max_file_size,
100        });
101    }
102    Ok((file, version))
103}
104
105fn bounded_capacity(reported_size: u64, limit: u64) -> usize {
106    let bounded = min(reported_size, limit);
107    usize::try_from(bounded).unwrap_or(usize::MAX).min(MAX_PREALLOC_BYTES)
108}
109
110fn read_bounded<R: Read>(
111    reader: &mut R,
112    reported_size: u64,
113    limit: u64,
114    path: &Path,
115) -> Result<Vec<u8>> {
116    let mut result = Vec::new();
117    result.try_reserve(bounded_capacity(reported_size, limit)).map_err(|error| {
118        WinxError::ResourceAllocationError {
119            message: format!(
120                "Unable to reserve a bounded file buffer for {}: {error}",
121                path.display()
122            ),
123        }
124    })?;
125    let mut chunk = vec![0_u8; READ_CHUNK_SIZE];
126
127    loop {
128        let count = match reader.read(&mut chunk) {
129            Ok(0) => break,
130            Ok(count) => count,
131            Err(error) if error.kind() == std::io::ErrorKind::Interrupted => continue,
132            Err(error) => return Err(access_error(path, "Error reading file", error)),
133        };
134        let observed = u64::try_from(result.len())
135            .unwrap_or(u64::MAX)
136            .saturating_add(u64::try_from(count).unwrap_or(u64::MAX));
137        if observed > limit {
138            return Err(WinxError::FileTooLarge {
139                path: path.to_path_buf(),
140                size: observed,
141                max_size: limit,
142            });
143        }
144        result.try_reserve(count).map_err(|error| WinxError::ResourceAllocationError {
145            message: format!("Unable to grow file buffer for {}: {error}", path.display()),
146        })?;
147        result.extend_from_slice(&chunk[..count]);
148    }
149    Ok(result)
150}
151
152fn snapshot_is_stable(file: &File, path: &Path, before: &FileVersion, observed: u64) -> bool {
153    let descriptor_after = file.metadata().ok().as_ref().map(FileVersion::from);
154    let path_after = std::fs::metadata(path).ok().as_ref().map(FileVersion::from);
155    let length_matches = before.len == 0 || observed == before.len;
156    descriptor_after.as_ref() == Some(before)
157        && path_after.as_ref() == Some(before)
158        && length_matches
159}
160
161fn changed_during_read(path: &Path) -> WinxError {
162    WinxError::ConcurrentFileModification {
163        path: path.to_path_buf(),
164        attempts: MAX_STABILITY_ATTEMPTS,
165    }
166}
167
168fn read_file_optimized_with_hook<F>(
169    path: &Path,
170    max_file_size: u64,
171    mut after_read: F,
172) -> Result<Vec<u8>>
173where
174    F: FnMut(usize, &Path),
175{
176    for attempt in 0..MAX_STABILITY_ATTEMPTS {
177        let (mut file, before) = open_with_version(path, max_file_size)?;
178        let bytes = read_bounded(&mut file, before.len, max_file_size, path)?;
179        after_read(attempt, path);
180        let observed = u64::try_from(bytes.len()).unwrap_or(u64::MAX);
181        if snapshot_is_stable(&file, path, &before, observed) {
182            return Ok(bytes);
183        }
184        debug!(
185            path = %path.display(),
186            attempt = attempt + 1,
187            "file changed during bounded read; retrying from a fresh descriptor"
188        );
189    }
190    Err(changed_during_read(path))
191}
192
193/// Read a complete file through bounded, race-detecting safe I/O.
194pub fn read_file_optimized(path: &Path, max_file_size: u64) -> Result<Vec<u8>> {
195    read_file_optimized_with_hook(path, max_file_size, |_, _| {})
196}
197
198/// Read a specific byte segment through bounded, race-detecting safe I/O.
199pub fn read_file_segment(
200    path: &Path,
201    offset: u64,
202    length: u64,
203    max_file_size: u64,
204) -> Result<Vec<u8>> {
205    for attempt in 0..MAX_STABILITY_ATTEMPTS {
206        let (mut file, before) = open_with_version(path, max_file_size)?;
207        if offset >= before.len {
208            return Err(WinxError::FileAccessError {
209                path: path.to_path_buf(),
210                message: format!("Offset {offset} exceeds file size {}", before.len),
211            });
212        }
213        let requested = min(length, before.len - offset);
214        file.seek(SeekFrom::Start(offset))
215            .map_err(|error| access_error(path, "Failed to seek to requested offset", error))?;
216        let bytes = {
217            let mut limited = (&mut file).take(requested);
218            read_bounded(&mut limited, requested, requested, path)?
219        };
220        let observed = u64::try_from(bytes.len()).unwrap_or(u64::MAX);
221        if observed == requested && snapshot_is_stable(&file, path, &before, before.len) {
222            return Ok(bytes);
223        }
224        debug!(
225            path = %path.display(),
226            attempt = attempt + 1,
227            "file changed during segment read; retrying from a fresh descriptor"
228        );
229    }
230    Err(changed_during_read(path))
231}
232
233/// Read a UTF-8 file through the bounded snapshot reader.
234pub fn read_file_to_string(path: &Path, max_file_size: u64) -> Result<String> {
235    let bytes = read_file_optimized(path, max_file_size)?;
236    String::from_utf8(bytes)
237        .map_err(|error| access_error(path, "Failed to decode file as UTF-8", error))
238}
239
240/// Process UTF-8 lines in parallel once the immutable snapshot exceeds 1 MB.
241pub fn process_text_file_parallel<F>(
242    path: &Path,
243    max_file_size: u64,
244    line_processor: F,
245) -> Result<()>
246where
247    F: Fn(&str) + Sync,
248{
249    let content = read_file_to_string(path, max_file_size)?;
250    if content.len() > 1_000_000 {
251        content.par_lines().for_each(&line_processor);
252    } else {
253        content.lines().for_each(line_processor);
254    }
255    Ok(())
256}
257
258/// Read one UTF-8 byte segment through the bounded snapshot reader.
259pub fn read_file_segment_to_string(
260    path: &Path,
261    offset: u64,
262    length: u64,
263    max_file_size: u64,
264) -> Result<String> {
265    let bytes = read_file_segment(path, offset, length, max_file_size)?;
266    String::from_utf8(bytes)
267        .map_err(|error| access_error(path, "Failed to decode file segment as UTF-8", error))
268}
269
270/// Cloneable immutable file snapshot retained under the historical API name.
271#[derive(Clone)]
272pub struct ShareableMap {
273    data: Arc<[u8]>,
274    path: PathBuf,
275}
276
277impl ShareableMap {
278    /// Snapshot a non-empty file without retaining an alias to the live inode.
279    pub fn new(path: &Path) -> Result<Self> {
280        let data = read_file_optimized(path, MAX_MMAP_SIZE)?;
281        if data.is_empty() {
282            return Err(WinxError::FileAccessError {
283                path: path.to_path_buf(),
284                message: "Cannot create a shared snapshot of an empty file".to_string(),
285            });
286        }
287        Ok(Self { data: Arc::from(data), path: path.to_path_buf() })
288    }
289
290    /// Snapshot a non-empty file segment without retaining a live mapping.
291    pub fn new_segment(path: &Path, offset: u64, length: u64) -> Result<Self> {
292        if length == 0 {
293            return Err(WinxError::FileAccessError {
294                path: path.to_path_buf(),
295                message: "Cannot create a shared snapshot of a zero-length segment".to_string(),
296            });
297        }
298        let file_size = std::fs::metadata(path)
299            .map_err(|error| access_error(path, "Failed to get file metadata", error))?
300            .len();
301        let data = read_file_segment(path, offset, length, file_size)?;
302        Ok(Self { data: Arc::from(data), path: path.to_path_buf() })
303    }
304
305    pub fn as_slice(&self) -> &[u8] {
306        &self.data
307    }
308
309    pub fn path(&self) -> &Path {
310        &self.path
311    }
312
313    pub fn len(&self) -> usize {
314        self.data.len()
315    }
316
317    pub fn is_empty(&self) -> bool {
318        self.data.is_empty()
319    }
320}
321
322#[cfg(test)]
323mod tests {
324    #![allow(clippy::cast_possible_truncation)]
325
326    use std::io::Write as _;
327    use std::sync::atomic::{AtomicBool, Ordering};
328
329    use tempfile::NamedTempFile;
330
331    use super::*;
332
333    fn create_test_file(size: usize) -> Result<(NamedTempFile, Vec<u8>)> {
334        let mut file = NamedTempFile::new()?;
335        let data = (0..size).map(|index| (index % 256) as u8).collect::<Vec<_>>();
336        file.write_all(&data)?;
337        file.flush()?;
338        Ok((file, data))
339    }
340
341    #[test]
342    fn bounded_reader_preserves_small_and_large_files() -> Result<()> {
343        for size in [10 * 1024, 11 * 1024 * 1024] {
344            let (file, expected) = create_test_file(size)?;
345            assert_eq!(read_file_optimized(file.path(), size as u64)?, expected);
346        }
347        Ok(())
348    }
349
350    #[test]
351    fn concurrent_replacement_retries_from_a_fresh_descriptor() -> Result<()> {
352        let mut file = NamedTempFile::new()?;
353        file.write_all(b"first revision")?;
354        file.flush()?;
355        let replaced = AtomicBool::new(false);
356
357        let result = read_file_optimized_with_hook(file.path(), 1_024, |attempt, path| {
358            if attempt == 0 && !replaced.swap(true, Ordering::AcqRel) {
359                let replaced = std::fs::write(path, b"second revision is current");
360                assert!(replaced.is_ok(), "failed to replace test file: {replaced:?}");
361            }
362        })?;
363
364        assert_eq!(result, b"second revision is current");
365        Ok(())
366    }
367
368    #[test]
369    fn file_size_limit_is_enforced_before_allocating_the_payload() -> Result<()> {
370        let (file, _) = create_test_file(4_096)?;
371        assert!(matches!(
372            read_file_optimized(file.path(), 1_024),
373            Err(WinxError::FileTooLarge { .. })
374        ));
375        Ok(())
376    }
377
378    #[test]
379    fn file_segment_read_uses_the_same_stable_snapshot_contract() -> Result<()> {
380        let (file, data) = create_test_file(1024 * 1024)?;
381        let offset = 100 * 1024;
382        let length = 200 * 1024;
383        let expected = &data[offset..offset + length];
384        let result =
385            read_file_segment(file.path(), offset as u64, length as u64, data.len() as u64)?;
386        assert_eq!(result, expected);
387        Ok(())
388    }
389
390    #[test]
391    fn shareable_map_is_an_owned_snapshot() -> Result<()> {
392        let (file, data) = create_test_file(100 * 1024)?;
393        let snapshot = ShareableMap::new(file.path())?;
394        std::fs::write(file.path(), b"changed after snapshot")?;
395        assert_eq!(snapshot.as_slice(), &data);
396
397        let segment = ShareableMap::new_segment(snapshot.path(), 0, 7)?;
398        assert_eq!(segment.as_slice(), b"changed");
399        Ok(())
400    }
401
402    #[test]
403    fn parallel_processing_observes_every_line() -> Result<()> {
404        let mut file = NamedTempFile::new()?;
405        let expected = (0..1_000).map(|index| format!("Line {index}")).collect::<Vec<_>>();
406        for line in &expected {
407            writeln!(file, "{line}")?;
408        }
409        file.flush()?;
410
411        let processed = std::sync::Mutex::new(Vec::new());
412        process_text_file_parallel(file.path(), 1_000_000, |line| {
413            if let Ok(mut entries) = processed.lock() {
414                entries.push(line.to_string());
415            }
416        })?;
417        let result = processed.lock().map_err(|error| WinxError::ResourceAllocationError {
418            message: format!("Failed to lock processed lines: {error}"),
419        })?;
420        assert_eq!(result.len(), expected.len());
421        assert!(expected.iter().all(|line| result.contains(line)));
422        Ok(())
423    }
424}
winx_code_agent/utils/mmap.rs

winx_code_agent/utils/
mmap.rs
