//! The staged-publication pipeline (#61 extraction): a converter writes its
//! output store into a nonce-named staging directory beside the destination,
//! advances through a one-way Stage transaction (write, pack, verify, discard
//! the raw generation, rename), and the final rename is the sole commit
//! point -- a crash at any earlier stage leaves a sweepable `.zit-tmp-*` and
//! an untouched destination. Extracted verbatim from `git/git_import_cli.zig`
//! so `import-git` and future history converters (import-fossil) share one
//! publication discipline; that file re-exports these names, so its tests
//! and callers are unchanged.
const std = @import("std");
const builtin = @import("builtin");
extern "c" fn renameatx_np(
from_fd: c_int,
from: [*:0]const u8,
to_fd: c_int,
to: [*:0]const u8,
flags: c_uint,
) c_int;
const checkout = @import("checkout.zig");
pub const PublicationHooks = struct {
pub const FailurePoint = enum {
create_store,
before_flush_store,
after_flush_store,
before_sync_store,
after_sync_store,
before_close_store,
after_close_store,
before_flush_map,
after_flush_map,
before_sync_map,
after_sync_map,
before_close_map,
after_close_map,
// Generation 2. `create_store` above opens generation 1; these open,
// write, and freeze the packed generation that is actually published.
create_packed_store,
before_pack,
after_pack,
before_sync_packed_store,
after_sync_packed_store,
before_close_packed_store,
after_close_packed_store,
// The pack filter is captured from the live importer, before the freeze
// that invalidates it -- the earliest step that is packing's and not
// generation 1's, and so its own boundary.
capture_drop_set,
// Generation 1 leaves staging between verification and the rename;
// failing on either side of that must still leave DEST absent.
before_discard_raw_generation,
after_discard_raw_generation,
before_sync_temporary_directory,
after_sync_temporary_directory,
before_rename_no_replace,
after_rename_no_replace,
verify_raw_store,
verify_store,
verify_map,
sync_parent_after_publish,
// Compatibility selectors retain the former one-sided fault API.
flush_store,
sync_store,
close_store,
flush_map,
sync_map,
close_map,
sync_temporary_directory,
rename_no_replace,
};
pub const Observer = struct {
context: *anyopaque,
notify: *const fn (context: *anyopaque, point: FailurePoint) void,
};
/// Test-only seam for replacing the staging sibling after it was opened but
/// before its identity is checked at the publication boundary.
pub const StagingObserver = struct {
context: *anyopaque,
replace: *const fn (context: *anyopaque, parent: std.Io.Dir, name: []const u8) void,
};
fail_at: ?FailurePoint = null,
observer: ?Observer = null,
staging_observer: ?StagingObserver = null,
pub const real_filesystem: PublicationHooks = .{};
pub fn check(self: PublicationHooks, point: FailurePoint) !void {
const legacy: ?FailurePoint = switch (point) {
.before_flush_store => .flush_store,
.before_sync_store => .sync_store,
.before_close_store => .close_store,
.before_flush_map => .flush_map,
.before_sync_map => .sync_map,
.before_close_map => .close_map,
.before_sync_temporary_directory => .sync_temporary_directory,
.before_rename_no_replace => .rename_no_replace,
else => null,
};
if (self.fail_at == point or (legacy != null and self.fail_at == legacy)) return error.InjectedPublicationFailure;
if (self.observer) |observer| observer.notify(observer.context, point);
}
fn observeStaging(self: PublicationHooks, parent: std.Io.Dir, name: []const u8) void {
if (self.staging_observer) |observer| observer.replace(observer.context, parent, name);
}
};
pub const raw_generation_name = ".zit-import-raw";
/// Generation 2: the packed log, and the only store any import publishes.
pub const published_generation_name = ".zit";
pub const PublicationBundle = struct {
/// The one-way transaction. Every step is a durability boundary the
/// previous step has already crossed, so the order is the guarantee: a
/// pack may not read a store still being written, verification may not read
/// a store still being packed, and the rename may not commit a store
/// nothing verified. `advance` admits exactly one successor, so a future
/// edit that reorders the body fails at run time rather than publishing
/// something unverified.
/// Width is deliberate. `advance` computes `@intFromEnum(stage) + 1` in
/// the tag type, so a tag exactly wide enough for today's six members
/// would turn the seventh into a `ReleaseSafe` overflow PANIC instead of
/// the `PublicationStageOutOfOrder` this returns -- a checked error
/// silently becoming a crash on a future edit.
pub const Stage = enum(u8) {
writing_raw = 0,
raw_frozen = 1,
packing = 2,
packed_frozen = 3,
verified = 4,
published = 5,
};
gpa: std.mem.Allocator,
io: std.Io,
parent: std.Io.Dir,
temporary: std.Io.Dir,
temporary_inode: std.Io.File.INode,
temporary_name: []u8,
owned: bool = true,
stage: Stage = .writing_raw,
/// Whether generation 1 is still a file inside staging. Tracked so the
/// discard is idempotent and so nothing has to re-derive "is it there?"
/// from the filesystem at the one point where the answer decides what the
/// rename ships.
raw_present: bool = false,
/// Set once, before staging fills: which manifest `verifyStagingManifest`
/// holds the rename to, and whether `discardInstrumentation` runs.
publish_instrumentation: bool = false,
pub fn create(gpa: std.mem.Allocator, io: std.Io, parent: std.Io.Dir, destination: []const u8) !PublicationBundle {
var writable_parent = try parent.openDir(io, ".", .{
.access_sub_paths = true,
.follow_symlinks = false,
});
errdefer writable_parent.close(io);
const private_permissions: std.Io.Dir.Permissions = if (comptime builtin.os.tag == .windows or builtin.os.tag == .wasi)
.default_dir
else
@enumFromInt(0o700);
var attempt: usize = 0;
while (attempt != 1024) : (attempt += 1) {
var nonce: [16]u8 = undefined;
try io.randomSecure(&nonce);
const encoded_nonce = std.fmt.bytesToHex(nonce[0..], .lower);
const temporary_name = try std.fmt.allocPrint(gpa, ".zit-tmp-import-{s}", .{encoded_nonce[0..]});
errdefer gpa.free(temporary_name);
writable_parent.createDir(io, temporary_name, private_permissions) catch |err| switch (err) {
error.PathAlreadyExists => {
gpa.free(temporary_name);
continue;
},
else => return err,
};
var temporary: ?std.Io.Dir = null;
var temporary_inode: ?std.Io.File.INode = null;
errdefer {
if (temporary) |directory| directory.close(io);
cleanupCreatedStaging(io, writable_parent, temporary_name, temporary_inode);
}
temporary = try writable_parent.openDir(io, temporary_name, .{
.access_sub_paths = true,
.follow_symlinks = false,
});
const temporary_stat = try temporary.?.stat(io);
temporary_inode = temporary_stat.inode;
if (temporary_stat.kind != .directory) return error.StagingEntryReplaced;
// `destination` is intentionally part of this constructor's shape:
// ownership is established only after the caller performed the
// destination preflight, and publication has one atomic commit point.
_ = destination;
const result: PublicationBundle = .{
.gpa = gpa,
.io = io,
.parent = writable_parent,
.temporary = temporary.?,
.temporary_inode = temporary_stat.inode,
.temporary_name = temporary_name,
};
temporary = null;
return result;
}
return error.NoTemporaryName;
}
/// The transaction's only forward edge. A skipped, repeated, or reversed
/// stage is a CHECKED error rather than an assert: it gates an atomic
/// rename that runs in `ReleaseSafe` and `ReleaseFast` alike, where an
/// assert is worth nothing.
pub fn advance(self: *PublicationBundle, to: Stage) !void {
if (@intFromEnum(to) != @intFromEnum(self.stage) + 1) return error.PublicationStageOutOfOrder;
self.stage = to;
}
pub fn createRawGeneration(self: *PublicationBundle) !std.Io.File {
if (self.stage != .writing_raw) return error.PublicationStageOutOfOrder;
const file = try self.temporary.createFile(self.io, raw_generation_name, .{
.read = true,
.exclusive = true,
});
self.raw_present = true;
return file;
}
pub fn createPackedGeneration(self: *PublicationBundle) !std.Io.File {
if (self.stage != .packing) return error.PublicationStageOutOfOrder;
return self.temporary.createFile(self.io, published_generation_name, .{
.read = true,
.exclusive = true,
});
}
/// Remove generation 1 from staging. The rename publishes the staging
/// DIRECTORY whole, so this is what keeps the raw log out of `DEST` — not a
/// cleanup nicety but the last step of the transaction's contract. It runs
/// only after verification has read both generations, and it re-checks
/// staging identity first, so a swapped sibling is never written through.
/// Idempotent: a second call on an already-discarded generation is a no-op.
pub fn discardRawGeneration(self: *PublicationBundle) !void {
if (self.stage != .verified) return error.PublicationStageOutOfOrder;
if (!self.raw_present) return;
var current = try self.openVerifiedStaging();
defer current.close(self.io);
current.deleteFile(self.io, raw_generation_name) catch |err| switch (err) {
error.FileNotFound => {},
else => return err,
};
self.raw_present = false;
}
/// Remove the instrumentation pair from staging once the verifier is done
/// with them, at the same stage the raw generation leaves and for the same
/// reason: the rename ships the directory whole, and without `--provenance`
/// these are working files of the run, not payload of the repository.
pub fn discardInstrumentation(self: *PublicationBundle) !void {
if (self.stage != .verified) return error.PublicationStageOutOfOrder;
if (self.publish_instrumentation) return;
var current = try self.openVerifiedStaging();
defer current.close(self.io);
for (instrumentation_manifest) |name| {
current.deleteFile(self.io, name) catch |err| switch (err) {
error.FileNotFound => {},
// `refs` is a directory tree.
error.IsDir => current.deleteTree(self.io, name) catch return error.Storage,
else => return err,
};
}
}
/// Exactly what a published import contains. Named here rather than left
/// implicit in the write order because the rename ships the staging
/// directory WHOLE: this list is the publication's payload, and anything
/// beside it is published by accident.
///
/// The default is the repository and its checkout record, full stop --
/// fossil's own shape. The instrumentation pair is appended only under
/// `--provenance`; both are always WRITTEN and verified in staging either
/// way, because the in-run verifier reads them before publication.
pub const publish_manifest = [_][]const u8{
published_generation_name,
checkout.filename,
};
pub const instrumentation_manifest = [_][]const u8{
".zit-import-map.jsonl",
"refs",
};
/// The last checked statement before the commit point: staging holds the
/// publish manifest and nothing else.
///
/// The stage machine proves the ORDER of the transaction, and
/// `discardRawGeneration` proves generation 1 was asked to leave. Neither
/// proves what is actually on disk at the instant the directory is synced
/// and renamed, and that is the only fact the rename acts on. One
/// `readdir` per import buys it directly -- and it cannot be outlived by a
/// later edit that adds another intermediate file, which a raw-generation
/// check alone could be.
pub fn verifyStagingManifest(self: *PublicationBundle) !void {
var current = try self.openVerifiedStaging();
defer current.close(self.io);
const expected: []const []const u8 = if (self.publish_instrumentation)
&(publish_manifest ++ instrumentation_manifest)
else
&publish_manifest;
var seen: [publish_manifest.len + instrumentation_manifest.len]bool = @splat(false);
var entries = current.iterate();
while (try entries.next(self.io)) |entry| {
const index = for (expected, 0..) |name, position| {
if (std.mem.eql(u8, entry.name, name)) break position;
} else return error.UnpublishableStagingEntry;
// A repeat is impossible from a directory read; treating it as one
// more unexpected entry keeps this total rather than assuming so.
if (seen[index]) return error.UnpublishableStagingEntry;
seen[index] = true;
}
for (seen[0..expected.len]) |present| {
if (!present) return error.IncompletePublication;
}
}
pub fn abort(self: *PublicationBundle) void {
if (self.stage != .published and self.owned) {
var current = self.openVerifiedStaging() catch null;
if (current) |*directory| {
const emptied = deleteOwnedContents(directory, self.io);
directory.close(self.io);
if (emptied) {
var rechecked = self.openVerifiedStaging() catch null;
if (rechecked) |*verified| {
verified.close(self.io);
self.parent.deleteDir(self.io, self.temporary_name) catch {};
}
}
}
}
self.temporary.close(self.io);
self.parent.close(self.io);
self.gpa.free(self.temporary_name);
self.* = undefined;
}
pub fn publish(self: *PublicationBundle, destination: []const u8, hooks: PublicationHooks) !void {
// The commit point admits exactly one predecessor. Reaching it from any
// earlier stage would publish a store that was never packed, never
// frozen, or never verified -- and after the rename there is nothing
// left to unwind.
if (self.stage != .verified) return error.PublicationNotVerified;
// `.verified` is BOTH the pre-discard and post-discard stage, so the
// stage alone does not say whether generation 1 already left staging.
// The rename publishes the staging DIRECTORY whole, so a raw log still
// sitting in it would ship into DEST; without this the only thing
// stopping that is statement order in `importInto`, which is exactly
// the reliance the stage machine exists to remove.
if (self.raw_present) return error.PublicationRawGenerationPresent;
if (!self.owned) return error.StagingEntryReplaced;
try hooks.check(.before_rename_no_replace);
hooks.observeStaging(self.parent, self.temporary_name);
if (comptime builtin.os.tag == .macos) {
const from = try self.gpa.dupeZ(u8, self.temporary_name);
defer self.gpa.free(from);
const to = try self.gpa.dupeZ(u8, destination);
defer self.gpa.free(to);
var current = try self.openVerifiedStaging();
defer current.close(self.io);
// Darwin's standard `renamePreserve` emulates no-replace with a
// hard link, which cannot publish a directory. `RENAME_EXCL`
// supplies the required atomic directory commit point instead.
if (renameatx_np(self.parent.handle, from.ptr, self.parent.handle, to.ptr, 0x0000_0004) != 0) {
return error.RenameNoReplaceFailed;
}
} else {
var current = try self.openVerifiedStaging();
defer current.close(self.io);
try self.parent.renamePreserve(self.temporary_name, self.parent, destination, self.io);
}
self.owned = false;
try hooks.check(.after_rename_no_replace);
var published = self.parent.openDir(self.io, destination, .{
.access_sub_paths = true,
.follow_symlinks = false,
}) catch return error.PublishedDestinationReplaced;
defer published.close(self.io);
const published_stat = published.stat(self.io) catch return error.PublishedDestinationReplaced;
if (published_stat.kind != .directory or published_stat.inode != self.temporary_inode) {
return error.PublishedDestinationReplaced;
}
// Through `advance` like every other edge. A direct assignment here
// would leave the terminal transition -- the one that disarms `abort`
// -- as the single in-module bypass of the machine.
try self.advance(.published);
}
fn openVerifiedStaging(self: *PublicationBundle) !std.Io.Dir {
if (!self.owned) return error.StagingEntryReplaced;
const held = self.temporary.stat(self.io) catch {
self.owned = false;
return error.StagingEntryReplaced;
};
if (held.kind != .directory or held.inode != self.temporary_inode) {
self.owned = false;
return error.StagingEntryReplaced;
}
var current = self.parent.openDir(self.io, self.temporary_name, .{
.access_sub_paths = true,
.follow_symlinks = false,
// `verifyStagingManifest` iterates this handle. Without the flag
// Linux opens an O_PATH-style descriptor that getdents rejects
// (BADF at runtime); macOS tolerates the omission, which is how
// it survived until the first Linux CI run.
.iterate = true,
}) catch {
self.owned = false;
return error.StagingEntryReplaced;
};
errdefer current.close(self.io);
const entry = current.stat(self.io) catch {
self.owned = false;
return error.StagingEntryReplaced;
};
if (entry.kind != .directory or entry.inode != self.temporary_inode) {
self.owned = false;
return error.StagingEntryReplaced;
}
return current;
}
fn cleanupCreatedStaging(
io: std.Io,
parent: std.Io.Dir,
name: []const u8,
expected_inode: ?std.Io.File.INode,
) void {
const inode = expected_inode orelse return;
var current = parent.openDir(io, name, .{
.access_sub_paths = true,
.follow_symlinks = false,
// `deleteOwnedContents` iterates this handle (Linux enforces
// iterate rights at readdir time; macOS does not).
.iterate = true,
}) catch return;
defer current.close(io);
const entry = current.stat(io) catch return;
if (entry.kind != .directory or entry.inode != inode) return;
if (!deleteOwnedContents(¤t, io)) return;
parent.deleteDir(io, name) catch {};
}
fn deleteOwnedContents(directory: *std.Io.Dir, io: std.Io) bool {
var entries = directory.iterate();
while (entries.next(io) catch return false) |entry| {
directory.deleteTree(io, entry.name) catch return false;
}
return true;
}
};