diff --git a/src/tail/checkpoint/store.zig b/src/tail/checkpoint/store.zig index 114f563a..4726444d 100644 --- a/src/tail/checkpoint/store.zig +++ b/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); +} diff --git a/src/tail/watch.zig b/src/tail/watch.zig index b9389e11..fd6f94bd 100644 --- a/src/tail/watch.zig +++ b/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); +}