diff --git a/src/commands.zig b/src/commands.zig
index 7a3bb36..8c5eaf2 100644
--- a/src/commands.zig
+++ b/src/commands.zig
@@ -188,6 +188,18 @@ pub fn dispatch(ctx: *Context, msg: *wire.Message, reply: *wire.Reply) !void {
         // The write is durable by now, so a compaction failure is a maintenance
         // problem and not the client's. Report it and hand the request back
         // rather than turning an applied write into an error the client retries.
+        // A checkpoint reclaims the log, so an open stops paying for every write
+        // ever made. Runs here, with no lock held, for the same reason
+        // compaction does: it takes the log lock and must not do that while
+        // holding a collection lock.
+        if (ctx.engine.take_checkpoint()) {
+            ctx.engine.checkpoint() catch |err| {
+                // Durability is unaffected -- the log still holds everything.
+                // The cost is a slower next open, which is not the client's
+                // problem, so report and carry on.
+                std.debug.print("multiforadb: checkpoint failed: {s}\n", .{@errorName(err)});
+            };
+        }
         if (ctx.engine.take_compact()) {
             ctx.engine.compact() catch |err| {
                 std.debug.print("multiforadb: compaction failed: {s}\n", .{@errorName(err)});
diff --git a/src/db.zig b/src/db.zig
index 5968bde..765d697 100644
--- a/src/db.zig
+++ b/src/db.zig
@@ -231,6 +231,14 @@ pub const Engine = struct {
     /// rewrite is worth doing — see `note_compact`.
     live_docs: u64 = 0,
     dead_docs: u64 = 0,
+    /// Set when the log has grown enough since the last checkpoint to be worth
+    /// reclaiming. Read by the write epilogue and the TTL monitor, both of which
+    /// run without holding a collection lock.
+    checkpoint_pending: std.atomic.Value(bool) = std.atomic.Value(bool).init(false),
+    /// Log bytes that trigger a checkpoint. Distinct from the compaction
+    /// threshold: compaction is about the *garbage share* of the data, a
+    /// checkpoint is about how much replay an open would otherwise have to do.
+    checkpoint_threshold: u64 = 32 * 1024 * 1024,
     /// Whether replay must maintain index entries as it goes.
     ///
     /// A full replay does not: it puts documents in place and lets
@@ -729,6 +737,7 @@ pub const Engine = struct {
         // is dead. Leaving it promised would grow the file on every write.
         self.pager.release_reservation();
         self.note_compact();
+        self.note_checkpoint();
     }
 
     /// Remove a document by its `_id` value. Returns true if it existed.
@@ -1057,6 +1066,18 @@ pub const Engine = struct {
     /// wanted. It runs in the command epilogue, after the collection lock is
     /// released — never inline, since compact takes the collection locks
     /// itself and would deadlock against the caller's.
+    /// Arm a checkpoint when the log has grown past the threshold. Cheap enough
+    /// to call on every write: one relaxed load and a compare.
+    fn note_checkpoint(self: *Engine) void {
+        if (self.log.log_bytes < self.checkpoint_threshold) return;
+        self.checkpoint_pending.store(true, .release);
+    }
+
+    /// Claim a pending checkpoint, if there is one.
+    pub fn take_checkpoint(self: *Engine) bool {
+        return self.checkpoint_pending.swap(false, .acq_rel);
+    }
+
     fn note_compact(self: *Engine) void {
         // The threshold counts data volume (uncompressed record bytes), not
         // the on-disk size: a compressed log would otherwise stay under any
@@ -1597,6 +1618,17 @@ pub const Engine = struct {
                 return err;
             };
             self.pager.release_reservation();
+            // The watermark is durable, so every record it covers is now
+            // redundant. Strictly after the publish: the other order loses data
+            // if a crash lands between them.
+            self.log.truncate_to_header() catch |err| {
+                // A failed truncation wastes space and costs replay time on the
+                // next open; it does not lose anything, because the records are
+                // still there and still above no watermark. Not worth failing
+                // the checkpoint that already succeeded.
+                std.debug.print("multiforadb: WARNING: log truncation failed: {s}\n", .{@errorName(err)});
+            };
+            self.committed_seq = snapshot_seq;
             self.log_lock.unlock(self.io);
             return;
         }
@@ -2508,6 +2540,85 @@ test "a checkpoint lets the next open skip the log it covers" {
     }
 }
 
+test "a checkpoint reclaims the log and the data survives" {
+    // The payoff of a lagging checkpoint: once the data file holds the effect of
+    // a record, the record is redundant and the log can be reclaimed. Without
+    // this the log only ever grows and every open pays for every write ever made.
+    //
+    // Mutation check, red: skipping the truncation.
+    //
+    // Not covered: moving the truncation *before* the publish. That is still
+    // correct in the absence of a crash -- the publish follows immediately -- and
+    // the hazard is precisely a crash landing between the two, with the records
+    // gone from the log and not yet in any image. Catching it needs process-level
+    // crash injection, which the milestone's gates cover; an in-process test
+    // cannot express "stop here and die". The order stays because it is the
+    // whole reason a lagging checkpoint is safe.
+    var threaded: std.Io.Threaded = .init_single_threaded;
+    defer threaded.deinit();
+    var env = test_env(&threaded);
+    const io = env.io;
+    const gpa = testing.allocator;
+
+    var tmp = try TmpLog.init(gpa);
+    defer tmp.deinit(gpa);
+    const data_path = try std.fmt.allocPrint(gpa, "{s}.data", .{tmp.path});
+    defer gpa.free(data_path);
+    defer std.Io.Dir.cwd().deleteFile(io, data_path) catch {};
+
+    var log_after_checkpoint: u64 = 0;
+    {
+        var engine = try Engine.open(gpa, io, tmp.path);
+        defer engine.deinit();
+        try engine.lock();
+        var i: i32 = 0;
+        while (i < 200) : (i += 1) {
+            var d = try make_user(gpa, i, "a@x.io");
+            defer d.deinit();
+            try engine.insert("app", "users", &d, &env.gen);
+        }
+        engine.unlock();
+
+        // Commit first, so the records are actually on disk: appends buffer in
+        // the log's open block, and only a commit seals and writes it. Measuring
+        // before that reads a file that is still just its header.
+        try engine.commit();
+        const before = try engine.log.file.length(io);
+        try testing.expect(before > storage.file_header_len);
+
+        try engine.checkpoint();
+        log_after_checkpoint = try engine.log.file.length(io);
+        // The log is back to just its header.
+        try testing.expect(log_after_checkpoint < before);
+        try testing.expectEqual(@as(u64, storage.file_header_len), log_after_checkpoint);
+        // And writing still works afterwards, at a sequence above the watermark.
+        try engine.lock();
+        var d = try make_user(gpa, 999, "z@x.io");
+        defer d.deinit();
+        try engine.insert("app", "users", &d, &env.gen);
+        engine.unlock();
+        try testing.expect(engine.seq > engine.pager.loaded.seq);
+    }
+
+    // Everything is still there after a reopen: 200 from the image, 1 from the
+    // log records written after the truncation.
+    {
+        var engine = try Engine.open(gpa, io, tmp.path);
+        defer engine.deinit();
+        try engine.lock();
+        defer engine.unlock();
+        const coll = engine.get_collection("app", "users").?;
+        try testing.expectEqual(@as(usize, 201), coll.docs.count());
+        try testing.expectEqual(@as(usize, 201), coll.id_index.count());
+        const id_key = try bson.serialize_value(gpa, .{ .int32 = 999 });
+        defer gpa.free(id_key);
+        try testing.expect(engine.get_doc("app", "users", id_key) != null);
+        const first_key = try bson.serialize_value(gpa, .{ .int32 = 0 });
+        defer gpa.free(first_key);
+        try testing.expect(engine.get_doc("app", "users", first_key) != null);
+    }
+}
+
 test "index drop survives reopen" {
     var threaded: std.Io.Threaded = .init_single_threaded;
     defer threaded.deinit();
diff --git a/src/pager.zig b/src/pager.zig
index 27c647e..90f5670 100644
--- a/src/pager.zig
+++ b/src/pager.zig
@@ -626,15 +626,29 @@ pub const Pager = struct {
             // writes from here on.
             self.stable_pages = wm.alloc_tail;
             try self.read_freelist(wm);
-        } else if (self.file_pages > page_first_data) {
+        } else if (self.watermark_attempted()) {
+            // Only worth saying when a watermark was *written* and cannot be
+            // read: a data file that simply never reached its first checkpoint
+            // is the normal state of a young database, and warning about it
+            // trains people to ignore the warning that matters.
             std.debug.print(
-                "multiforadb: WARNING: data file '{s}' has no valid watermark; " ++
+                "multiforadb: WARNING: data file '{s}' has a damaged watermark; " ++
                     "treating it as having no checkpoint and replaying the log in full\n",
                 .{self.path},
             );
         }
     }
 
+    /// Whether either slot holds anything at all. A never-checkpointed file has
+    /// both slots as written by create: all zeroes.
+    fn watermark_attempted(self: *const Pager) bool {
+        for ([_]u32{ page_watermark_a, page_watermark_b }) |p| {
+            if (p >= self.mapped_pages) continue;
+            for (self.page(p)) |byte| if (byte != 0) return true;
+        }
+        return false;
+    }
+
     fn read_slot(self: *const Pager, p: u32) ?Watermark {
         if (p >= self.mapped_pages) return null;
         const b = self.page(p);
diff --git a/src/server.zig b/src/server.zig
index 5a86042..c5b8065 100644
--- a/src/server.zig
+++ b/src/server.zig
@@ -73,6 +73,12 @@ fn ttl_monitor(io: std.Io, server: *Server) error{Canceled}!void {
             std.debug.print("multiforadb: TTL sweep failed: {s}\n", .{@errorName(err)});
             continue;
         };
+        if (server.engine.take_checkpoint()) {
+            server.engine.checkpoint() catch |err| {
+                std.debug.print("multiforadb: checkpoint failed: {s}\n", .{@errorName(err)});
+                server.engine.checkpoint_pending.store(true, .release);
+            };
+        }
         if (server.engine.take_compact()) {
             server.engine.compact() catch |err| {
                 std.debug.print("multiforadb: compaction failed: {s}\n", .{@errorName(err)});
diff --git a/src/storage.zig b/src/storage.zig
index 9568ead..039c9f4 100644
--- a/src/storage.zig
+++ b/src/storage.zig
@@ -460,6 +460,27 @@ pub const Log = struct {
     /// Appends never sync (see `append_record`), so nothing is durable until
     /// this returns -- which is why Engine.commit calls it exactly once per
     /// write command, coalescing every writer in flight into a single fsync.
+    /// Discard every record, keeping only the file header.
+    ///
+    /// Only a checkpoint may call this, and only after the watermark that covers
+    /// these records is durable. The whole point of a lagging checkpoint is that
+    /// the log can be reclaimed once the data file holds its effect -- and the
+    /// order is not negotiable: truncate before the watermark is durable and a
+    /// crash in between leaves records gone from the log and absent from the
+    /// image.
+    ///
+    /// Caller holds the log lock. The open block is dropped rather than sealed:
+    /// its records are below the watermark too, so writing them out would only
+    /// be work the next open throws away.
+    pub fn truncate_to_header(self: *Log) !void {
+        self.block.clearRetainingCapacity();
+        try self.file.setLength(self.io, file_header_len);
+        try self.file.sync(self.io);
+        self.end_pos = file_header_len;
+        self.log_bytes = 0;
+        self.data_bytes = 0;
+    }
+
     pub fn sync(self: *Log) !void {
         try self.seal_block();
         try self.file.sync(self.io);