Skip to content
zit Artifact [39cf87e72e]
Login

Artifact [39cf87e72e]

Check-ins Using Download Hex

Artifact 39cf87e72e6c9b65a5fd5b27a44df41d0bf700b04f35a17491eb63047ca090b9:

//! The DST sweep for the writer lock: several writers append to ONE log,
//! interleaved step by step off the seed, through a simulated file that
//! models the lock the way the OS provides it.
//!
//! The storage sweep drives one writer against a faulty file, which is the
//! right shape for crash recovery and the wrong one for this: a lock nobody
//! else contends for is never exercised, so a writer that skipped it, or read
//! the file's end before taking it, passed every seed. Here the lock is part
//! of the model. Each writer is a small state machine walked one action per
//! step, in the order a real verb does them:
//!
//!   make a handle      the `stat` a caller takes to build its FileStorage
//!   LogStore.open      takes the lock, then scans for the end
//!   put × k            appends, one per step
//!   commit, close      fsync, then the lock goes with the handle
//!
//! so another writer can run between any two of them. A writer whose lock is
//! held by someone else is simply not advanced until it is free.
//!
//! The file itself is the oracle for the protocol. It records two violations
//! the moment they happen, which is exactly the damage an unserialised writer
//! does and which no checksum can see afterwards:
//!
//!   an append or truncate by a handle that does not hold the lock
//!   an append that does not land at the file's current end, i.e. one that
//!     writes over frames already there
//!
//! The writer's own view is checked too, because a writer can put every byte
//! in the right place and still be wrong about where it put them. A writer
//! that read the end before taking the lock indexes a stale file: once it
//! holds the lock it must already see every artifact committed before it, and
//! everything it puts must read back through its own handle, including deltas
//! against artifacts other writers committed, whose basis offsets are computed
//! from the end it believes in.
//!
//! At the end, every artifact a writer committed must be present and
//! byte-identical through a fresh reader, and the log must verify.
const std = @import("std");
const storage = @import("zit").storage;
const logstore = @import("zit").logstore;

const hash_len = 32;

pub const LockSweepError = error{
    OutOfMemory,
    Io,
    /// A handle appended or truncated without holding the writer lock.
    UnlockedWrite,
    /// A handle appended somewhere other than the file's end: over frames
    /// another writer had already written.
    OverwroteFrames,
    /// A committed artifact is missing or reads back different bytes.
    MissingAcknowledged,
    /// The shared log no longer opens or verifies.
    LockSweepCorrupt,
    /// A writer holding the lock does not see an artifact committed before it
    /// took the lock, or cannot read back what it just put.
    StaleView,
};

/// The one file every writer shares: its bytes and its writer lock.
const Shared = struct {
    bytes: std.ArrayList(u8) = .empty,
    owner: ?u32 = null,
    unlocked_writes: u32 = 0,
    overwrites: u32 = 0,

    fn deinit(self: *Shared, gpa: std.mem.Allocator) void {
        self.bytes.deinit(gpa);
    }
};

/// One open handle on the shared file, with `FileStorage`'s shape: an append
/// watermark captured when the handle is made, which is what a caller's
/// `stat` gives it before `LogStore.open`.
const Handle = struct {
    gpa: std.mem.Allocator,
    shared: *Shared,
    id: u32,
    end: u64,

    fn storageFile(self: *Handle) storage.StorageFile {
        return .{ .context = self, .vtable = &vtable };
    }

    /// Closing the handle releases a lock it holds, as closing a file
    /// description releases its `flock`.
    fn close(self: *Handle) void {
        if (self.shared.owner == self.id) self.shared.owner = null;
    }

    const vtable: storage.StorageFile.VTable = .{
        .append = appendImpl,
        .pread = preadImpl,
        .sync = syncImpl,
        .size = sizeImpl,
        .truncate = truncateImpl,
        .lock = lockImpl,
    };

    fn appendImpl(ctx: *anyopaque, bytes: []const u8) storage.StorageError!void {
        const self: *Handle = @ptrCast(@alignCast(ctx));
        const s = self.shared;
        if (s.owner != self.id) s.unlocked_writes += 1;
        if (self.end != s.bytes.items.len) s.overwrites += 1;
        // Write where this handle believes the end is, exactly as a positional
        // write would: over existing bytes if the belief is stale.
        const at: usize = @intCast(self.end);
        const need = at + bytes.len;
        if (need > s.bytes.items.len) {
            const old = s.bytes.items.len;
            s.bytes.resize(self.gpa, need) catch return error.Io;
            if (at > old) @memset(s.bytes.items[old..at], 0);
        }
        @memcpy(s.bytes.items[at..need], bytes);
        self.end += bytes.len;
    }

    fn preadImpl(ctx: *anyopaque, offset: u64, buf: []u8) storage.StorageError!usize {
        const self: *Handle = @ptrCast(@alignCast(ctx));
        const items = self.shared.bytes.items;
        if (offset >= items.len) return 0;
        const n = @min(buf.len, items.len - @as(usize, @intCast(offset)));
        @memcpy(buf[0..n], items[@intCast(offset)..][0..n]);
        return n;
    }

    fn syncImpl(_: *anyopaque) storage.StorageError!void {}

    fn sizeImpl(ctx: *anyopaque) storage.StorageError!u64 {
        const self: *Handle = @ptrCast(@alignCast(ctx));
        return self.end;
    }

    fn truncateImpl(ctx: *anyopaque, len: u64) storage.StorageError!void {
        const self: *Handle = @ptrCast(@alignCast(ctx));
        const s = self.shared;
        if (s.owner != self.id) s.unlocked_writes += 1;
        const want: usize = @intCast(len);
        const old = s.bytes.items.len;
        s.bytes.resize(self.gpa, want) catch return error.Io;
        if (want > old) @memset(s.bytes.items[old..want], 0);
        self.end = len;
    }

    /// Free or already ours: take it and re-read the true end. Held by another
    /// writer: fail, which `LogStore.open` reports as a storage error and the
    /// sweep treats as "blocked, try this writer again later". A real `flock`
    /// would wait instead; the interleaving is the same.
    fn lockImpl(ctx: *anyopaque) storage.StorageError!void {
        const self: *Handle = @ptrCast(@alignCast(ctx));
        const s = self.shared;
        if (s.owner) |o| if (o != self.id) return error.Io;
        s.owner = self.id;
        self.end = s.bytes.items.len;
    }
};

const Artifact = struct { hash: [hash_len]u8, bytes: []u8 };

const Writer = struct {
    id: u32,
    state: enum { idle, handle_made, open } = .idle,
    handle: Handle = undefined,
    log: logstore.LogStore = undefined,
    to_put: u32 = 0,
    /// Put this session, not yet committed: not owed back until the commit.
    session: std.ArrayList(Artifact) = .empty,
};

/// Run one seed: `writers` interleaved writers, `steps` scheduler steps.
pub fn run(gpa: std.mem.Allocator, seed: u64, steps: usize) LockSweepError!void {
    var prng = std.Random.DefaultPrng.init(seed ^ 0x6c6f636b73776565); // "locksweep"
    const random = prng.random();

    var shared: Shared = .{};
    defer shared.deinit(gpa);

    var committed: std.ArrayList(Artifact) = .empty;
    defer {
        for (committed.items) |a| gpa.free(a.bytes);
        committed.deinit(gpa);
    }

    const n_writers: u32 = 2 + random.uintLessThan(u32, 3); // 2..4
    var writers: [4]Writer = undefined;
    for (writers[0..n_writers], 0..) |*w, i| w.* = .{ .id = @intCast(i + 1) };
    defer for (writers[0..n_writers]) |*w| {
        for (w.session.items) |a| gpa.free(a.bytes);
        w.session.deinit(gpa);
        if (w.state == .open) {
            w.log.deinit(gpa);
            w.handle.close();
        }
    };

    var step: usize = 0;
    while (step < steps) : (step += 1) {
        const w = &writers[random.uintLessThan(u32, n_writers)];
        switch (w.state) {
            .idle => {
                // The stat: whatever the file's length is right now, which may
                // be stale by the time this writer takes the lock.
                w.handle = .{ .gpa = gpa, .shared = &shared, .id = w.id, .end = shared.bytes.items.len };
                w.state = .handle_made;
            },
            .handle_made => {
                if (logstore.LogStore.open(gpa, w.handle.storageFile(), 0)) |log| {
                    w.log = log;
                    w.to_put = 1 + random.uintLessThan(u32, 4);
                    w.state = .open;
                    // Everything committed before the lock was taken is visible.
                    for (committed.items) |a| if (w.log.lookup(&a.hash) == null) return error.StaleView;
                } else |e| switch (e) {
                    // Blocked on another writer's lock: try again on a later step.
                    error.Storage => if (shared.owner != null and shared.owner.? != w.id) {} else return error.LockSweepCorrupt,
                    error.OutOfMemory => return error.OutOfMemory,
                    else => return error.LockSweepCorrupt,
                }
            },
            .open => {
                if (w.to_put > 0) {
                    // Half the time, a small edit of an artifact another
                    // writer committed, stored as a delta against it: its
                    // basis offset is only right if this writer's end is.
                    const base: ?Artifact = if (committed.items.len > 0 and random.boolean())
                        committed.items[random.uintLessThan(usize, committed.items.len)]
                    else
                        null;
                    const bytes = if (base) |b| blk: {
                        const edited = try gpa.alloc(u8, b.bytes.len + 8);
                        @memcpy(edited[0..b.bytes.len], b.bytes);
                        random.bytes(edited[b.bytes.len..]);
                        break :blk edited;
                    } else blk: {
                        const fresh = try gpa.alloc(u8, 1 + random.uintLessThan(usize, 200));
                        random.bytes(fresh);
                        break :blk fresh;
                    };
                    errdefer gpa.free(bytes);
                    var hash: [hash_len]u8 = undefined;
                    std.crypto.hash.sha2.Sha256.hash(bytes, &hash, .{});
                    const base_rid: ?logstore.Rid = if (base) |b| w.log.lookup(&b.hash) else null;
                    const rid = w.log.putArtifact(&hash, bytes, base_rid) catch return error.LockSweepCorrupt;
                    // What was just put reads back through this writer's handle.
                    const got = (w.log.content(gpa, rid) catch return error.StaleView) orelse return error.StaleView;
                    defer gpa.free(got);
                    if (!std.mem.eql(u8, got, bytes)) return error.StaleView;
                    try w.session.append(gpa, .{ .hash = hash, .bytes = bytes });
                    w.to_put -= 1;
                } else {
                    w.log.commit() catch return error.LockSweepCorrupt;
                    try committed.appendSlice(gpa, w.session.items);
                    w.session.clearRetainingCapacity();
                    w.log.deinit(gpa);
                    w.handle.close();
                    w.state = .idle;
                }
            },
        }
        // The protocol is checked as it is broken, not only at the end: the
        // first bad write is the one worth reporting.
        if (shared.unlocked_writes != 0) return error.UnlockedWrite;
        if (shared.overwrites != 0) return error.OverwroteFrames;
    }

    try checkCommitted(gpa, &shared, committed.items);
}

/// Every committed artifact is present and byte-identical through a fresh
/// reader that holds no lock, the way `zit log` or `verify` would read it.
fn checkCommitted(gpa: std.mem.Allocator, shared: *Shared, committed: []const Artifact) LockSweepError!void {
    if (shared.bytes.items.len == 0) {
        if (committed.len == 0) return;
        return error.MissingAcknowledged;
    }
    var reader: Handle = .{ .gpa = gpa, .shared = shared, .id = 0, .end = shared.bytes.items.len };
    var log = logstore.LogStore.openReadOnly(gpa, reader.storageFile()) catch return error.LockSweepCorrupt;
    defer log.deinit(gpa);
    for (committed) |a| {
        const rid = log.lookup(&a.hash) orelse return error.MissingAcknowledged;
        const got = (log.content(gpa, rid) catch return error.LockSweepCorrupt) orelse return error.MissingAcknowledged;
        defer gpa.free(got);
        if (!std.mem.eql(u8, got, a.bytes)) return error.MissingAcknowledged;
    }
}

test "lock sweep: interleaved writers never write unlocked, never overwrite, and lose nothing" {
    var seed: u64 = 1;
    while (seed <= 25) : (seed += 1) try run(std.testing.allocator, seed, 400);
}