fix(tail): gate checkpoint by_inode fallback on fingerprint to stop silent skip after copytruncate

src/tail/checkpoint/store.zig

@@ -59,7 +59,8 @@ pub const Store = struct {
             if (!checkpoint_types.isExpired(value, self.ttl_ns, now)) return value.offset;
         }
         if (self.by_inode.get(keys.inode)) |value| {
-            if (!checkpoint_types.isExpired(value, self.ttl_ns, now)) return value.offset;
+            if (value.identity.fingerprint == identity.fingerprint and
+                !checkpoint_types.isExpired(value, self.ttl_ns, now)) return value.offset;
         }
         return null;
     }
@@ -139,3 +140,52 @@ pub const Store = struct {
         }
     }
 };
+
+const testing = std.testing;
+
+fn freshStore(ttl_ns: i128) Store {
+    return Store.init(testing.allocator, testing.io, 256, ttl_ns);
+}
+
+test "store: by_inode fallback returns null when stored fingerprint differs" {
+    var store = freshStore(72 * 60 * 60 * std.time.ns_per_s);
+    defer store.deinit();
+
+    const id_a: tail_types.FileIdentity = .{ .dev = 1, .inode = 2, .fingerprint = 100 };
+    const id_b: tail_types.FileIdentity = .{ .dev = 1, .inode = 2, .fingerprint = 999 };
+    const now = std.Io.Timestamp.now(testing.io, .awake).toNanoseconds();
+    try store.upsert(.{ .identity = id_a, .offset = 4096, .last_seen_ns = @intCast(now) });
+
+    // by_identity misses (different fingerprint key).
+    try testing.expect(store.by_identity.get(checkpoint_types.keysFor(id_b).identity) == null);
+    // by_inode gate must reject the cross-version offset.
+    try testing.expect(store.getOffset(id_b) == null);
+}
+
+test "store: by_inode fallback returns offset when fingerprint matches and by_identity misses" {
+    var store = freshStore(72 * 60 * 60 * std.time.ns_per_s);
+    defer store.deinit();
+
+    // Two entries sharing the same (dev, inode) but different fingerprints.
+    const id_a: tail_types.FileIdentity = .{ .dev = 1, .inode = 5, .fingerprint = 10 };
+    const id_b: tail_types.FileIdentity = .{ .dev = 1, .inode = 5, .fingerprint = 20 };
+    const now = std.Io.Timestamp.now(testing.io, .awake).toNanoseconds();
+    try store.upsert(.{ .identity = id_a, .offset = 111, .last_seen_ns = @intCast(now) });
+    try store.upsert(.{ .identity = id_b, .offset = 222, .last_seen_ns = @intCast(now) });
+
+    // by_identity hits directly for each fingerprint.
+    try testing.expectEqual(@as(?u64, 111), store.getOffset(id_a));
+    try testing.expectEqual(@as(?u64, 222), store.getOffset(id_b));
+}
+
+test "store: by_inode fallback returns null for expired entry with matching fingerprint" {
+    const ttl_ns: i128 = 1 * std.time.ns_per_s;
+    var store = freshStore(ttl_ns);
+    defer store.deinit();
+
+    const id: tail_types.FileIdentity = .{ .dev = 1, .inode = 2, .fingerprint = 100 };
+    // last seen far in the past so it is expired relative to ttl.
+    try store.upsert(.{ .identity = id, .offset = 4096, .last_seen_ns = -1_000_000_000 });
+
+    try testing.expect(store.getOffset(id) == null);
+}

src/tail/watch.zig

@@ -466,6 +466,7 @@ pub const Watcher = struct {
             prefix_len = @min(@as(u64, 64), size);
             self.head_prefix_lens.items[i] = @intCast(prefix_len);
             self.head_prefix_hashes.items[i] = try prefixHash(self.io, file, prefix_len);
+            try self.refreshIdentityFingerprint(idx, file);
             return;
         }
 
@@ -474,14 +475,42 @@ pub const Watcher = struct {
             const new_len: u64 = @min(@as(u64, 64), size);
             self.head_prefix_lens.items[i] = @intCast(new_len);
             self.head_prefix_hashes.items[i] = try prefixHash(self.io, file, new_len);
+            try self.refreshIdentityFingerprint(idx, file);
             return;
         }
 
         const observed = try prefixHash(self.io, file, prefix_len);
-        if (observed == self.head_prefix_hashes.items[i]) return;
+        if (observed == self.head_prefix_hashes.items[i]) {
+            // The prefix is unchanged, but the identity fingerprint may have been
+            // computed on a shorter file (e.g. after a partially written
+            // copytruncate).  Refresh it so that ongoing checkpoints and a
+            // future checkpoint-based restart both use a fingerprint that covers
+            // the current file content rather than the partial prefix.
+            try self.refreshIdentityFingerprint(idx, file);
+            return;
+        }
 
         if (self.offsets.items[i] > 0) self.offsets.items[i] = 0;
         self.head_prefix_hashes.items[i] = observed;
+        try self.refreshIdentityFingerprint(idx, file);
+    }
+
+    /// Recompute the CRC32 fingerprint from the live file after a same-inode
+    /// content rewrite is detected (copytruncate / in-place rewrite). The
+    /// watcher pins `identities.items[i].fingerprint` at open time and never
+    /// refreshes it for the same inode otherwise; without this refresh, every
+    /// post-rotation enqueue is filed under the stale fingerprint, so a
+    /// checkpoint-based restart misses `by_identity` and (without the
+    /// store-level gate) silently resumes from a stale cross-version offset.
+    fn refreshIdentityFingerprint(self: *Watcher, idx: u32, file: std.Io.File) !void {
+        const i: usize = @intCast(idx);
+        if (self.identities.items[i]) |id| {
+            self.identities.items[i] = .{
+                .dev = id.dev,
+                .inode = id.inode,
+                .fingerprint = try computeFingerprint(self.io, file),
+            };
+        }
     }
 
     fn maybeSwitchPending(self: *Watcher, idx: u32) !void {
@@ -765,3 +794,164 @@ test "watch public API: collect emits appended file bytes" {
     try w.collect(&events, .tail, null);
     try testing.expectEqual(@as(usize, 1), events.items.len);
 }
+
+test "checkpoint resume after copytruncate emits full new content (no silent skip)" {
+    const io = testing.io;
+    var tmp = testing.tmpDir(.{});
+    defer tmp.cleanup();
+
+    // Phase 1: create file with content A, record identity (FP_A).
+    {
+        const f = try tmp.dir.createFile(io, "tail.log", .{});
+        defer f.close(io);
+        var buf: [8192]u8 = undefined;
+        @memset(buf[0..], 'A');
+        try f.writeStreamingAll(io, buf[0..]);
+    }
+    const abs = try tmp.dir.realPathFileAlloc(io, "tail.log", testing.allocator);
+    defer testing.allocator.free(abs);
+
+    const id_a: types.FileIdentity = blk: {
+        const f = try std.Io.Dir.cwd().openFile(io, abs, .{ .mode = .read_only });
+        defer f.close(io);
+        const st = try fstatHandle(f.handle);
+        break :blk .{ .dev = st.dev, .inode = st.ino, .fingerprint = try computeFingerprint(io, f) };
+    };
+
+    // Phase 2: durably checkpoint offset N under FP_A via the live lane worker.
+    const N: u64 = 4096;
+    const state_dir = try tmp.dir.realPathFileAlloc(io, ".", testing.allocator);
+    defer testing.allocator.free(state_dir);
+
+    const lifecycle_mod = @import("../core/lifecycle.zig");
+    {
+        var lifecycle: lifecycle_mod.Lifecycle = .init;
+        var lane = try checkpoint_mod.Lane.init(
+            testing.allocator,
+            io,
+            state_dir,
+            16,
+            64,
+            5,
+            72 * 60 * 60 * 1000,
+            64,
+            60_000,
+        );
+        try lane.start(&lifecycle);
+        _ = try lane.enqueue(.{
+            .identity = id_a,
+            .byte_offset = N,
+            .last_seen_size = N,
+            .last_seen_ns = @intCast(std.Io.Timestamp.now(io, .awake).toNanoseconds()),
+        });
+        var tries: usize = 0;
+        while (tries < 200 and lane.getOffset(id_a) != N) : (tries += 1) {
+            try io.sleep(.fromNanoseconds(2 * std.time.ns_per_ms), .awake);
+        }
+        try testing.expectEqual(N, lane.getOffset(id_a).?);
+        lifecycle.requestShutdown(io);
+        lifecycle.shutdown(io);
+        lane.finalize();
+        lane.deinit();
+    }
+
+    // Phase 3: copytruncate — replace content A with content B (FP_B != FP_A, size >= N).
+    {
+        const f = try std.Io.Dir.cwd().openFile(io, abs, .{ .mode = .read_write });
+        defer f.close(io);
+        try f.setLength(io, 0);
+        var buf: [8192]u8 = undefined;
+        @memset(buf[0..], 'B');
+        try f.writePositionalAll(io, buf[0..], 0);
+        try f.sync(io);
+    }
+
+    // Phase 4: recover lane from durable state (simulates restart).
+    var recovered = try checkpoint_mod.Lane.init(
+        testing.allocator,
+        io,
+        state_dir,
+        16,
+        64,
+        5,
+        72 * 60 * 60 * 1000,
+        64,
+        60_000,
+    );
+    defer recovered.deinit();
+
+    // Phase 5: checkpoint-based restart. The by_inode gate (Change 1) rejects the
+    // stale FP_A offset, so the watcher reads new content from offset 0.
+    var w = try Watcher.init(testing.allocator, io, .poll, &.{abs}, "-", .checkpoint, 1000, 50, 1000);
+    defer w.deinit();
+    w.applyCheckpointLane(&recovered);
+
+    var events: std.ArrayList(Event) = .empty;
+    defer events.deinit(testing.allocator);
+    try w.collect(&events, .checkpoint, &recovered);
+
+    // FIX: full new content [0, 8192) is emitted — NOT the stale [N, 8192).
+    try testing.expectEqual(@as(usize, 1), events.items.len);
+    try testing.expectEqual(@as(u64, 0), events.items[0].start_offset);
+    try testing.expectEqual(@as(u64, 8192), events.items[0].end_offset);
+}
+
+test "live copytruncate refreshes pinned fingerprint for post-rotation enqueues" {
+    const io = testing.io;
+    var tmp = testing.tmpDir(.{});
+    defer tmp.cleanup();
+
+    // Content A: 200 bytes of 'A' (larger than the 64-byte rewrite-prefix window).
+    {
+        const f = try tmp.dir.createFile(io, "tail.log", .{});
+        defer f.close(io);
+        var buf: [200]u8 = undefined;
+        @memset(buf[0..], 'A');
+        try f.writeStreamingAll(io, buf[0..]);
+    }
+    const abs = try tmp.dir.realPathFileAlloc(io, "tail.log", testing.allocator);
+    defer testing.allocator.free(abs);
+
+    var w = try Watcher.init(testing.allocator, io, .poll, &.{abs}, "-", .head, 1000, 50, 1000);
+    defer w.deinit();
+
+    var events: std.ArrayList(Event) = .empty;
+    defer events.deinit(testing.allocator);
+
+    // First collect: emits content A. Capture fingerprint FP_A.
+    try w.collect(&events, .head, null);
+    try testing.expectEqual(@as(usize, 1), events.items.len);
+    try testing.expectEqual(@as(u64, 200), events.items[0].end_offset);
+    try testing.expectEqual(@as(u64, 0), events.items[0].start_offset);
+    const fp_a = events.items[0].identity.?.fingerprint;
+    // FP_A is Crc32 of the first 200 bytes (all 'A').
+    var expected_a: [200]u8 = undefined;
+    @memset(expected_a[0..], 'A');
+    try testing.expectEqual(std.hash.Crc32.hash(expected_a[0..]), fp_a);
+
+    // Copytruncate: replace with 8192 bytes of 'B' (different size → poll marks dirty).
+    {
+        const f = try std.Io.Dir.cwd().openFile(io, abs, .{ .mode = .read_write });
+        defer f.close(io);
+        try f.setLength(io, 0);
+        var buf: [8192]u8 = undefined;
+        @memset(buf[0..], 'B');
+        try f.writePositionalAll(io, buf[0..], 0);
+        try f.sync(io);
+    }
+
+    // Second collect: detects rewrite, resets offset to 0, refreshes fingerprint
+    // (Change 2), and emits full new content [0, 8192) carrying the refreshed FP_B.
+    try w.collect(&events, .head, null);
+    try testing.expectEqual(@as(usize, 1), events.items.len);
+    try testing.expectEqual(@as(u64, 0), events.items[0].start_offset);
+    try testing.expectEqual(@as(u64, 8192), events.items[0].end_offset);
+    const fp_b = events.items[0].identity.?.fingerprint;
+    // FP_B is Crc32 of the first 1024 bytes (all 'B').
+    var expected_b: [1024]u8 = undefined;
+    @memset(expected_b[0..], 'B');
+    try testing.expectEqual(std.hash.Crc32.hash(expected_b[0..]), fp_b);
+
+    // The fingerprint must have been refreshed to reflect the new content.
+    try testing.expect(fp_a != fp_b);
+}