From 11216f9b36f065117004efd0b39e6781c59cb723 Mon Sep 17 00:00:00 2001
From: BlivionIaG <15323295+BlivionIaG@users.noreply.github.com>
Date: Tue, 27 Oct 2026 19:48:13 +0100
Subject: [PATCH] KV pool: host tier

A Tier beside the Pool for families whose pages are cached on the device from a host store. The pool stays the only
owner of page ids and of what is promised; the tier adds a second currency (device frames a stream may pin at once),
the page-to-frame mapping, an LRU order, pins for the frames a call reads, a dirty flag so a frame is dropped only
once its rows reached the host store, and a generation per frame so a table written before a frame was recycled is
told from a current one. A stream can also be lent frames for its life, mapped to no page, for a family that manages
them itself. The pool tells the tier when a page loses its last holder. It moves no bytes and uses no device API: the
family copies what a miss names, and a snapshot reads canonical rows from the host store.

Tests: reservation conservation in both currencies, eviction order, pin rules, lent frames, generation safety,
canonical rows, a randomised serving loop that audits the mapping and checks every row read against the last write,
two ranks in lockstep (and a step or a lent count taken by one rank caught by the digest), and mutants (no generation,
no pin rule, dirty frames unpinned) that the checks fail on.

A failed batch unmaps the misses it mapped (their generation moves on), a page is marked dirty only under a pin
of the marking lease, a stream cannot end while it holds the last pin of a dirty page, and a family can lend a run of
consecutive frames (lendRun).
---
 zig/src/core/kv_pool.zig      |  15 +
 zig/src/core/kv_tier.zig      | 432 ++++++++++++++++++++++++++
 zig/src/core/kv_tier_test.zig | 564 ++++++++++++++++++++++++++++++++++
 3 files changed, 1011 insertions(+)
 create mode 100644 zig/src/core/kv_tier.zig
 create mode 100644 zig/src/core/kv_tier_test.zig

diff --git a/zig/src/core/kv_pool.zig b/zig/src/core/kv_pool.zig
index fb41dab2d7..6ea9dcf9db 100644
--- a/zig/src/core/kv_pool.zig
+++ b/zig/src/core/kv_pool.zig
@@ -142,6 +142,12 @@ pub const Notify = struct {
 /// A family's pool as the load ledger reads it: tokens a page, and the contiguous run it needs (0: none).
 pub const Source = struct { pool: *Pool, page_tokens: u32, run_pages: u32 = 0 };
 
+/// A tier's notice that a page lost its last holder (its frame, if any, is free again).
+pub const Freed = struct {
+    ctx: *anyopaque,
+    call: *const fn (ctx: *anyopaque, id: u32) void,
+};
+
 /// What an idle pool must show: nothing promised, every page free or the store's alone.
 pub const Census = struct { total: usize, free: usize, promised: usize, kept: usize, leased: usize };
 
@@ -160,6 +166,7 @@ pub const Pool = struct {
     reclaim: ?Reclaim = null,
     notify: ?Notify = null,
     depth: u8 = 0, // public changes in progress: the outermost one notifies, after nested calls and rollbacks
+    freed: ?Freed = null,
 
     pub fn init(gpa: Allocator, count: usize, page_bytes: u64) !Pool {
         const refs = try gpa.alloc(u32, count);
@@ -220,6 +227,7 @@ pub const Pool = struct {
         if (p.leased[id] == 0 or p.refs[id] < p.leased[id]) std.debug.panic("kv pool: page {d} released with no holder", .{id});
         p.ownership(id, p.refs[id] - 1, p.leased[id] - 1);
         if (p.refs[id] > 0) return;
+        if (p.freed) |f| f.call(f.ctx, id);
         p.free[p.nfree] = id;
         p.nfree += 1;
     }
@@ -256,6 +264,7 @@ pub const Pool = struct {
         if (p.refs[id] <= p.leased[id]) return error.NotHeld; // no holder past the leases' (callers drop theirs first)
         p.ownership(id, p.refs[id] - 1, p.leased[id]);
         if (p.refs[id] > 0) return;
+        if (p.freed) |f| f.call(f.ctx, id);
         p.free[p.nfree] = id;
         p.nfree += 1;
     }
@@ -431,6 +440,7 @@ pub const Pool = struct {
         for (start..start + k) |id| {
             if (p.refs[id] != 1 or p.leased[id] != 1) std.debug.panic("kv pool: run page {d} released while not a run's", .{id});
             p.ownership(id, 0, 0);
+            if (p.freed) |f| f.call(f.ctx, @intCast(id));
             p.free[p.nfree] = @intCast(id);
             p.nfree += 1;
         }
@@ -456,8 +466,13 @@ pub const Pool = struct {
     }
 };
 
+/// The host tier: logical pages in host RAM and device frames that cache them (kv_tier.zig).
+pub const tier = @import("kv_tier.zig");
+pub const Tier = tier.Tier;
+
 test {
     _ = @import("kv_pool_test.zig");
+    _ = @import("kv_tier.zig");
     _ = @import("kv_pool_counts_test.zig");
     _ = @import("kv_pool_walk_test.zig");
     _ = @import("kv_pool_notify_test.zig");
diff --git a/zig/src/core/kv_tier.zig b/zig/src/core/kv_tier.zig
new file mode 100644
index 0000000000..3d40f1d98c
--- /dev/null
+++ b/zig/src/core/kv_tier.zig
@@ -0,0 +1,432 @@
+//! A host tier for the page pool: logical pages whose canonical rows live in host RAM, cached by device frames.
+const std = @import("std");
+const kv = @import("kv_pool.zig");
+const Allocator = std.mem.Allocator;
+
+/// No frame (for a page) or no page (for a frame).
+pub const none: u32 = std.math.maxInt(u32);
+
+/// What the ranks of one model agree on before they open a tier: the smallest of each.
+pub const Shape = struct {
+    host_pages: usize,
+    frames: usize,
+
+    pub fn agree(a: Shape, b: Shape) Shape {
+        return .{ .host_pages = @min(a.host_pages, b.host_pages), .frames = @min(a.frames, b.frames) };
+    }
+};
+
+/// A deliberate bug a test turns on to show that the check beside it fails without the rule: never set outside tests.
+pub const Fault = enum { none, keep_generation, ignore_pins, ignore_dirty, unpinned_dirty, keep_failed_miss };
+
+/// A page a call pinned: its frame, the frame's generation, and whether the family must copy the page in.
+pub const Fill = struct { page: u32, frame: u32, gen: u32, miss: bool };
+
+/// A page's frame as a device table saw it: current while the frame's generation is the same.
+pub const Ticket = struct { frame: u32, gen: u32 };
+
+/// A stream's reservation in both currencies: pages (the pool's lease) and the frames it may pin at once.
+pub const Lease = struct {
+    pages: kv.Lease = .{},
+    frames: usize = 0,
+    pins: std.ArrayList(u32) = .empty, // one entry per pin, a page id
+    lent: std.ArrayList(u32) = .empty, // frames lent to the stream: mapped to no page, the family's to manage
+};
+
+pub const Census = struct { frames: usize, free: usize, resident: usize, pinned: usize, dirty: usize, promised: usize, lent: usize };
+
+pub const Tier = struct {
+    gpa: Allocator,
+    pool: kv.Pool, // logical pages; the host store holds `pool.total()` of them
+    host: []u8, // canonical rows, `pool.total() * page_bytes`, borrowed (pinned host memory in a server)
+    frame_of: []u32, // per page: its frame or `none`
+    dirty: []bool, // per page: the frame holds rows the host store lacks
+    page_of: []u32, // per frame: its page or `none`
+    gen: []u32, // per frame: changes whenever the frame stops holding the page it held
+    stamp: []u64, // per frame: the LRU clock at its last use
+    pinned: []u32, // per frame: calls reading it
+    free_frames: []u32,
+    lent_mark: []bool, // per frame: lent to a lease
+    nfree: usize,
+    promised: usize = 0, // frames promised, summed over leases
+    nlent: usize = 0, // frames lent, summed over leases
+    clock: u64 = 0,
+    fault: Fault = .none,
+
+    /// In place (the pool's release hook points here): `host.len / page_bytes` logical pages, `frames` frames.
+    pub fn init(t: *Tier, gpa: Allocator, host: []u8, frames: usize, page_bytes: u64) !void {
+        const npages: usize = @intCast(host.len / page_bytes);
+        var pool = try kv.Pool.init(gpa, npages, page_bytes);
+        errdefer pool.deinit();
+        const frame_of = try gpa.alloc(u32, npages);
+        errdefer gpa.free(frame_of);
+        const dirty = try gpa.alloc(bool, npages);
+        errdefer gpa.free(dirty);
+        const page_of = try gpa.alloc(u32, frames);
+        errdefer gpa.free(page_of);
+        const gen = try gpa.alloc(u32, frames);
+        errdefer gpa.free(gen);
+        const stamp = try gpa.alloc(u64, frames);
+        errdefer gpa.free(stamp);
+        const pinned = try gpa.alloc(u32, frames);
+        errdefer gpa.free(pinned);
+        const free_frames = try gpa.alloc(u32, frames);
+        errdefer gpa.free(free_frames);
+        const lent_mark = try gpa.alloc(bool, frames);
+        @memset(frame_of, none);
+        @memset(dirty, false);
+        @memset(page_of, none);
+        @memset(gen, 0);
+        @memset(stamp, 0);
+        @memset(pinned, 0);
+        @memset(lent_mark, false);
+        for (free_frames, 0..) |*f, i| f.* = @intCast(frames - 1 - i); // frame 0 first
+        t.* = .{ .gpa = gpa, .pool = pool, .host = host, .frame_of = frame_of, .dirty = dirty, .page_of = page_of, .gen = gen, .stamp = stamp, .pinned = pinned, .free_frames = free_frames, .lent_mark = lent_mark, .nfree = frames };
+        t.pool.freed = .{ .ctx = t, .call = freed };
+    }
+
+    pub fn deinit(t: *Tier) void {
+        t.pool.deinit();
+        t.gpa.free(t.frame_of);
+        t.gpa.free(t.dirty);
+        t.gpa.free(t.page_of);
+        t.gpa.free(t.gen);
+        t.gpa.free(t.stamp);
+        t.gpa.free(t.pinned);
+        t.gpa.free(t.free_frames);
+        t.gpa.free(t.lent_mark);
+        t.* = undefined;
+    }
+
+    pub fn count(t: *const Tier) usize {
+        return t.page_of.len;
+    }
+
+    pub fn shape(t: *const Tier) Shape {
+        return .{ .host_pages = t.pool.total(), .frames = t.count() };
+    }
+
+    /// Promise `pages` pages and `n` frames to `l`, or neither: LaneMemory or FrameMemory, whichever does not fit.
+    pub fn reserve(t: *Tier, l: *Lease, pages: usize, n: usize) error{ LaneMemory, FrameMemory }!void {
+        if (t.promised + n > t.count()) return error.FrameMemory;
+        try t.pool.promise(&l.pages, pages);
+        t.promised += n;
+        l.frames += n;
+    }
+
+    /// Whether `pages` and `n` frames could be promised now (the store's idle pages counted).
+    pub fn fits(t: *const Tier, pages: usize, n: usize) bool {
+        return t.promised + n <= t.count() and t.pool.fits(pages);
+    }
+
+    /// `l` covers `n` pages, as `Pool.cover`.
+    pub fn cover(t: *Tier, l: *Lease, n: usize, table: ?kv.Table) !void {
+        return t.pool.cover(&l.pages, n, table);
+    }
+
+    /// The frame of `id`, if it has one.
+    pub fn resident(t: *const Tier, id: u32) ?Ticket {
+        const f = t.frame_of[id];
+        return if (f == none) null else .{ .frame = f, .gen = t.gen[f] };
+    }
+
+    /// Whether a table entry written from `tk` still names `id`'s rows.
+    pub fn valid(t: *const Tier, id: u32, tk: Ticket) bool {
+        return t.page_of[tk.frame] == id and t.gen[tk.frame] == tk.gen;
+    }
+
+    /// Pin `id` for a call: a frame if it has none (a miss: copy `host_page(id)` in); FrameMemory past the promise.
+    pub fn pin(t: *Tier, l: *Lease, id: u32) error{ FrameMemory, OutOfMemory }!Fill {
+        if (t.pool.refs[id] == 0) std.debug.panic("kv tier: page {d} pinned while free", .{id});
+        if (l.pins.items.len + l.lent.items.len >= l.frames) return error.FrameMemory;
+        try l.pins.ensureUnusedCapacity(t.gpa, 1);
+        var miss = false;
+        if (t.frame_of[id] == none) {
+            const f = t.takeFrame() orelse return error.FrameMemory;
+            t.page_of[f] = id;
+            t.frame_of[id] = f;
+            miss = true;
+        }
+        const f = t.frame_of[id];
+        t.pinned[f] += 1;
+        t.touch(id);
+        l.pins.appendAssumeCapacity(id);
+        return .{ .page = id, .frame = f, .gen = t.gen[f], .miss = miss };
+    }
+
+    /// Pin every page of `ids` into `out`, or none of them.
+    pub fn fill(t: *Tier, l: *Lease, ids: []const u32, out: []Fill) !void {
+        std.debug.assert(out.len >= ids.len);
+        var done: usize = 0;
+        errdefer {
+            for (ids[0..done]) |id| t.unpin(l, id) catch {};
+            for (out[0..done]) |o| t.undo(o); // a miss was never copied in: its frame must not stay mapped
+        }
+        for (ids, 0..) |id, i| {
+            out[i] = try t.pin(l, id);
+            done += 1;
+        }
+    }
+
+    /// The frame a failed batch took for a miss goes back unmapped (generation moved on), unless something pins it.
+    fn undo(t: *Tier, o: Fill) void {
+        if (t.fault == .keep_failed_miss or !o.miss or t.frame_of[o.page] != o.frame or t.pinned[o.frame] != 0) return;
+        t.unmap(o.frame);
+        t.free_frames[t.nfree] = o.frame;
+        t.nfree += 1;
+    }
+
+    /// One pin of `id` fewer. Dirty: the last pin of a page whose rows the host store lacks cannot go (sync it first).
+    pub fn unpin(t: *Tier, l: *Lease, id: u32) error{ NotPinned, Dirty }!void {
+        var i = l.pins.items.len;
+        while (i > 0) : (i -= 1) if (l.pins.items[i - 1] == id) break;
+        if (i == 0) return error.NotPinned;
+        const f = t.frame_of[id];
+        if (t.pinned[f] == 1 and t.dirty[id] and t.fault != .ignore_dirty) return error.Dirty;
+        _ = l.pins.orderedRemove(i - 1);
+        t.pinned[f] -= 1;
+    }
+
+    /// Lend `n` frames to `l` for its life: mapped to no page, never a victim, counted in its promise like pins.
+    pub fn lend(t: *Tier, l: *Lease, n: usize) error{ FrameMemory, OutOfMemory }!void {
+        if (l.pins.items.len + l.lent.items.len + n > l.frames) return error.FrameMemory;
+        try l.lent.ensureUnusedCapacity(t.gpa, n);
+        var taken: usize = 0;
+        errdefer t.giveBackTo(l, l.lent.items.len - taken);
+        while (taken < n) : (taken += 1) {
+            const f = t.takeFrame() orelse return error.FrameMemory;
+            t.gen[f] +%= 1;
+            t.lent_mark[f] = true;
+            l.lent.appendAssumeCapacity(f);
+            t.nlent += 1;
+        }
+    }
+
+    /// Lend `l` `n` consecutive free frames (a family's state run) for its life; the first frame, or FrameMemory.
+    pub fn lendRun(t: *Tier, l: *Lease, n: usize) error{ FrameMemory, OutOfMemory }!u32 {
+        if (n == 0 or l.pins.items.len + l.lent.items.len + n > l.frames) return error.FrameMemory;
+        var start: usize = 0;
+        var run: usize = 0;
+        for (t.page_of, t.lent_mark, 0..) |p, m, f| {
+            if (p == none and !m) run += 1 else run = 0;
+            if (run == n) {
+                start = f + 1 - n;
+                break;
+            }
+        } else return error.FrameMemory;
+        try l.lent.ensureUnusedCapacity(t.gpa, n);
+        var kept: usize = 0;
+        for (t.free_frames[0..t.nfree]) |f| if (f < start or f >= start + n) {
+            t.free_frames[kept] = f;
+            kept += 1;
+        };
+        t.nfree = kept;
+        for (start..start + n) |f| {
+            t.gen[f] +%= 1;
+            t.lent_mark[f] = true;
+            l.lent.appendAssumeCapacity(@intCast(f));
+        }
+        t.nlent += n;
+        return @intCast(start);
+    }
+
+    /// The frames lent to `l` go back to the free frames, their generation moved on.
+    pub fn giveBack(t: *Tier, l: *Lease) void {
+        t.giveBackTo(l, 0);
+    }
+
+    fn giveBackTo(t: *Tier, l: *Lease, keep: usize) void {
+        while (l.lent.items.len > keep) {
+            const f = l.lent.pop().?;
+            t.gen[f] +%= 1;
+            t.lent_mark[f] = false;
+            t.free_frames[t.nfree] = f;
+            t.nfree += 1;
+            t.nlent -= 1;
+        }
+    }
+
+    /// The LRU clock moves on for `id` (a hit the device saw, or a call that read it).
+    pub fn touch(t: *Tier, id: u32) void {
+        const f = t.frame_of[id];
+        if (f == none) return;
+        t.clock += 1;
+        t.stamp[f] = t.clock;
+    }
+
+    /// Rows were written into `id`'s frame, which `l` pins: the host store lacks them until `sync`.
+    pub fn markDirty(t: *Tier, l: *const Lease, id: u32) error{NotPinned}!void {
+        if (t.fault != .unpinned_dirty and std.mem.indexOfScalar(u32, l.pins.items, id) == null) return error.NotPinned;
+        if (t.frame_of[id] == none) std.debug.panic("kv tier: page {d} written with no frame", .{id});
+        t.dirty[id] = true;
+    }
+
+    /// The family copied `id`'s frame to `host_page(id)`: the frame may be dropped.
+    pub fn sync(t: *Tier, id: u32) void {
+        t.dirty[id] = false;
+    }
+
+    /// Where a page's canonical rows go (a flush writes here).
+    pub fn host_page(t: *Tier, id: u32) []u8 {
+        const b: usize = @intCast(t.pool.page_bytes);
+        return t.host[id * b ..][0..b];
+    }
+
+    /// The canonical rows of `id` for a snapshot or a fill; Unsynced while its frame holds rows not flushed yet.
+    pub fn canonical(t: *const Tier, id: u32) error{Unsynced}![]const u8 {
+        if (t.dirty[id]) return error.Unsynced;
+        const b: usize = @intCast(t.pool.page_bytes);
+        return t.host[id * b ..][0..b];
+    }
+
+    /// Drop up to `n` unpinned frames, least recently used first; the count dropped.
+    pub fn evict(t: *Tier, n: usize) usize {
+        var done: usize = 0;
+        while (done < n) : (done += 1) {
+            const f = t.victim() orelse break;
+            t.unmap(f);
+            t.free_frames[t.nfree] = f;
+            t.nfree += 1;
+        }
+        return done;
+    }
+
+    /// Whether `l` can end: none of its pinned pages is left dirty and unpinned (a page only it holds goes with it).
+    pub fn canFinish(t: *const Tier, l: *const Lease) bool {
+        if (t.fault == .ignore_dirty) return true;
+        for (l.pins.items) |id| {
+            if (!t.dirty[id]) continue;
+            const f = t.frame_of[id];
+            var mine: usize = 0; // this lease's own pins of the page
+            for (l.pins.items) |q| mine += @intFromBool(q == id);
+            if (f != none and t.pinned[f] > mine) continue; // another stream still pins it
+            const own = std.mem.indexOfScalar(u32, l.pages.ids.items, id) != null;
+            if (!own or t.pool.refs[id] > 1) return false;
+        }
+        for (l.pages.ids.items) |id| if (t.dirty[id] and t.pool.refs[id] > 1) return false;
+        return true;
+    }
+
+    /// The stream ended: its pins, frames, promise and pages go back (a page nobody else holds loses its frame).
+    pub fn finish(t: *Tier, l: *Lease) void {
+        if (t.fault == .none and !t.canFinish(l)) std.debug.panic("kv tier: a stream ended with unflushed rows pinned or shared", .{});
+        for (l.pins.items) |id| {
+            const f = t.frame_of[id];
+            if (f != none) t.pinned[f] -= 1 else if (t.fault == .none) std.debug.panic("kv tier: pinned page {d} lost its frame", .{id});
+        }
+        l.pins.deinit(t.gpa);
+        t.giveBack(l);
+        l.lent.deinit(t.gpa);
+        t.promised -= l.frames;
+        t.pool.finish(&l.pages);
+        l.* = .{};
+    }
+
+    pub fn census(t: *const Tier) Census {
+        var c: Census = .{ .frames = t.count(), .free = t.nfree, .resident = 0, .pinned = 0, .dirty = 0, .promised = t.promised, .lent = t.nlent };
+        for (t.page_of, t.pinned) |p, n| {
+            c.resident += @intFromBool(p != none);
+            c.pinned += @intFromBool(n > 0);
+        }
+        for (t.dirty) |d| c.dirty += @intFromBool(d);
+        return c;
+    }
+
+    /// Nothing promised, pinned or dirty, and every frame free or holding a page still held (the store's).
+    pub fn idle(t: *const Tier) bool {
+        const c = t.census();
+        return t.pool.idle() and c.promised == 0 and c.pinned == 0 and c.dirty == 0 and c.lent == 0 and c.free + c.resident == c.frames;
+    }
+
+    /// The mapping's own consistency; an error names the broken rule. Tests call it after every step.
+    pub fn audit(t: *const Tier) error{Broken}!void {
+        var mapped: usize = 0;
+        for (t.page_of, 0..) |p, f| {
+            if (p == none) {
+                if (t.pinned[f] != 0) return error.Broken; // a pin on an empty frame
+                continue;
+            }
+            mapped += 1;
+            if (t.frame_of[p] != f) return error.Broken; // page and frame disagree
+            if (t.pool.refs[p] == 0) return error.Broken; // a free page keeps a frame
+        }
+        if (mapped + t.nfree + t.nlent != t.count()) return error.Broken; // frames neither free, mapped nor lent, or two
+        for (t.free_frames[0..t.nfree]) |f| if (t.page_of[f] != none or t.lent_mark[f]) return error.Broken;
+        var marked: usize = 0;
+        for (t.lent_mark, t.page_of) |m, p| {
+            marked += @intFromBool(m);
+            if (m and p != none) return error.Broken; // a lent frame holds a page
+        }
+        if (marked != t.nlent) return error.Broken;
+        for (t.frame_of, 0..) |f, p| {
+            if (f != none and t.page_of[f] != p) return error.Broken;
+            if (t.dirty[p] and f == none) return error.Broken; // unflushed rows with no frame
+        }
+        if (t.promised > t.count()) return error.Broken;
+    }
+
+    /// A hash of what two ranks must hold alike: mapping, generations, LRU order, pins, holders and promises.
+    pub fn digest(t: *const Tier) u64 {
+        var h = std.hash.Wyhash.init(0);
+        h.update(std.mem.sliceAsBytes(t.frame_of));
+        h.update(std.mem.sliceAsBytes(t.page_of));
+        h.update(std.mem.sliceAsBytes(t.gen));
+        h.update(std.mem.sliceAsBytes(t.stamp));
+        h.update(std.mem.sliceAsBytes(t.pinned));
+        h.update(std.mem.sliceAsBytes(t.free_frames[0..t.nfree]));
+        h.update(std.mem.sliceAsBytes(t.pool.refs));
+        h.update(std.mem.sliceAsBytes(t.pool.free[0..t.pool.nfree]));
+        const counts = [_]u64{ t.promised, t.pool.promised, t.clock, t.nlent };
+        h.update(std.mem.sliceAsBytes(&counts));
+        return h.final();
+    }
+
+    /// A free frame, or the least recently used unpinned one dropped.
+    fn takeFrame(t: *Tier) ?u32 {
+        if (t.nfree > 0) {
+            t.nfree -= 1;
+            return t.free_frames[t.nfree];
+        }
+        const f = t.victim() orelse return null;
+        t.unmap(f);
+        return f;
+    }
+
+    /// The mapped frame with the oldest stamp, not pinned and with no unflushed rows (the lowest index among equals).
+    fn victim(t: *const Tier) ?u32 {
+        var best: ?u32 = null;
+        for (t.page_of, 0..) |p, f| {
+            if (p == none) continue;
+            if (t.pinned[f] > 0 and t.fault != .ignore_pins) continue;
+            if (t.dirty[p] and t.fault != .ignore_dirty) continue;
+            if (best == null or t.stamp[f] < t.stamp[best.?]) best = @intCast(f);
+        }
+        return best;
+    }
+
+    /// The frame stops holding its page: the generation moves on so a table naming it is stale.
+    fn unmap(t: *Tier, f: u32) void {
+        const p = t.page_of[f];
+        if (t.dirty[p] and t.fault != .ignore_dirty) std.debug.panic("kv tier: frame {d} dropped with page {d} unflushed", .{ f, p });
+        t.frame_of[p] = none;
+        t.dirty[p] = false;
+        t.page_of[f] = none;
+        if (t.fault != .keep_generation) t.gen[f] +%= 1;
+    }
+
+    /// The pool freed `id` (its last holder let go): its frame is free again.
+    fn freed(ctx: *anyopaque, id: u32) void {
+        const t: *Tier = @ptrCast(@alignCast(ctx));
+        const f = t.frame_of[id];
+        if (f == none) return;
+        if (t.pinned[f] != 0 and t.fault == .none) std.debug.panic("kv tier: page {d} freed while pinned", .{id});
+        t.dirty[id] = false; // the page is gone: rows nobody can read need no flush
+        t.unmap(f);
+        t.free_frames[t.nfree] = f;
+        t.nfree += 1;
+    }
+};
+
+test {
+    _ = @import("kv_tier_test.zig");
+}
diff --git a/zig/src/core/kv_tier_test.zig b/zig/src/core/kv_tier_test.zig
new file mode 100644
index 0000000000..95437b2bea
--- /dev/null
+++ b/zig/src/core/kv_tier_test.zig
@@ -0,0 +1,564 @@
+//! The host tier: two currencies, eviction order, pins, generations, canonical rows, lockstep ranks, mutants.
+const std = @import("std");
+const kv = @import("kv_pool.zig");
+const kt = @import("kv_tier.zig");
+const Tier = kt.Tier;
+const Lease = kt.Lease;
+const expect = std.testing.expect;
+const expectEqual = std.testing.expectEqual;
+const gpa = std.testing.allocator;
+
+const page_bytes = 8;
+
+/// A tier of `pages` logical pages and `frames` frames over a host store and a fake device.
+const Rig = struct {
+    host: []u8,
+    dev: []u8,
+    tier: Tier,
+
+    fn open(r: *Rig, pages: usize, frames: usize) !void {
+        r.host = try gpa.alloc(u8, pages * page_bytes);
+        errdefer gpa.free(r.host);
+        @memset(r.host, 0);
+        r.dev = try gpa.alloc(u8, frames * page_bytes);
+        errdefer gpa.free(r.dev);
+        @memset(r.dev, 0);
+        try r.tier.init(gpa, r.host, frames, page_bytes);
+    }
+
+    fn close(r: *Rig) void {
+        r.tier.deinit();
+        gpa.free(r.host);
+        gpa.free(r.dev);
+    }
+
+    fn frame(r: *Rig, f: u32) []u8 {
+        return r.dev[f * page_bytes ..][0..page_bytes];
+    }
+
+    /// A stream with `pages` pages covered and `frames` frames promised.
+    fn lease(r: *Rig, pages: usize, frames: usize) !Lease {
+        return r.leaseLent(pages, frames, 0);
+    }
+
+    /// A stream that also has `lent` of its frames lent for its life.
+    fn leaseLent(r: *Rig, pages: usize, frames: usize, lent: usize) !Lease {
+        var l: Lease = .{};
+        errdefer r.tier.finish(&l);
+        try r.tier.reserve(&l, pages, frames);
+        try r.tier.cover(&l, pages, null);
+        try r.tier.lend(&l, lent);
+        return l;
+    }
+
+    /// Pin `id`, copying its canonical rows in on a miss.
+    fn pin(r: *Rig, l: *Lease, id: u32) !kt.Fill {
+        const f = try r.tier.pin(l, id);
+        if (f.miss) @memcpy(r.frame(f.frame), r.tier.host_page(id));
+        return f;
+    }
+
+    /// Write `b` into every row byte of the pinned page and flush it to the host store.
+    fn write(r: *Rig, l: *const Lease, id: u32, b: u8) void {
+        const tk = r.tier.resident(id).?;
+        @memset(r.frame(tk.frame), b);
+        r.tier.markDirty(l, id) catch unreachable;
+    }
+
+    fn flush(r: *Rig, id: u32) void {
+        const tk = r.tier.resident(id).?;
+        @memcpy(r.tier.host_page(id), r.frame(tk.frame));
+        r.tier.sync(id);
+    }
+};
+
+test "two currencies: a promise that fits one and not the other takes nothing, and finishing returns both" {
+    var r: Rig = undefined;
+    try r.open(8, 3);
+    defer r.close();
+    const t = &r.tier;
+    var a = try r.lease(4, 2);
+    try std.testing.expectError(error.FrameMemory, t.reserve(&a, 1, 2)); // frames: 2 + 2 > 3
+    try expectEqual(@as(usize, 0), a.pages.promised);
+    try expectEqual(@as(usize, 2), t.promised);
+    var b: Lease = .{};
+    try std.testing.expectError(error.LaneMemory, t.reserve(&b, 5, 1)); // pages: 4 + 5 > 8
+    try expectEqual(@as(usize, 2), t.promised); // the frame promise was not kept
+    try expectEqual(@as(usize, 0), t.pool.promised);
+    try expect(!t.fits(5, 1) and t.fits(4, 1) and !t.fits(4, 2));
+    var c = try r.lease(4, 1);
+    try expect(!t.fits(1, 0)); // every page is promised or taken
+    t.finish(&a);
+    t.finish(&c);
+    t.finish(&b);
+    try expect(t.idle());
+    try t.audit();
+}
+
+test "a lease cannot pin more frames than it promised, and a failed batch pins nothing" {
+    var r: Rig = undefined;
+    try r.open(8, 4);
+    defer r.close();
+    const t = &r.tier;
+    var a = try r.lease(4, 2);
+    var out: [3]kt.Fill = undefined;
+    const ids = a.pages.ids.items;
+    try std.testing.expectError(error.FrameMemory, t.fill(&a, ids[0..3], &out));
+    try expectEqual(@as(usize, 0), t.census().pinned);
+    try expectEqual(@as(usize, 0), a.pins.items.len);
+    try expect(t.resident(ids[0]) == null and t.resident(ids[1]) == null); // the misses of the failed batch were unmapped
+    try t.fill(&a, ids[0..2], out[0..2]);
+    try expect(out[0].miss and out[1].miss);
+    try expectEqual(@as(usize, 2), t.census().pinned);
+    try std.testing.expectError(error.FrameMemory, t.pin(&a, ids[2]));
+    t.finish(&a);
+    try expect(t.idle());
+}
+
+test "eviction takes the least recently used unpinned frame, and a touch moves a page to the back" {
+    var r: Rig = undefined;
+    try r.open(8, 3);
+    defer r.close();
+    const t = &r.tier;
+    var a = try r.lease(6, 1);
+    const ids = a.pages.ids.items;
+    for (ids[0..3]) |id| { // one frame promised: pin, read, unpin, three times over
+        _ = try r.pin(&a, id);
+        try t.unpin(&a, id);
+    }
+    try expectEqual(@as(usize, 3), t.census().resident);
+    t.touch(ids[0]); // order now 1, 2, 0
+    const f = try r.pin(&a, ids[3]); // evicts page 1
+    try expect(f.miss);
+    try expect(t.resident(ids[1]) == null and t.resident(ids[0]) != null and t.resident(ids[2]) != null);
+    try t.unpin(&a, ids[3]);
+    try expectEqual(@as(usize, 1), t.evict(1)); // page 2 is next
+    try expect(t.resident(ids[2]) == null and t.resident(ids[0]) != null);
+    try expectEqual(@as(usize, 2), t.evict(5)); // 0 and 3: all that is left
+    try expectEqual(@as(usize, 3), t.census().free);
+    try t.audit();
+    t.finish(&a);
+}
+
+test "pinned frames are never evicted, and a dirty page stays until its rows reach the host store" {
+    var r: Rig = undefined;
+    try r.open(8, 2);
+    defer r.close();
+    const t = &r.tier;
+    var a = try r.lease(2, 1);
+    var b = try r.lease(2, 1);
+    const pa = a.pages.ids.items[0];
+    const pb = b.pages.ids.items;
+    _ = try r.pin(&a, pa);
+    r.write(&a, pa, 7);
+    _ = try r.pin(&b, pb[0]);
+    try t.unpin(&b, pb[0]);
+    _ = try r.pin(&b, pb[1]); // the one frame that can go is pb[0]'s
+    try expect(t.resident(pa) != null and t.resident(pb[0]) == null);
+    try std.testing.expectError(error.Dirty, t.unpin(&a, pa));
+    try expectEqual(@as(usize, 0), t.evict(2)); // pa pinned and dirty, pb[1] pinned
+    try t.unpin(&b, pb[1]);
+    try expectEqual(@as(usize, 1), t.evict(2)); // only pb[1]
+    try expect(t.resident(pa) != null);
+    try std.testing.expectError(error.Unsynced, t.canonical(pa));
+    r.flush(pa);
+    try t.unpin(&a, pa);
+    try expectEqual(@as(u8, 7), (try t.canonical(pa))[3]);
+    try expectEqual(@as(usize, 1), t.evict(2));
+    t.finish(&a);
+    t.finish(&b);
+    try expect(t.idle());
+}
+
+/// Whether a table entry written before a frame was dropped and filled again with the same page is taken for current.
+fn staleAccepted(fault: kt.Fault) !bool {
+    var r: Rig = undefined;
+    try r.open(4, 1);
+    defer r.close();
+    r.tier.fault = fault;
+    var a = try r.lease(2, 1);
+    defer r.tier.finish(&a);
+    const p = a.pages.ids.items;
+    const f0 = try r.pin(&a, p[0]);
+    const old: kt.Ticket = .{ .frame = f0.frame, .gen = f0.gen };
+    try r.tier.unpin(&a, p[0]);
+    _ = try r.pin(&a, p[1]); // the only frame is recycled
+    try r.tier.unpin(&a, p[1]);
+    _ = try r.pin(&a, p[0]); // and returns to the first page: same frame, new contents
+    try r.tier.unpin(&a, p[0]);
+    return r.tier.valid(p[0], old);
+}
+
+test "generations: a recycled frame is not read through a table written before" {
+    try expect(!try staleAccepted(.none));
+    try expect(try staleAccepted(.keep_generation)); // mutation: without the generation the stale table passes
+}
+
+/// Whether a frame a call pins survives another stream's miss.
+fn pinnedSurvives(fault: kt.Fault) !bool {
+    var r: Rig = undefined;
+    try r.open(4, 2);
+    defer r.close();
+    r.tier.fault = fault;
+    var a = try r.lease(1, 1);
+    defer r.tier.finish(&a);
+    var b = try r.lease(2, 1);
+    defer r.tier.finish(&b);
+    const pa = a.pages.ids.items[0];
+    _ = try r.pin(&a, pa); // the oldest stamp, and pinned
+    _ = try r.pin(&b, b.pages.ids.items[0]);
+    try r.tier.unpin(&b, b.pages.ids.items[0]);
+    _ = try r.pin(&b, b.pages.ids.items[1]);
+    return r.tier.resident(pa) != null;
+}
+
+test "pin rule: the frame a call holds survives a miss; mutation: ignoring pins evicts it" {
+    try expect(try pinnedSurvives(.none));
+    try expect(!try pinnedSurvives(.ignore_pins));
+}
+
+test "freeing a page (its last holder lets go) frees its frame; a page the store still holds keeps it" {
+    var r: Rig = undefined;
+    try r.open(4, 2);
+    defer r.close();
+    const t = &r.tier;
+    var a = try r.lease(2, 2);
+    const p = try gpa.dupe(u32, a.pages.ids.items); // ids outlive the lease
+    defer gpa.free(p);
+    _ = try r.pin(&a, p[0]);
+    r.write(&a, p[0], 5);
+    r.flush(p[0]);
+    _ = try r.pin(&a, p[1]);
+    t.pool.retain(p[0]); // the prompt store keeps the first page
+    t.finish(&a);
+    try expect(t.resident(p[0]) != null and t.resident(p[1]) == null);
+    try expectEqual(@as(usize, 1), t.census().free);
+    try expectEqual(@as(u8, 5), (try t.canonical(p[0]))[0]);
+    try t.pool.drop(p[0]);
+    try expect(t.resident(p[0]) == null and t.idle());
+    try t.audit();
+}
+
+/// A fake serving loop: streams open, pin, write, flush, unpin, finish; every row read equals the page's last write.
+const Sim = struct {
+    r: Rig,
+    rng: std.Random.DefaultPrng,
+    slots: [3]?Lease = .{ null, null, null },
+    model: [16]u8 = @splat(0), // the byte last written to a page (0: none)
+    next: u8 = 1,
+    misses: usize = 0,
+    reads: usize = 0,
+
+    fn open(s: *Sim, seed: u64, frames: usize) !void {
+        s.* = .{ .r = undefined, .rng = .init(seed) };
+        try s.r.open(16, frames);
+    }
+
+    fn close(s: *Sim) void {
+        for (&s.slots) |*l| if (l.*) |*x| s.r.tier.finish(x);
+        s.r.close();
+    }
+
+    fn check(s: *Sim, id: u32, tk: kt.Ticket) !void {
+        if (s.model[id] == 0) return;
+        for (s.r.frame(tk.frame)) |x| try expectEqual(s.model[id], x);
+        s.reads += 1;
+    }
+
+    fn step(s: *Sim) !void {
+        const t = &s.r.tier;
+        const rnd = s.rng.random();
+        const op = rnd.uintLessThan(u32, 100);
+        const k = rnd.uintLessThan(usize, s.slots.len);
+        const l: *Lease = if (s.slots[k]) |*x| x else {
+            if (op < 60) s.slots[k] = s.r.leaseLent(4, 3, 1) catch null;
+            return;
+        };
+        const ids = l.pages.ids.items;
+        const id = ids[rnd.uintLessThan(usize, ids.len)];
+        switch (op) {
+            0...34 => {
+                const f = s.r.pin(l, id) catch |e| {
+                    if (e == error.FrameMemory and l.pins.items.len + l.lent.items.len >= l.frames) return;
+                    return e;
+                };
+                s.misses += @intFromBool(f.miss);
+                try s.check(id, .{ .frame = f.frame, .gen = f.gen });
+            },
+            35...59 => if (l.pins.items.len > 0) {
+                const p = l.pins.items[rnd.uintLessThan(usize, l.pins.items.len)];
+                s.model[p] = s.next;
+                s.next = if (s.next == 255) 1 else s.next + 1;
+                s.r.write(l, p, s.model[p]);
+            },
+            60...79 => if (l.pins.items.len > 0) {
+                const p = l.pins.items[rnd.uintLessThan(usize, l.pins.items.len)];
+                if (t.dirty[p]) s.r.flush(p);
+                try t.unpin(l, p);
+            },
+            80...89 => _ = t.evict(1),
+            else => {
+                for (l.pins.items) |p| if (t.dirty[p]) s.r.flush(p);
+                for (ids) |p| s.model[p] = 0;
+                t.finish(l);
+                s.slots[k] = null;
+            },
+        }
+        try t.audit();
+        for (t.page_of, 0..) |p, f| if (p != kt.none and !t.dirty[p]) try s.check(p, .{ .frame = @intCast(f), .gen = t.gen[f] });
+    }
+};
+
+test "a random serving loop with fewer frames than pages: the audit holds, every read equals the last write, and it ends idle" {
+    inline for (.{ 1, 2, 3 }) |seed| {
+        var s: Sim = undefined;
+        try s.open(seed, 6);
+        defer s.close();
+        for (0..4000) |_| try s.step();
+        try expect(s.misses > 100 and s.reads > 500);
+        for (&s.slots) |*l| if (l.*) |*x| {
+            for (x.pins.items) |p| if (s.r.tier.dirty[p]) s.r.flush(p);
+            s.r.tier.finish(x);
+            l.* = null;
+        };
+        _ = s.r.tier.evict(100);
+        try expect(s.r.tier.idle());
+    }
+}
+
+test "two ranks in lockstep: the same openings and pins give the same frames, generations and order, step after step" {
+    var a: Sim = undefined;
+    var b: Sim = undefined;
+    try a.open(9, 6);
+    defer a.close();
+    try b.open(9, 6);
+    defer b.close();
+    for (0..3000) |_| {
+        try a.step();
+        try b.step();
+        try expectEqual(a.r.tier.digest(), b.r.tier.digest());
+    }
+}
+
+test "two ranks: a step only one rank takes is caught by the digest at once" {
+    var a: Sim = undefined;
+    var b: Sim = undefined;
+    try a.open(9, 6);
+    defer a.close();
+    try b.open(9, 6);
+    defer b.close();
+    var caught: ?usize = null;
+    for (0..3000) |i| {
+        try a.step();
+        if (i == 1500) _ = b.r.tier.evict(1) else try b.step(); // rank 1 evicts instead of the step rank 0 took
+        if (a.r.tier.digest() != b.r.tier.digest()) {
+            caught = i;
+            break;
+        }
+    }
+    try expect(caught != null);
+}
+
+test "ranks agree on the smaller shape before they open" {
+    const a: kt.Shape = .{ .host_pages = 100, .frames = 40 };
+    const b: kt.Shape = .{ .host_pages = 90, .frames = 48 };
+    try expectEqual(kt.Shape{ .host_pages = 90, .frames = 40 }, a.agree(b));
+    try expectEqual(a.agree(b), b.agree(a));
+}
+
+test "lent frames: counted in the frame promise, never victims or mapped, back at finish with a new generation" {
+    var r: Rig = undefined;
+    try r.open(8, 4);
+    defer r.close();
+    const t = &r.tier;
+    var a = try r.leaseLent(4, 3, 2); // two lent, one left to pin
+    try expectEqual(@as(usize, 2), t.census().lent);
+    try expectEqual(@as(usize, 2), a.lent.items.len);
+    try std.testing.expectError(error.FrameMemory, t.lend(&a, 2));
+    const gen0 = t.gen[a.lent.items[0]];
+    var b = try r.lease(4, 1);
+    const ids = a.pages.ids.items;
+    _ = try r.pin(&a, ids[0]);
+    try std.testing.expectError(error.FrameMemory, t.pin(&a, ids[1])); // the lent frames used the promise
+    try t.unpin(&a, ids[0]);
+    for (b.pages.ids.items) |id| { // misses cycle through the two frames left; the lent ones are never taken
+        _ = try r.pin(&b, id);
+        try t.unpin(&b, id);
+        _ = try r.pin(&a, ids[1]);
+        try t.unpin(&a, ids[1]);
+        for (a.lent.items) |f| try expect(t.page_of[f] == kt.none);
+    }
+    try expect(t.evict(10) <= 2); // only the mapped frames can go, the lent ones are not among them
+    try t.audit();
+    const lent0 = a.lent.items[0];
+    t.finish(&a);
+    try expect(t.gen[lent0] != gen0);
+    try expectEqual(@as(usize, 0), t.census().lent);
+    t.finish(&b);
+    try expect(t.idle());
+    try t.audit();
+}
+
+test "ranks lending a different count are told apart by the digest" {
+    var a: Rig = undefined;
+    var b: Rig = undefined;
+    try a.open(8, 4);
+    defer a.close();
+    try b.open(8, 4);
+    defer b.close();
+    var la = try a.leaseLent(4, 3, 2);
+    defer a.tier.finish(&la);
+    var lb = try b.leaseLent(4, 3, 2);
+    defer b.tier.finish(&lb);
+    try expectEqual(a.tier.digest(), b.tier.digest());
+    try b.tier.lend(&lb, 0);
+    var lc = try b.leaseLent(0, 1, 1);
+    defer b.tier.finish(&lc);
+    try expect(a.tier.digest() != b.tier.digest());
+}
+
+test "a lent run is consecutive frames, leaves the free frames, and comes back at finish" {
+    var r: Rig = undefined;
+    try r.open(8, 6);
+    defer r.close();
+    const t = &r.tier;
+    var a = try r.lease(2, 4);
+    _ = try r.pin(&a, a.pages.ids.items[0]); // a mapped frame in the way
+    const first = try t.lendRun(&a, 3);
+    try expectEqual(@as(usize, 3), t.census().lent);
+    for (a.lent.items, 0..) |f, i| try expectEqual(first + i, f);
+    try t.audit();
+    var b = try r.lease(2, 2);
+    try std.testing.expectError(error.FrameMemory, t.lendRun(&b, 3)); // past b's promise
+    _ = try r.pin(&b, b.pages.ids.items[0]);
+    _ = try r.pin(&b, b.pages.ids.items[1]);
+    try std.testing.expectEqual(@as(usize, 0), t.nfree); // 1 mapped (a) + 3 lent + 2 mapped (b)
+    t.finish(&a);
+    t.finish(&b);
+    try expect(t.idle());
+    try t.audit();
+    var c = try r.lease(1, 6);
+    defer t.finish(&c);
+    try expectEqual(@as(u32, 0), try t.lendRun(&c, 6));
+}
+
+/// A failed batch, then the same pages pinned again through the rig: the frames must equal the host pages.
+fn failedFillReads(fault: kt.Fault) !bool {
+    var r: Rig = undefined;
+    try r.open(8, 4);
+    defer r.close();
+    r.tier.fault = fault;
+    for (r.host, 0..) |*b, i| b.* = @intCast(1 + i / page_bytes); // page k holds k + 1
+    @memset(r.dev, 0xee); // the frames hold something else
+    var a = try r.lease(4, 2);
+    defer r.tier.finish(&a);
+    const ids = a.pages.ids.items;
+    var out: [3]kt.Fill = undefined;
+    try std.testing.expectError(error.FrameMemory, r.tier.fill(&a, ids[0..3], &out));
+    try r.tier.audit();
+    for (ids[0..2]) |id| {
+        const f = try r.pin(&a, id);
+        if (!std.mem.eql(u8, r.frame(f.frame), r.tier.host_page(id))) return false;
+    }
+    return true;
+}
+
+test "a failed batch leaves no miss mapped: the pages pinned again read the host rows; mutation: keeping the mapping reads stale frames" {
+    try expect(try failedFillReads(.none));
+    try expect(!try failedFillReads(.keep_failed_miss));
+}
+
+test "a failed batch moves the generation of the frames it unmapped" {
+    var r: Rig = undefined;
+    try r.open(4, 2);
+    defer r.close();
+    var a = try r.lease(3, 2);
+    defer r.tier.finish(&a);
+    const ids = a.pages.ids.items;
+    var out: [3]kt.Fill = undefined;
+    const before = [2]u32{ r.tier.gen[0], r.tier.gen[1] };
+    try std.testing.expectError(error.FrameMemory, r.tier.fill(&a, ids[0..3], &out));
+    try expect(r.tier.gen[0] != before[0] and r.tier.gen[1] != before[1]);
+    try expectEqual(@as(usize, 2), r.tier.census().free);
+}
+
+test "dirty needs a pin by the lease: an unpinned frame cannot be marked; mutation: a dirty frame nobody pins starves a lease inside its promise" {
+    var r: Rig = undefined;
+    try r.open(4, 2);
+    defer r.close();
+    var a = try r.lease(1, 1);
+    defer r.tier.finish(&a);
+    var b = try r.lease(1, 1);
+    defer r.tier.finish(&b);
+    const pa = a.pages.ids.items[0];
+    _ = try r.pin(&a, pa);
+    try r.tier.unpin(&a, pa);
+    try std.testing.expectError(error.NotPinned, r.tier.markDirty(&a, pa));
+    try std.testing.expectError(error.NotPinned, r.tier.markDirty(&b, pa)); // pinned by nobody of b
+    try r.tier.audit();
+    // the mutant lets the mark through on a frame no lease pins, and the only frame is then stuck
+    var m: Rig = undefined;
+    try m.open(4, 2);
+    defer m.close();
+    m.tier.fault = .unpinned_dirty;
+    var c = try m.lease(2, 1);
+    defer m.tier.finish(&c);
+    var d = try m.lease(1, 1);
+    defer m.tier.finish(&d);
+    for (c.pages.ids.items) |pc| { // both frames hold c's pages, unpinned
+        _ = try m.pin(&c, pc);
+        try m.tier.unpin(&c, pc);
+    }
+    for (c.pages.ids.items) |pc| try m.tier.markDirty(&c, pc);
+    try std.testing.expectError(error.FrameMemory, m.pin(&d, d.pages.ids.items[0]));
+}
+
+/// A lease that pins a dirty page it does not own, after the owner let go of its pin: whether it may end.
+fn endsWithForeignDirty(fault: kt.Fault) !bool {
+    var r: Rig = undefined;
+    try r.open(4, 2);
+    defer r.close();
+    r.tier.fault = fault;
+    var a = try r.lease(1, 1);
+    var b = try r.lease(1, 1);
+    const pa = a.pages.ids.items[0];
+    _ = try r.pin(&a, pa);
+    _ = try r.pin(&b, pa);
+    r.write(&a, pa, 9);
+    try r.tier.unpin(&a, pa); // b still pins it
+    const ok = r.tier.canFinish(&b);
+    if (ok) {
+        r.tier.finish(&b);
+        r.flush(pa); // too late for a real stream; the test only wants the tier consistent to close
+    } else {
+        r.flush(pa);
+        r.tier.finish(&b);
+    }
+    r.tier.finish(&a);
+    return ok;
+}
+
+test "a stream cannot end while it holds the last pin of a dirty page, its own table or not; mutation: ignoring dirty lets it" {
+    try expect(!try endsWithForeignDirty(.none));
+    try expect(try endsWithForeignDirty(.ignore_dirty));
+}
+
+test "a stream ends cleanly once the dirty pages it pins are flushed, and one that shares a dirty page of its own cannot" {
+    var r: Rig = undefined;
+    try r.open(4, 2);
+    defer r.close();
+    const t = &r.tier;
+    var a = try r.lease(2, 2);
+    const p = a.pages.ids.items;
+    _ = try r.pin(&a, p[0]);
+    r.write(&a, p[0], 3);
+    try expect(t.canFinish(&a)); // its own page and its only holder: the page goes with it
+    t.pool.retain(p[0]);
+    try expect(!t.canFinish(&a)); // now another holder would read rows the host lacks
+    r.flush(p[0]);
+    try expect(t.canFinish(&a));
+    const keep = p[0];
+    t.finish(&a);
+    try t.pool.drop(keep);
+    try expect(t.idle());
+}