//! The staged-publication pipeline: a converter writes its
//! output store into a nonce-named staging directory beside the destination,
//! advances through a one-way Stage transaction (write, freeze, verify,
//! discard the working files, 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,
// The mirror store: the working log that holds blobs written at a
// mirror-candidate path until the drop set is final. Opened beside the
// published store, frozen with it, and never published.
create_mirror_store,
before_sync_mirror_store,
after_sync_mirror_store,
before_close_mirror_store,
after_close_mirror_store,
// The drop set is captured from the live importer, before the freeze
// that invalidates it, and the mirror store's survivors are copied
// into the published store under it.
capture_drop_set,
copy_mirror_survivors,
// The mirror store leaves staging between verification and the
// rename; failing on either side of that must still leave DEST absent.
before_discard_mirror_store,
after_discard_mirror_store,
before_sync_temporary_directory,
after_sync_temporary_directory,
before_rename_no_replace,
after_rename_no_replace,
verify_mirror_store,
verify_store,
verify_map,
sync_parent_after_publish,
// An import that adds to a destination: before the appended frames
// are committed, and after they are but before the map is written.
before_into_commit,
before_into_map,
into_before_prepare,
into_before_prepare_sync,
into_after_prepare_rename,
into_after_prepare,
into_before_recovery_sync,
into_after_recovery_sync,
into_before_append,
into_append_chunk,
into_before_live_sync,
into_after_live_sync,
into_before_map_replace,
into_after_map_replace,
into_after_map_sync,
into_after_retire,
into_after_retire_sync,
// 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,
};
/// Test-only seam for changing staging while the import verifier is
/// reading the map a window at a time.
pub const MapReadObserver = struct {
context: *anyopaque,
read: *const fn (context: *anyopaque) void,
};
fail_at: ?FailurePoint = null,
observer: ?Observer = null,
staging_observer: ?StagingObserver = null,
map_read_observer: ?MapReadObserver = 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);
}
};
/// The mirror store: blobs written at a mirror-candidate path, held apart
/// until the drop set is final. A working file of the run, never published.
pub const mirror_store_name = ".zit-import-mirror";
/// The log the converter writes, 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:
/// verification may not read a store still being written, 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 four members
/// would turn the fifth 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 = 0,
frozen = 1,
verified = 2,
published = 3,
};
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,
/// Whether the mirror store 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.
mirror_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
@fromBackingInt(@intCast(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 (@backingInt(to) != @backingInt(self.stage) + 1) return error.PublicationStageOutOfOrder;
self.stage = to;
}
/// The published store, written in place: the file the rename ships is
/// the file the converter writes, so nothing is ever rewritten into it.
pub fn createPublishedGeneration(self: *PublicationBundle) !std.Io.File {
if (self.stage != .writing) return error.PublicationStageOutOfOrder;
return self.temporary.createFile(self.io, published_generation_name, .{
.read = true,
.exclusive = true,
});
}
pub fn createMirrorStore(self: *PublicationBundle) !std.Io.File {
if (self.stage != .writing) return error.PublicationStageOutOfOrder;
const file = try self.temporary.createFile(self.io, mirror_store_name, .{
.read = true,
.exclusive = true,
});
self.mirror_present = true;
return file;
}
/// Remove the mirror store from staging. The rename publishes the staging
/// DIRECTORY whole, so this is what keeps it out of `DEST` — not a cleanup
/// nicety but the last step of the transaction's contract. It runs only
/// after verification has read it, and it re-checks staging identity
/// first, so a swapped sibling is never written through. Idempotent: a
/// second call, or a call on a run that created none, is a no-op.
pub fn discardMirrorStore(self: *PublicationBundle) !void {
if (self.stage != .verified) return error.PublicationStageOutOfOrder;
if (!self.mirror_present) return;
var current = try self.openVerifiedStaging();
defer current.close(self.io);
current.deleteFile(self.io, mirror_store_name) catch |err| switch (err) {
error.FileNotFound => {},
else => return err,
};
self.mirror_present = false;
}
/// Remove the instrumentation pair from staging once the verifier is done
/// with them, at the same stage the mirror store 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,
};
/// Published when the converter wrote it, never required: the Git
/// importer's commit map, which lets a later import add to the
/// destination. Published rather than instrumentation because without it
/// the destination can only ever be re-imported from scratch.
pub const optional_publish_manifest = [_][]const u8{
@import("git/git_import_map.zig").map_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
/// `discardMirrorStore` proves the mirror store 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 mirror-store
/// 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 ++ optional_publish_manifest)
else
&(publish_manifest ++ optional_publish_manifest);
const required = expected.len - optional_publish_manifest.len;
var seen: [publish_manifest.len + instrumentation_manifest.len + optional_publish_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..required]) |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 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 the mirror store already left
// staging. The rename publishes the staging DIRECTORY whole, so a
// mirror store 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.mirror_present) return error.PublicationMirrorStorePresent;
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.allocSentinel(u8, self.temporary_name.len, 0);
@memcpy(from, self.temporary_name);
defer self.gpa.free(from);
const to = try self.gpa.allocSentinel(u8, destination.len, 0);
@memcpy(to, 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;
}
};