diffs_rs/
input.rs

1use std::{
2    io::{BufRead, BufReader, Read},
3    os::unix::ffi::{OsStrExt, OsStringExt},
4    path::PathBuf,
5    sync::{
6        Arc,
7        atomic::{AtomicBool, AtomicUsize, Ordering},
8    },
9};
10
11use anyhow::{Context, Result, ensure};
12use serde::Serialize;
13
14use crate::{
15    source::{CheckedPair, Snapshot, SourceStore, check_pair},
16    storage::{self, FileIndex, StoreWriter},
17};
18
19mod files;
20mod git;
21mod metadata;
22mod patch;
23#[cfg(test)]
24mod tests;
25mod worker;
26pub use worker::{Message, Worker};
27
28pub enum PatchInput {
29    Path(PathBuf),
30    /// Reads from the current offset. Requires exclusive control of reading, the shared
31    /// cursor, and status flags, including any retained duplicates, until worker close.
32    Stream(std::fs::File),
33}
34
35#[derive(Clone, Debug)]
36pub enum LocalInput {
37    Files {
38        old: PathBuf,
39        new: PathBuf,
40    },
41    Git {
42        repo: PathBuf,
43        base: String,
44        head: String,
45    },
46}
47
48impl LocalInput {
49    pub fn label(&self) -> String {
50        let label = match self {
51            Self::Files { old, new } => format!("{} → {}", old.display(), new.display()),
52            Self::Git { repo, base, head } => format!("{} · {base} → {head}", repo.display()),
53        };
54        crate::text::safe_label(label.as_bytes())
55    }
56}
57
58#[derive(Debug)]
59pub struct Cancelled;
60impl std::fmt::Display for Cancelled {
61    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
62        f.write_str("Source preparation canceled")
63    }
64}
65impl std::error::Error for Cancelled {}
66
67#[derive(Clone, Default)]
68pub struct Cancellation {
69    stopped: Arc<AtomicBool>,
70    destination: Option<(Arc<AtomicUsize>, usize)>,
71}
72
73impl Cancellation {
74    pub fn stop(&self) {
75        self.stopped.store(true, Ordering::Release);
76    }
77    pub fn check(&self) -> Result<()> {
78        if self.stopped.load(Ordering::Acquire)
79            || self
80                .destination
81                .as_ref()
82                .is_some_and(|(current, expected)| current.load(Ordering::Acquire) != *expected)
83        {
84            return Err(Cancelled.into());
85        }
86        Ok(())
87    }
88}
89
90#[derive(Debug, Serialize)]
91#[serde(tag = "kind", rename_all = "snake_case")]
92pub enum Origin {
93    Files {
94        old: PathBuf,
95        new: PathBuf,
96    },
97    Git {
98        repository: PathBuf,
99        base: String,
100        head: String,
101    },
102}
103
104/// Unique within this process, independent of display labels.
105#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
106pub struct ReviewId(u64);
107
108#[derive(Clone, Debug)]
109pub enum SourceState {
110    Pending,
111    Checked(Arc<CheckedPair>),
112    Metadata,
113    Unavailable(Arc<str>),
114}
115
116impl SourceState {
117    pub fn status(&self) -> &'static str {
118        match self {
119            Self::Pending => "Preparing source",
120            Self::Checked(_) => "Source checked",
121            Self::Metadata => "Metadata only",
122            Self::Unavailable(_) => "Source unavailable",
123        }
124    }
125}
126
127enum Provider {
128    Files {
129        old: Arc<Snapshot>,
130        new: Arc<Snapshot>,
131    },
132    Git {
133        git: git::Git,
134        entries: Vec<metadata::Entry>,
135    },
136}
137
138/// One complete local comparison. Source readiness is separate and belongs to each file occurrence.
139pub struct Review {
140    id: ReviewId,
141    doc: storage::Document,
142    origin: Origin,
143    sources: Vec<SourceState>,
144    provider: Provider,
145    storage: SourceStore,
146}
147
148impl Review {
149    pub fn id(&self) -> ReviewId {
150        self.id
151    }
152    pub fn doc(&self) -> &storage::Document {
153        &self.doc
154    }
155    pub fn origin(&self) -> &Origin {
156        &self.origin
157    }
158    pub fn sources(&self) -> &[SourceState] {
159        &self.sources
160    }
161    pub fn open(
162        input: LocalInput,
163        cancel: &Cancellation,
164        mut publish: impl FnMut(Arc<FileIndex>) -> Result<()>,
165    ) -> Result<Self> {
166        let patch_writer = StoreWriter::new()?;
167        let cache = patch_writer.store();
168        let mut doc = storage::Document::default();
169        let (origin, provider) = match input {
170            LocalInput::Files { old, new } => {
171                let (old_source, old_private) = files::capture(&old, &cache, cancel)?;
172                let (new_source, new_private) = files::capture(&new, &cache, cancel)?;
173                let git = git::Git::new(
174                    old_private
175                        .parent()
176                        .context("Private capture has no directory")?,
177                    cancel,
178                )?;
179                let mut args: Vec<&std::ffi::OsStr> =
180                    git::DIFF.iter().map(std::ffi::OsStr::new).collect();
181                args.extend([
182                    "--no-index".as_ref(),
183                    "--no-renames".as_ref(),
184                    "--".as_ref(),
185                    old_private.as_os_str(),
186                    new_private.as_os_str(),
187                ]);
188                let output = git.run(&args, None, true, cancel)?;
189                storage::read_with(
190                    Cancellable {
191                        input: output,
192                        cancel,
193                    },
194                    patch_writer,
195                    &|| cancel.check(),
196                    |mut file, _| {
197                        cancel.check()?;
198                        ensure!(
199                            file.id == 0,
200                            "Two-file comparison returned multiple file entries"
201                        );
202                        let index = Arc::get_mut(&mut file)
203                            .context("File entry was shared before binding")?;
204                        index.meta.old_path =
205                            crate::text::safe_label(old.as_os_str().as_bytes()).into_boxed_str();
206                        index.meta.path =
207                            crate::text::safe_label(new.as_os_str().as_bytes()).into_boxed_str();
208                        index.meta.raw_path = new.as_os_str().as_bytes().into();
209                        doc.append(file.clone())?;
210                        publish(file)
211                    },
212                )?;
213                (
214                    Origin::Files { old, new },
215                    Provider::Files {
216                        old: old_source,
217                        new: new_source,
218                    },
219                )
220            }
221            LocalInput::Git { repo, base, head } => {
222                let git = git::Git::new(&repo, cancel)?;
223                ensure!(
224                    git.small(&["rev-parse", "--show-object-format"], cancel)? == b"sha1\n",
225                    "Only SHA-1 Git repositories are supported"
226                );
227                let directory = git.small(&["rev-parse", "--absolute-git-dir"], cancel)?;
228                let repository = PathBuf::from(std::ffi::OsString::from_vec(
229                    directory
230                        .strip_suffix(b"\n")
231                        .context("Invalid repository identity")?
232                        .to_vec(),
233                ));
234                let base = git.pin(&base, cancel)?.to_string();
235                let head = git.pin(&head, cancel)?.to_string();
236                let mut args: Vec<&std::ffi::OsStr> =
237                    git::DIFF.iter().map(std::ffi::OsStr::new).collect();
238                args.extend(
239                    [
240                        "--raw",
241                        "-z",
242                        "--abbrev=40",
243                        "--no-renames",
244                        "--find-renames=50%",
245                        "-l1000",
246                        &base,
247                        &head,
248                        "--",
249                    ]
250                    .map(std::ffi::OsStr::new),
251                );
252                let output = git.run(&args, None, false, cancel)?;
253                let mut output = BufReader::new(output);
254                let entries = metadata::envelope(&mut output, cancel)?;
255                storage::read_with(
256                    Cancellable {
257                        input: output,
258                        cancel,
259                    },
260                    patch_writer,
261                    &|| cancel.check(),
262                    |mut file, identity| {
263                        cancel.check()?;
264                        entries
265                            .get(file.id)
266                            .context("Patch has more entries than its Git inventory")?
267                            .matches(&file, &identity)?;
268                        let entry = &entries[file.id];
269                        let index = Arc::get_mut(&mut file)
270                            .context("File entry was shared before binding")?;
271                        index.meta.old_mode = format!("{:o}", entry.old.mode).into_boxed_str();
272                        index.meta.new_mode = format!("{:o}", entry.new.mode).into_boxed_str();
273                        doc.append(file.clone())?;
274                        publish(file)
275                    },
276                )?;
277                ensure!(
278                    doc.files.len() == entries.len(),
279                    "Git inventory has entries missing from its patch"
280                );
281                (
282                    Origin::Git {
283                        repository,
284                        base,
285                        head,
286                    },
287                    Provider::Git { git, entries },
288                )
289            }
290        };
291        let sources = doc
292            .files
293            .iter()
294            .map(|file| {
295                if file.status.as_ref() == "Binary change"
296                    || file.old_mode.as_ref() == "160000"
297                    || file.new_mode.as_ref() == "160000"
298                {
299                    SourceState::Metadata
300                } else {
301                    SourceState::Pending
302                }
303            })
304            .collect();
305        Ok(Self {
306            id: ReviewId(cache.identity()),
307            doc,
308            origin,
309            sources,
310            provider,
311            storage: SourceStore::sharing(&cache)?,
312        })
313    }
314
315    pub fn prepare(&mut self, id: usize, cancel: &Cancellation) -> Result<SourceState> {
316        cancel.check()?;
317        let state = self
318            .sources
319            .get(id)
320            .context("Source request is outside this review")?;
321        if !matches!(state, SourceState::Pending) {
322            return Ok(state.clone());
323        }
324        let checkpoint = self.storage.checkpoint();
325        let result = self.prepare_pair(id, cancel);
326        let state = match result {
327            Ok(pair) => SourceState::Checked(Arc::new(pair)),
328            Err(error) => {
329                self.storage.rollback(checkpoint)?;
330                if error.is::<Cancelled>() || error.is::<crate::source_spans::ReadFailure>() {
331                    return Err(error);
332                }
333                SourceState::Unavailable(
334                    crate::text::safe_text(format!("{error:#}").as_bytes()).into(),
335                )
336            }
337        };
338        self.sources[id] = state.clone();
339        Ok(state)
340    }
341
342    fn prepare_pair(&mut self, id: usize, cancel: &Cancellation) -> Result<CheckedPair> {
343        cancel.check()?;
344        let (old, new) = match &self.provider {
345            Provider::Files { old, new } => (Some(old.clone()), Some(new.clone())),
346            Provider::Git { git, entries } => {
347                let entry = &entries[id];
348                let ids: Vec<_> = [&entry.old, &entry.new]
349                    .iter()
350                    .filter(|side| side.mode != 0)
351                    .map(|side| side.oid)
352                    .collect();
353                let mut output = BufReader::new(git.objects(&ids, cancel)?);
354                let mut capture = |side: &metadata::Endpoint| -> Result<Option<Arc<Snapshot>>> {
355                    if side.mode == 0 {
356                        return Ok(None);
357                    }
358                    let header = metadata::field(&mut output, b'\n')?;
359                    let mut fields = header.split(|byte| *byte == b' ');
360                    ensure!(
361                        git::oid(fields.next().context("Missing object ID")?)? == side.oid,
362                        "Git delivered the wrong source object"
363                    );
364                    ensure!(
365                        fields.next() == Some(b"blob"),
366                        "Git source object is not a blob"
367                    );
368                    let length: u64 =
369                        std::str::from_utf8(fields.next().context("Missing object length")?)?
370                            .parse()?;
371                    ensure!(fields.next().is_none(), "Invalid object header");
372                    let source = self
373                        .storage
374                        .capture(output.by_ref().take(length), &|| cancel.check())?;
375                    ensure!(source.len() == length, "Truncated source object");
376                    let mut newline = [0];
377                    output.read_exact(&mut newline)?;
378                    ensure!(newline == [b'\n'], "Invalid source object terminator");
379                    Ok(Some(source))
380                };
381                let old = capture(&entry.old)?;
382                let new = capture(&entry.new)?;
383                ensure!(
384                    output.fill_buf()?.is_empty(),
385                    "Unexpected data after source objects"
386                );
387                (old, new)
388            }
389        };
390        check_pair(self.doc.files[id].clone(), old, new, &|| cancel.check())
391    }
392
393    pub fn report(&mut self, cancel: &Cancellation) -> Result<serde_json::Value> {
394        let mut files = Vec::new();
395        for id in 0..self.doc.files.len() {
396            let state = self.prepare(id, cancel)?;
397            let sides = if let SourceState::Checked(pair) = &state {
398                let describe = |side: Option<&Arc<Snapshot>>| {
399                    side.map(|s| serde_json::json!({ "bytes": s.len(), "lines": s.lines(), "final_newline": s.final_newline() }))
400                };
401                serde_json::json!({ "old": describe(pair.old()), "new": describe(pair.new_source()) })
402            } else {
403                serde_json::Value::Null
404            };
405            let provenance = match &self.provider {
406                Provider::Git { entries, .. } => {
407                    let endpoint = |side: &metadata::Endpoint| serde_json::json!({"path_bytes": side.path, "mode": format!("{:o}", side.mode), "object": (!side.oid.is_zero()).then(|| side.oid.to_string())});
408                    serde_json::json!({"old": endpoint(&entries[id].old), "new": endpoint(&entries[id].new)})
409                }
410                Provider::Files { .. } => serde_json::Value::Null,
411            };
412            files.push(serde_json::json!({ "file": id, "path": self.doc.files[id].path, "source": state.status(), "error": if let SourceState::Unavailable(reason) = &state { Some(reason.as_ref()) } else { None }, "sides": sides, "provenance": provenance }));
413        }
414        let source_storage_bytes = match &self.provider {
415            Provider::Files { old, new } => [old, new]
416                .iter()
417                .map(|s| s.len() + (u64::from(s.lines()) + 1) * 8)
418                .sum(),
419            Provider::Git { .. } => self.storage.bytes(),
420        };
421        Ok(
422            serde_json::json!({"review_id": self.id, "comparison": self.origin, "completeness": "complete", "files": files, "source_storage_bytes": source_storage_bytes}),
423        )
424    }
425}
426
427struct Cancellable<'a, R> {
428    input: R,
429    cancel: &'a Cancellation,
430}
431impl<R: Read> Read for Cancellable<'_, R> {
432    fn read(&mut self, bytes: &mut [u8]) -> std::io::Result<usize> {
433        self.cancel.check().map_err(std::io::Error::other)?;
434        self.input.read(bytes)
435    }
436}
diffs_rs/input.rs

diffs_rs/
input.rs
