Skip to content
zit Artifact [33d94c54d3]
Login

Artifact [33d94c54d3]

Check-ins Using Download Hex

Artifact 33d94c54d3f0a44152c5d483d7ff624db66f0a8df61b08cfd6f7a1ea3487fa3c:

//! The DST sweep for the writer lock: several writers append to ONE log,
//! interleaved step by step off the seed, through a simulated file and a
//! simulated lock file that follow the store's locking protocol.
//!
//! 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 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:
//!
//!   take the lock      `cli.openStoreAppend` creates `.zit.lock` exclusively
//!   stat               the length the caller builds its FileStorage from
//!   LogStore.open      re-reads the end, then scans
//!   put × k            appends, one per step
//!   commit, release    fsync, then the lock file is deleted
//!
//! so another writer can run between any two of them. A writer whose lock is
//! held by a live process is simply not advanced until it is free.
//!
//! A writer can also die at any step while it holds the lock, as a killed
//! process does. Its lock file stays behind naming it, and the next writer
//! must break it, because its holder no longer exists, and never break the
//! lock of one that does. The simulated file 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 whose process 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. Once it holds the lock it must 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, and its lock file.
const Shared = struct {
    bytes: std.ArrayList(u8) = .empty,
    /// The process named in `.zit.lock`, or null when there is none.
    lock_holder: ?u32 = null,
    /// Processes that have died. A lock naming one of them is stale.
    dead: std.ArrayList(u32) = .empty,
    unlocked_writes: u32 = 0,
    overwrites: u32 = 0,

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

    fn isDead(self: *const Shared, pid: u32) bool {
        return std.mem.indexOfScalar(u32, self.dead.items, pid) != null;
    }
};

/// One open handle on the shared file, with `FileStorage`'s shape: an append
/// watermark captured by the caller's `stat` and re-read by `refresh_end`.
const Handle = struct {
    gpa: std.mem.Allocator,
    shared: *Shared,
    pid: u32,
    end: u64,

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

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

    fn appendImpl(ctx: *anyopaque, bytes: []const u8) storage.StorageError!void {
        const self: *Handle = @ptrCast(@alignCast(ctx));
        const s = self.shared;
        if (s.lock_holder != self.pid) 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.lock_holder != self.pid) 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;
    }

    fn refreshEndImpl(ctx: *anyopaque) storage.StorageError!void {
        const self: *Handle = @ptrCast(@alignCast(ctx));
        self.end = self.shared.bytes.items.len;
    }
};

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

const Writer = struct {
    /// The process this writer currently is. A writer that dies comes back
    /// as a new process, as a restarted command does.
    pid: u32,
    state: enum { idle, locked, 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,

    fn dropSession(self: *Writer, gpa: std.mem.Allocator) void {
        for (self.session.items) |a| gpa.free(a.bytes);
        self.session.clearRetainingCapacity();
    }
};

/// One step of taking the lock, as `storelock.acquire` does it: create the
/// lock file if there is none; if its holder no longer exists, break it; if
/// its holder is alive, wait. Returns whether the lock is now `pid`'s.
fn tryLock(shared: *Shared, pid: u32) bool {
    if (shared.lock_holder) |holder| {
        if (!shared.isDead(holder)) return false;
        // Stale: break it. A later step for this writer takes it.
        shared.lock_holder = null;
        return false;
    }
    shared.lock_holder = pid;
    return true;
}

/// 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 next_pid: u32 = 1;
    var writers: [4]Writer = undefined;
    for (writers[0..n_writers]) |*w| {
        w.* = .{ .pid = next_pid };
        next_pid += 1;
    }
    defer for (writers[0..n_writers]) |*w| {
        w.dropSession(gpa);
        w.session.deinit(gpa);
        if (w.state == .open) w.log.deinit(gpa);
    };

    var step: usize = 0;
    while (step < steps) : (step += 1) {
        const w = &writers[random.uintLessThan(u32, n_writers)];

        // A writer holding the lock can die at any step. Its lock file stays,
        // naming a process that no longer exists; what it put but never
        // committed is not owed to anyone.
        if (w.state != .idle and random.uintLessThan(u32, 40) == 0) {
            if (w.state == .open) w.log.deinit(gpa);
            w.dropSession(gpa);
            try shared.dead.append(gpa, w.pid);
            w.pid = next_pid;
            next_pid += 1;
            w.state = .idle;
            continue;
        }

        switch (w.state) {
            .idle => {
                if (tryLock(&shared, w.pid)) w.state = .locked;
            },
            .locked => {
                // The caller's stat, under the lock.
                w.handle = .{ .gpa = gpa, .shared = &shared, .pid = w.pid, .end = shared.bytes.items.len };
                w.state = .handle_made;
            },
            .handle_made => {
                w.log = logstore.LogStore.open(gpa, w.handle.storageFile(), 0) catch |e| switch (e) {
                    error.OutOfMemory => return error.OutOfMemory,
                    else => return error.LockSweepCorrupt,
                };
                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;
            },
            .open => {
                if (w.to_put > 0) {
                    try putOne(gpa, random, w, committed.items);
                    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);
                    // Release: delete the lock file, if it is still ours.
                    if (shared.lock_holder == w.pid) shared.lock_holder = null;
                    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);
}

/// Put one artifact through `w`'s log and read it straight back. Half the
/// time it is 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.
fn putOne(gpa: std.mem.Allocator, random: std.Random, w: *Writer, committed: []const Artifact) LockSweepError!void {
    const base: ?Artifact = if (committed.len > 0 and random.boolean())
        committed[random.uintLessThan(usize, committed.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;
    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 });
}

/// 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, .pid = 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);
}