diff --git a/src/internal/event_log.zig b/src/internal/event_log.zig index ad70701..1a388c2 100644 --- a/src/internal/event_log.zig +++ b/src/internal/event_log.zig @@ -972,18 +972,22 @@ pub const DiskPersist = struct { self.mutex.lockUncancelable(self.io); defer self.mutex.unlock(self.io); + if (self.evtbuf.items.len >= default_flush_threshold) try self.flushLocked(); + try self.evtbuf.ensureUnusedCapacity(self.allocator, 1); + try self.outbuf.ensureUnusedCapacity(self.allocator, data.len); + // assign seq const seq = self.cur_seq; self.cur_seq += 1; std.mem.writeInt(u64, data[20..28], seq, .little); - try self.evtbuf.append(self.allocator, .{ .data = data, .seq = seq }); - try self.outbuf.appendSlice(self.allocator, data); + self.evtbuf.appendAssumeCapacity(.{ .data = data, .seq = seq }); + self.outbuf.appendSliceAssumeCapacity(data); self.event_counter += 1; // flush if threshold hit if (self.evtbuf.items.len >= default_flush_threshold) { - try self.flushLocked(); + self.flushLocked() catch {}; } return seq; @@ -1341,12 +1345,13 @@ pub const DiskPersist = struct { var name_buf: [64]u8 = undefined; const name = std.fmt.bufPrint(&name_buf, "evts-{d}", .{start_seq}) catch unreachable; - if (self.current_file) |f| f.close(self.io); - if (self.current_file_path) |p| self.allocator.free(p); - - self.current_file = try self.dir.createFile(self.io, name, .{ .truncate = false, .read = true }); - self.current_file_pos = 0; - self.current_file_path = try self.allocator.dupe(u8, name); + const path = try self.allocator.dupe(u8, name); + errdefer self.allocator.free(path); + const file = try self.dir.createFile(self.io, name, .{ .exclusive = true, .read = true }); + errdefer { + file.close(self.io); + self.dir.deleteFile(self.io, name) catch {}; + } // register in Postgres _ = try self.db.exec( @@ -1354,6 +1359,11 @@ pub const DiskPersist = struct { .{ name, @as(i64, @intCast(start_seq)) }, ); + if (self.current_file) |f| f.close(self.io); + if (self.current_file_path) |p| self.allocator.free(p); + self.current_file = file; + self.current_file_pos = 0; + self.current_file_path = path; self.event_counter = 0; } @@ -1806,6 +1816,91 @@ test "persist and playback" { try std.testing.expectEqualStrings("payload-three", entries.items[2].data); } +test "persist allocation failures preserve sequence and buffered records" { + const Case = struct { + fn run(allocator: Allocator) !void { + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + const file = try tmp.dir.createFile(std.testing.io, "events", .{ .read = true }); + defer file.close(std.testing.io); + var dp: DiskPersist = .{ + .allocator = allocator, + .dir_path = "", + .dir = tmp.dir, + .db = undefined, + .current_file = file, + .did_cache = undefined, + .host_id_cache = undefined, + .io = std.testing.io, + }; + defer { + for (dp.evtbuf.items) |job| allocator.free(job.data); + dp.evtbuf.deinit(allocator); + dp.outbuf.deinit(allocator); + } + const seq = dp.persist(.commit, 1, "payload") catch |err| { + try std.testing.expectEqual(@as(u64, 1), dp.cur_seq); + try std.testing.expectEqual(@as(usize, 0), dp.evtbuf.items.len); + try std.testing.expectEqual(@as(usize, 0), dp.outbuf.items.len); + return err; + }; + try std.testing.expectEqual(@as(u64, 1), seq); + try dp.flushLocked(); + const scan = try scanForLastSeq(file, std.testing.io, 1); + try std.testing.expectEqual(@as(?u64, 1), scan.last_seq); + } + }; + try std.testing.checkAllAllocationFailures(std.testing.allocator, Case.run, .{}); +} + +test "failed rotation preserves the active file and retries after database recovery" { + var tdb = try TestDb.create(try requireDatabaseUrl()); + defer tdb.destroy(); + var tmp = std.testing.tmpDir(.{}); + defer tmp.cleanup(); + const dir_path = try tmpDirRealPath(std.testing.allocator, tmp); + defer std.testing.allocator.free(dir_path); + var dp = try DiskPersist.init(std.testing.allocator, dir_path, tdb.url, 2, std.testing.io); + defer dp.deinit(); + + _ = try dp.persist(.commit, 1, "before"); + try dp.flushLocked(); + const old_pos = dp.current_file_pos; + { + var failing = std.testing.FailingAllocator.init(std.testing.allocator, .{ .fail_index = 0 }); + dp.allocator = failing.allocator(); + defer dp.allocator = std.testing.allocator; + try std.testing.expectError(error.OutOfMemory, dp.createLogFile(2)); + } + try std.testing.expectEqualStrings("evts-1", dp.current_file_path.?); + try tmp.dir.createDir(std.testing.io, "evts-2", .default_dir); + if (dp.createLogFile(2)) |_| return error.ExpectedRotationFailure else |_| {} + try std.testing.expectEqualStrings("evts-1", dp.current_file_path.?); + try std.testing.expectEqual(old_pos, dp.current_file_pos); + try tmp.dir.deleteDir(std.testing.io, "evts-2"); + + _ = try dp.db.exec("ALTER TABLE log_file_refs ADD CONSTRAINT reject_rotation CHECK (seq_start < 2)", .{}); + if (dp.createLogFile(2)) |_| return error.ExpectedRotationFailure else |_| {} + try std.testing.expectEqualStrings("evts-1", dp.current_file_path.?); + try std.testing.expectError(error.FileNotFound, tmp.dir.openFile(std.testing.io, "evts-2", .{})); + _ = try dp.persist(.commit, 1, "after"); + try dp.flushLocked(); + const scan = try scanForLastSeq(dp.current_file.?, std.testing.io, 1); + try std.testing.expectEqual(@as(?u64, 2), scan.last_seq); + + _ = try dp.db.exec("ALTER TABLE log_file_refs DROP CONSTRAINT reject_rotation", .{}); + try dp.createLogFile(3); + _ = try dp.persist(.commit, 1, "rotated"); + try dp.flushLocked(); + var entries: std.ArrayListUnmanaged(PlaybackEntry) = .empty; + defer { + for (entries.items) |entry| std.testing.allocator.free(entry.data); + entries.deinit(std.testing.allocator); + } + try dp.playback(0, std.testing.allocator, &entries, .{}); + try std.testing.expectEqual(@as(usize, 3), entries.items.len); +} + test "gc deletes expired segments and their refs, keeps the current file" { // written 2026-08-27, when production surfaced 10 weeks of never-pruned // segments (132G/150G): gc() had no test at all. this exercises the whole