Artifact [6a6562e06e]

Artifact 6a6562e06e3f2a1d3bd1e86fa54bbafba8fb3c4a377b415b07beb72ba5914677:

• File src/cmd/purge.zig — part of check-in [5c445bb3f3] at 2026-09-28 02:39:03 on branch trunk — Take the writer lock inside LogStore.open and test the lock class

  LogStore.open now takes the exclusive lock before it reads the file's length and scans for the end, and a read-only handle refuses to append. purge undo opens the store through the shared writer path.

  A new DST sweep runs two to four writers interleaved against one log through a file that models the lock. It flags unlocked or misplaced appends as they happen, checks each writer sees everything committed before it locked and reads back its own writes, including deltas. A source check fails on any write-mode open of the store outside openStoreAppend. An e2e test races commits, tags and user-table writes on one store and counts what landed. (user: dmestas, size: 16968) [annotate] [blame] [check-ins using]

//! `zit purge`: remove check-ins, keep them recoverable.
//!
//! The decision lives in `purge.zig` and the recovery store in `graveyard.zig`;
//! this file is the verb around them. It resolves the operator's names, renders
//! the plan, and — only under `--force` — buries the artifacts and hands the
//! drop set to the same `rewriteAndSwap` that `zit shun --force` uses.
//!
//! ORDER OF THE TWO DURABLE WRITES: the graveyard is written FIRST, then the
//! store is rewritten. A crash between them leaves a graveyard event whose
//! artifacts are still in the store — `purge undo` restores what is already
//! there, which is a no-op, and `purge obliterate` discards a copy of live
//! content. Both are harmless. The other order is not: a store rewritten before
//! its content was buried loses the content outright on a crash, which is
//! exactly the failure the graveyard exists to prevent.

const std = @import("std");
const zit = @import("zit");

const cli = zit.cli;
const graveyard = zit.graveyard;
const logstore = zit.logstore;
const pack = zit.pack;
const purge = zit.purge;
const storage = zit.storage;
const main = @import("../main.zig");
const cmd_sync = @import("sync.zig");
const Repository = zit.Repository;

const repo_filename = main.repo_filename;

pub fn runPurge(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    verb_args: []const []const u8,
    streams: cli.Streams,
    amb: cli.Ambient,
) cli.Exit {
    const parsed = switch (cli.parsePurgeArgs(verb_args)) {
        .ok => |p| p,
        .unknown_option => |tok| return cli.usageError(streams, "purge", .unknown_option, tok),
        .extra_argument => |tok| return cli.usageError(streams, "purge", .extra_argument, tok),
        .bad_subcommand => |tok| return cli.usageError(streams, "purge", .bad_subcommand, tok),
        .bad_reason => |tok| return cli.usageError(streams, "purge", .unknown_option, tok),
        .no_name => return cli.usageError(streams, "purge", .no_name, ""),
        .option_not_for_subcommand => |tok| return cli.usageError(streams, "purge", .option_not_for_subcommand, tok),
    };
    if (parsed.help) {
        streams.out.writeAll(main.purge_help) catch {};
        return .ok;
    }

    // Every subcommand needs the store directory; only `checkins` needs name
    // resolution, so the repository open that costs a derive is deferred into
    // that arm rather than paid by `purge list`.
    const root = (cli.resolveRoot(arena, io, amb.repo_override) catch
        return oom(streams)) orelse {
        streams.err.writeAll(main.not_a_repo_msg) catch {};
        return .failure;
    };
    var root_dir = root.dir;
    defer root_dir.close(io);
    // Worktrees: the graveyard and the rewrite both belong beside the STORE, so a
    // detached worktree resolves its pointer once, here.
    if (cli.storeDirOf(io, root_dir) catch return readFail(streams)) |resolved| {
        root_dir.close(io);
        root_dir = resolved;
    }

    return switch (parsed.sub) {
        .list => runList(gpa, io, arena, root_dir, streams),
        .undo => runUndo(gpa, io, arena, root_dir, parsed, streams),
        .obliterate => runObliterate(gpa, io, arena, root_dir, parsed, streams),
        .checkins => runCheckins(gpa, io, arena, root_dir, parsed, streams, amb),
    };
}

fn runList(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    root_dir: std.Io.Dir,
    streams: cli.Streams,
) cli.Exit {
    const events = graveyard.read(arena, io, root_dir) catch |e| return graveFail(streams, e);
    // Data only, one line per event, newest last — the order they were buried
    // in, which is the order an operator undoes them in.
    for (events) |ev| {
        streams.out.print("{d} {d} {d} {d} {s}\n", .{
            ev.id,
            ev.when,
            ev.artifacts.len,
            ev.size(),
            ev.reason,
        }) catch {};
    }
    _ = gpa;
    return .ok;
}

fn runUndo(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    root_dir: std.Io.Dir,
    parsed: cli.PurgeArgs,
    streams: cli.Streams,
) cli.Exit {
    if (parsed.name_count != 1) {
        streams.err.writeAll("zit purge undo: name exactly one event id\n") catch {};
        return .failure;
    }
    const want = std.fmt.parseInt(u32, parsed.names[0], 10) catch {
        streams.err.print("zit purge undo: '{s}' is not an event id\n", .{parsed.names[0]}) catch {};
        return .failure;
    };
    const events = graveyard.read(arena, io, root_dir) catch |e| return graveFail(streams, e);
    const at = indexOfEvent(events, want) orelse {
        streams.err.print("zit purge undo: no buried event {d}\n", .{want}) catch {};
        return .failure;
    };

    // Restoring is an ordinary append: the log is append-only, so putting the
    // artifacts back is the same operation that stored them the first time, and
    // the next open re-derives the check-in DAG from them. No rewrite runs.
    //
    // An append, so it takes the store's append lock like every other writer:
    // unserialized, a concurrent commit or pull scans the same end and one of
    // the two writes lands on top of the other with valid frames, which no
    // checksum can tell from a clean write. `openStoreAppend` also resolves a
    // detached worktree's store, and `LogStore.open` scans for the end only
    // once the lock is held; a reader's handle (`openReadOnly`) cannot append.
    // `hash_kind` 0 is a don't-care on a non-empty store: it is read from the
    // header.
    var file = cli.openStoreAppend(io, root_dir) catch
        return readFail(streams);
    defer file.close(io);
    const st = file.stat(io) catch return readFail(streams);
    var fs = storage.FileStorage.init(io, file, st.size);
    var log = logstore.LogStore.open(gpa, fs.storageFile(), 0) catch return readFail(streams);
    defer log.deinit(gpa);

    var restored: usize = 0;
    for (events[at].artifacts) |a| {
        // Already present: a crash between the two durable writes leaves
        // exactly this state, so it is a no-op rather than an error.
        if (log.lookup(a.hash) != null) continue;
        _ = log.putArtifact(a.hash, a.bytes, null) catch return writeFail(streams);
        restored += 1;
    }
    log.commit() catch return writeFail(streams);

    // The event is dropped only after its content is durably back.
    const kept = arena.alloc(graveyard.Event, events.len - 1) catch return oom(streams);
    var n: usize = 0;
    for (events, 0..) |ev, i| {
        if (i == at) continue;
        kept[n] = ev;
        n += 1;
    }
    graveyard.write(gpa, io, root_dir, kept) catch |e| return graveFail(streams, e);

    streams.out.print("restored: {d} artifact{s} from event {d}\n", .{
        restored,
        plural(restored),
        want,
    }) catch {};
    return .ok;
}

fn runObliterate(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    root_dir: std.Io.Dir,
    parsed: cli.PurgeArgs,
    streams: cli.Streams,
) cli.Exit {
    const events = graveyard.read(arena, io, root_dir) catch |e| return graveFail(streams, e);

    // Resolved before anything is discarded, so an unknown id refuses the whole
    // command rather than emptying the graves it did recognize.
    var drop_ids = arena.alloc(u32, parsed.name_count) catch return oom(streams);
    for (parsed.names[0..parsed.name_count], 0..) |tok, i| {
        drop_ids[i] = std.fmt.parseInt(u32, tok, 10) catch {
            streams.err.print("zit purge obliterate: '{s}' is not an event id\n", .{tok}) catch {};
            return .failure;
        };
        if (indexOfEvent(events, drop_ids[i]) == null) {
            streams.err.print("zit purge obliterate: no buried event {d}\n", .{drop_ids[i]}) catch {};
            return .failure;
        }
    }

    var bytes: u64 = 0;
    var kept: std.ArrayList(graveyard.Event) = .empty;
    for (events) |ev| {
        if (contains(drop_ids, ev.id)) {
            bytes += ev.size();
            continue;
        }
        kept.append(arena, ev) catch return oom(streams);
    }

    // Refuse-by-default, like the purge itself: this is the one step in the
    // whole verb that nothing can undo.
    if (!parsed.force) {
        streams.err.print(
            "zit purge obliterate: would discard {d} event{s} holding {d} bytes, unrecoverably\n" ++
                "nothing was discarded; pass --force to proceed\n",
            .{ drop_ids.len, plural(drop_ids.len), bytes },
        ) catch {};
        return .failure;
    }
    graveyard.write(gpa, io, root_dir, kept.items) catch |e| return graveFail(streams, e);
    streams.out.print("obliterated: {d} event{s}, {d} bytes\n", .{
        drop_ids.len,
        plural(drop_ids.len),
        bytes,
    }) catch {};
    return .ok;
}

fn runCheckins(
    gpa: std.mem.Allocator,
    io: std.Io,
    arena: std.mem.Allocator,
    root_dir: std.Io.Dir,
    parsed: cli.PurgeArgs,
    streams: cli.Streams,
    amb: cli.Ambient,
) cli.Exit {
    // Names resolve through the ordinary §8.3 procedure, so `zit purge
    // checkins tip` means what it means everywhere else. The repository is
    // opened for that alone and closed before the store is: the plan and the
    // rewrite work from a read-only open of the log, exactly as `shun` does,
    // and holding a derived state across a generation swap would leave it
    // describing a store that no longer exists.
    var digests = arena.alloc([]const u8, parsed.name_count) catch return oom(streams);
    var checkout_digest: ?[]const u8 = null;
    {
        var repo: Repository = undefined;
        var repo_dir: std.Io.Dir = undefined;
        const rc = main.openRepo(gpa, io, arena, streams, "purge", amb.repo_override, parsed.no_cache, false, &repo, &repo_dir);
        if (rc != .ok) return rc;
        defer repo_dir.close(io);
        defer repo.deinit(gpa);

        for (parsed.names[0..parsed.name_count], 0..) |name, i| {
            const rid = repo.resolve(name, .checkin) catch {
                streams.err.print("zit purge: no check-in named '{s}'\n", .{name}) catch {};
                return .failure;
            };
            digests[i] = arena.dupe(u8, repo.hashOf(rid)) catch return oom(streams);
        }
        if (repo.currentCheckoutRid()) |rid| {
            checkout_digest = arena.dupe(u8, repo.hashOf(rid)) catch return oom(streams);
        }
    }

    var file = root_dir.openFile(io, repo_filename, .{}) catch return readFail(streams);
    defer file.close(io);
    const st = file.stat(io) catch return readFail(streams);
    var fs = storage.FileStorage.init(io, file, st.size);
    var log = logstore.LogStore.openReadOnly(gpa, fs.storageFile()) catch return readFail(streams);
    defer log.deinit(gpa);
    const src_key = log.logKey() catch return readFail(streams);

    var targets = arena.alloc(purge.Target, parsed.name_count) catch return oom(streams);
    for (parsed.names[0..parsed.name_count], 0..) |name, i| {
        targets[i] = .{
            .name = name,
            // The repository view is keyed by lowercase-hex artifact NAME while
            // the log is keyed by binary digest, so the crossing goes through
            // the hex resolver rather than `lookup`. Resolved above through the
            // repository, so a name that got this far always has a rid here.
            .rid = (log.resolveHexPrefix(digests[i]) catch null) orelse {
                streams.err.print("zit purge: '{s}' names no artifact in this store\n", .{name}) catch {};
                return .failure;
            },
        };
    }
    const checkout_rid: ?logstore.Rid = if (checkout_digest) |d|
        (log.resolveHexPrefix(d) catch null)
    else
        null;

    const plan = (purge.plan(gpa, arena, &log, targets, checkout_rid) catch |e|
        return planFail(streams, e)) orelse {
        renderBlockers(streams.err, targets);
        return .failure;
    };

    renderPlan(streams.out, plan, parsed.explain, &log);
    if (parsed.dry_run) return .ok;
    if (!parsed.force) {
        streams.err.writeAll(
            "nothing was removed; pass --force to proceed\n" ++
                "a purge is recoverable with `zit purge undo` until it is obliterated\n",
        ) catch {};
        return .failure;
    }

    // Bury first — see the header note on the order of the two durable writes.
    var buried = arena.alloc(graveyard.Artifact, plan.drop.len) catch return oom(streams);
    for (plan.drop, 0..) |rid, i| {
        const bytes = (log.content(arena, rid) catch return readFail(streams)) orelse {
            streams.err.writeAll("zit purge: an artifact in the plan has no content to bury\n") catch {};
            return .failure;
        };
        buried[i] = .{
            .hash = arena.dupe(u8, log.hashOf(rid)) catch return oom(streams),
            .bytes = bytes,
        };
    }
    const events = graveyard.read(arena, io, root_dir) catch |e| return graveFail(streams, e);
    var all = arena.alloc(graveyard.Event, events.len + 1) catch return oom(streams);
    @memcpy(all[0..events.len], events);
    all[events.len] = .{
        .id = graveyard.nextId(events),
        .when = amb.now_epoch,
        .reason = parsed.reason orelse "",
        .artifacts = buried,
    };
    graveyard.write(gpa, io, root_dir, all) catch |e| return graveFail(streams, e);

    var set: pack.NameSet = .{};
    defer set.deinit(gpa);
    for (plan.drop) |rid| set.add(gpa, log.hashOf(rid)) catch |e| switch (e) {
        error.OutOfMemory => return oom(streams),
        // A stored artifact whose name is not a complete digest is a corrupt
        // store, and that is the only place this can come from.
        error.BadKey => return planFail(streams, error.Corrupt),
    };

    // No `extra_shuns`: a purge records no tombstone, which is exactly what
    // makes it recoverable and exactly why it is local. `shun` is the verb that
    // also stops the content coming back.
    const stats = cmd_sync.rewriteAndSwap(gpa, io, root_dir, &log, src_key, set.filter(), &.{}) catch |e| switch (e) {
        error.TailChanged => {
            streams.err.writeAll(
                "zit purge: the repository changed while it was being examined; nothing was removed\n",
            ) catch {};
            return .failure;
        },
        error.RewriteBusy => {
            streams.err.writeAll(
                "zit purge: another repository rewrite is in progress; nothing was removed\n",
            ) catch {};
            return .failure;
        },
        else => return planFail(streams, error.Storage),
    };
    streams.out.print("purged: {d} artifact{s}, {d} bytes reclaimed, event {d}\n", .{
        plan.drop.len,
        plural(plan.drop.len),
        stats.bytes_before -| stats.bytes_after,
        all[events.len].id,
    }) catch {};
    return .ok;
}

fn renderPlan(out: *std.Io.Writer, plan: purge.Plan, explain: bool, log: *logstore.LogStore) void {
    out.print("purge: {d} check-in{s}, {d} tag{s}, {d} other artifact{s}, {d} bytes\n", .{
        plan.checkins, plural(plan.checkins),
        plan.controls, plural(plan.controls),
        plan.others,   plural(plan.others),
        plan.bytes,
    }) catch {};
    if (!explain) return;
    for (plan.drop) |rid| {
        var buf: [2 * logstore.max_hash_len]u8 = undefined;
        out.print("  {s}\n", .{std.fmt.bufPrint(&buf, "{x}", .{log.hashOf(rid)}) catch continue}) catch {};
    }
}

fn renderBlockers(err: *std.Io.Writer, targets: []const purge.Target) void {
    err.writeAll("zit purge: refused; nothing was removed\n") catch {};
    for (targets) |t| {
        const b = t.blocker orelse continue;
        err.print("  {s}: {s}\n", .{ t.name, b.describe() }) catch {};
    }
}

fn indexOfEvent(events: []const graveyard.Event, id: u32) ?usize {
    for (events, 0..) |ev, i| if (ev.id == id) return i;
    return null;
}

fn contains(haystack: []const u32, needle: u32) bool {
    for (haystack) |h| if (h == needle) return true;
    return false;
}

fn plural(n: usize) []const u8 {
    return if (n == 1) "" else "s";
}

fn oom(streams: cli.Streams) cli.Exit {
    streams.err.writeAll("zit purge: out of memory\n") catch {};
    return .failure;
}

fn readFail(streams: cli.Streams) cli.Exit {
    streams.err.writeAll("zit purge: this repository could not be read\n") catch {};
    return .failure;
}

fn writeFail(streams: cli.Streams) cli.Exit {
    streams.err.writeAll("zit purge: this repository could not be written\n") catch {};
    return .failure;
}

fn graveFail(streams: cli.Streams, e: anyerror) cli.Exit {
    streams.err.print("zit purge: the graveyard could not be read or written ({s})\n", .{@errorName(e)}) catch {};
    return .failure;
}

fn planFail(streams: cli.Streams, e: anyerror) cli.Exit {
    streams.err.print("zit purge: the plan could not be computed ({s})\n", .{@errorName(e)}) catch {};
    return .failure;
}