Detect the rename by file identity, not by the final name existing

A file already sitting at the final name -- an earlier capture reusing
the name, or a previous copy -- counted as the rename the moment tailing
started, so the first writer lull longer than the grace period finalized
the transfer short. Treat the rename as observed only once the final path
is the same inode as the fd being tailed, which also still works when the
writer renames over such a leftover.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

src/tail.rs

@@ -2,7 +2,7 @@ use crate::chunker::{Chunker, ChunkKind};
 use crate::tracelog::Logger;
 use anyhow::{bail, Result};
 use std::fs::File;
-use std::os::unix::fs::FileExt;
+use std::os::unix::fs::{FileExt, MetadataExt};
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::Arc;
@@ -219,7 +219,7 @@ fn drain(
             send_chunk(events, kind, data)?;
         }
 
-        if !complete_signal && final_path.try_exists()? {
+        if !complete_signal && renamed_to(final_path, &file)? {
             // Rename observed -- signal now, well before finalization, per
             // section 7.4's repoint-then-drain-then-finalize ordering.
             complete_signal = true;
@@ -281,6 +281,22 @@ fn drain(
     }
 }
 
+/// Whether `final_path` now names the very file being read, i.e. the writer's
+/// rename has happened. This is stricter than section 7.5's "the final path
+/// exists": names get reused (section 6), so a file left at the final name by
+/// an earlier capture or copy would otherwise pass for the rename the moment
+/// tailing starts, and the transfer would finish early, short, at the first
+/// lull in the writer. Comparing identity also still works when the writer
+/// renames over such a leftover.
+fn renamed_to(final_path: &Path, file: &File) -> Result<bool> {
+    let ours = file.metadata()?;
+    match std::fs::metadata(final_path) {
+        Ok(theirs) => Ok(theirs.dev() == ours.dev() && theirs.ino() == ours.ino()),
+        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false),
+        Err(e) => Err(e.into()),
+    }
+}
+
 fn emit_ready(chunker: &mut dyn Chunker, events: &impl EventSink) -> Result<()> {
     while let Some((kind, data)) = chunker.ready_chunk() {
         send_chunk(events, kind, data)?;
@@ -731,4 +747,54 @@ mod tests {
         try_collect(rx);
         assert!(result.is_err());
     }
+
+    #[test]
+    fn a_stale_file_at_the_final_name_is_not_the_rename() {
+        // A leftover cap.pcap (an earlier capture reusing the name) used to
+        // pass for the rename the moment tailing started, so the first lull
+        // longer than the grace period finalized the copy short.
+        let dir = tempdir().unwrap();
+        let partial_path = dir.path().join("cap.pcap.partial");
+        let final_path = dir.path().join("cap.pcap");
+        std::fs::write(&final_path, b"yesterday's capture").unwrap();
+
+        let mut stream = header_bytes(0xA1B2C3D4);
+        stream.extend_from_slice(&record_bytes(&[1u8; 10], 1_700_000_000));
+        stream.extend_from_slice(&record_bytes(&[2u8; 10], 1_700_000_001));
+        let mut f = File::create(&partial_path).unwrap();
+        f.write_all(&stream).unwrap();
+        f.sync_all().unwrap();
+
+        let (tx, rx) = mpsc::channel();
+        let handle = std::thread::spawn({
+            let partial_path = partial_path.clone();
+            let final_path = final_path.clone();
+            move || {
+                run_tail(
+                    partial_path,
+                    true,
+                    final_path,
+                    Box::new(RecordAlignedChunker::new(DEFAULT_TARGET_BYTES, DEFAULT_FLUSH, true)),
+                    tx,
+                    Arc::new(Logger::none()),
+                    no_stop(),
+                )
+            }
+        });
+
+        // A writer lull longer than the grace period, then more packets and
+        // the real rename -- over the stale file.
+        std::thread::sleep(FINALIZE_GRACE_PERIOD + Duration::from_millis(500));
+        let rec3 = record_bytes(&[3u8; 10], 1_700_000_002);
+        f.write_all(&rec3).unwrap();
+        f.sync_all().unwrap();
+        stream.extend_from_slice(&rec3);
+        drop(f);
+        std::fs::rename(&partial_path, &final_path).unwrap();
+
+        handle.join().unwrap().unwrap();
+        let (data, eof) = collect(rx);
+        assert_eq!(data, stream);
+        assert!(eof);
+    }
 }