🬐 🬠 🬞🬂🬏 🬟 🬑 🬟 🬑 b4ae9279.. ╭b4ae9279..╮ see-em: 🬋🬋🬋 🬋🬋🬋 🬋 🬋 files │diff │ refs discuss issues new login 🬐 🬠 🬐 🬠 🬑 🬟 🬰 ╰──────────╯ user
╭ changed ────╮ │src/event.zig│ ╰─────────────╯ ╭────────────────────────────────────────────────────────────────────────────────────────────────╮ │428 // consuming an existing ref does not provide the set of changed patches │ │429 // explicit patch writes already saved their checks in the event transaction │ │430 if (role == .repo and repo_kind == .xit and host_kind == .server and events.len == 0) {│ │431 - pch.refreshMergeability(repo_opts, io, allocator, repo, null); │ │431 + pch.refreshOpenMergeability(repo_opts, io, allocator, repo, null); │ │432 } │ │433 } │ │434 │ ╰────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭ changed ────╮ │src/patch.zig│ ╰─────────────╯ ╭────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │44 } else { │ │45 const DB = evt.EventDB(repo_opts.hash); │ │46 │ │47 - // continue from a remote event history when the local ref is absent │ │48 - var first_parent: ?[1][hash.hexLen(repo_opts.hash)]u8 = null; │ │49 - if (try repo.readRef(io, evt.events_ref) == null) { │ │50 - var remotes = try repo.listRemotes(io, allocator); │ │51 - defer remotes.deinit(); │ │52 - for (remotes.sections.keys()) |name| { │ │53 - if (try repo.readRef(io, .{ .kind = .{ .remote = name }, .name = evt.events_ref.name })) |oid| {│ │54 - first_parent = .{oid}; │ │55 - break; │ │56 - } │ │57 - } │ │58 - } │ │47 + const first_parent = try eventsFirstParent(repo_kind, repo_opts, io, allocator, repo); │ │48 │ │49 // commit the patch, mergeability, and undo record together │ │50 { │ ╰────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │121 _ = try checks.update(state, io, arena.child_allocator); │ │122 } │ │123 │ │124 +// continue from a remote event history when the local ref is absent, so a │ │125 +// fresh clone's first event does not start a history of its own │ │126 +fn eventsFirstParent( │ │127 + comptime repo_kind: rp.RepoKind, │ │128 + comptime repo_opts: rp.RepoOpts(repo_kind), │ │129 + io: std.Io, │ │130 + allocator: std.mem.Allocator, │ │131 + repo: *rp.Repo(repo_kind, repo_opts), │ │132 +) !?[1][hash.hexLen(repo_opts.hash)]u8 { │ │133 + if (try repo.readRef(io, evt.events_ref) != null) return null; │ │134 + var remotes = try repo.listRemotes(io, allocator); │ │135 + defer remotes.deinit(); │ │136 + for (remotes.sections.keys()) |name| { │ │137 + if (try repo.readRef(io, .{ .kind = .{ .remote = name }, .name = evt.events_ref.name })) |oid| return .{oid};│ │138 + } │ │139 + return null; │ │140 +} │ │141 + │ │142 fn branchEvents( │ │143 comptime repo_kind: rp.RepoKind, │ │144 comptime repo_opts: rp.RepoOpts(repo_kind), │ ╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭──────────────────────────────────────────────────────────────────────────────────╮ │235 } │ │236 │ │237 // refresh the patches their source branch has outrun, inside the caller's │ │231 -// transaction, so a push and the revisions it causes land together. fork │ │232 -// backed patches are left to `refreshBranches`: their mergeability needs a │ │233 -// fork's lock, which has to be taken before this transaction opened │ │238 +// transaction, so a push and the revisions it causes land together. reports │ │239 +// whether any revision was written, so a caller that owns an otherwise empty│ │240 +// transaction can cancel it. mergeability is only saved for branch patches: │ │241 +// a fork's lock has to be taken before this transaction opened │ │242 pub fn refreshBranchesInTransaction( │ │243 + host_kind: evt.HostKind, │ │244 comptime repo_opts: rp.RepoOpts(.xit), │ │245 state: rp.Repo(.xit, repo_opts).State(.read_write), │ │246 moment: *evt.EventDB(repo_opts.hash).HashMap(.read_write), │ ╰──────────────────────────────────────────────────────────────────────────────────╯ ╭──────────────────────────────────────────────────────────────────────────────────────────────────────╮ │247 io: std.Io, │ │248 allocator: std.mem.Allocator, │ │249 updates: ?[]const xit.net_server_receive_pack.AppliedRefUpdate, │ │241 -) !void { │ │250 + first_parent: ?[1][hash.hexLen(repo_opts.hash)]u8, │ │251 +) !bool { │ │252 var arena = std.heap.ArenaAllocator.init(allocator); │ │253 defer arena.deinit(); │ │254 var patches: std.AutoArrayHashMapUnmanaged([evt.event_id_size]u8, evt.Patch.Record) = .empty;│ ╰──────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │255 defer patches.deinit(allocator); │ │256 │ │257 const events_moment = evt.currentMomentFromRepoMoment(repo_opts.hash, moment.readOnly()) catch |err| switch (err) { │ │248 - error.NotFound => return, │ │258 + error.NotFound => return false, │ │259 else => return err, │ │260 }; │ │251 - try collectOutdatedBranchPatches(.xit, repo_opts, io, allocator, &arena, state.readOnly(), events_moment, updates, .server, &patches); │ │261 + try collectOutdatedBranchPatches(.xit, repo_opts, io, allocator, &arena, state.readOnly(), events_moment, updates, host_kind, &patches);│ │262 │ │263 + var refreshed = false; │ │264 for (patches.keys(), patches.values()) |id, record| { │ │265 // a branch patch opens no fork, so its checks need no lock of their own │ │266 var checks: MergeCheckInputs(repo_opts) = .{}; │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │267 defer checks.deinit(io, allocator); │ │257 - try checks.add(io, allocator, state.core.work_path, id, record.event); │ │268 + if (host_kind == .server) try checks.add(io, allocator, state.core.work_path, id, record.event); │ │269 │ │270 const author = evt.CommitAuthor{ .name = "haxy", .email = record.author_email orelse "user@haxy" }; │ │260 - writeBranchPatchInTransaction(repo_opts, state, moment, io, &arena, std.fmt.bytesToHex(id, .lower), record.event, record.event, author, null, &checks) catch |err| { │ │271 + writeBranchPatchInTransaction(repo_opts, state, moment, io, &arena, std.fmt.bytesToHex(id, .lower), record.event, record.event, author, first_parent, &checks) catch |err| {│ │272 std.log.warn("failed to refresh branch patch: {s}", .{@errorName(err)}); │ │273 + continue; │ │274 }; │ │275 + refreshed = true; │ │276 } │ │277 + return refreshed; │ │278 } │ │279 │ │266 -// refresh open patches' tracked branches and, on the server, their mergeability. │ │267 -// this runs outside the caller's transaction. │ │280 +// refresh open patches' tracked branches and, on the server, their │ │281 +// mergeability. this runs after the caller's transaction, and on xit writes │ │282 +// every revision in one transaction of its own │ │283 pub fn refreshBranches( │ │284 host_kind: evt.HostKind, │ │285 comptime repo_kind: rp.RepoKind, │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │290 updates: ?[]const xit.net_server_receive_pack.AppliedRefUpdate, │ │291 progress_ctx_maybe: ?repo_opts.ProgressCtx, │ │292 ) !void { │ │278 - const DB = evt.EventDB(repo_opts.hash); │ │293 + if (repo_kind == .xit) { │ │294 + const DB = evt.EventDB(repo_opts.hash); │ │295 + │ │296 + // every revision shares one transaction, so the refresh is a single │ │297 + // undo entry. the transaction collects the patches itself, and is │ │298 + // cancelled when none of them moved │ │299 + { │ │300 + const Ctx = struct { │ │301 + core: *rp.Repo(.xit, repo_opts).Core, │ │302 + io: std.Io, │ │303 + allocator: std.mem.Allocator, │ │304 + host_kind: evt.HostKind, │ │305 + updates: ?[]const xit.net_server_receive_pack.AppliedRefUpdate, │ │306 + first_parent: ?[1][hash.hexLen(repo_opts.hash)]u8, │ │307 + │ │308 + pub fn run(ctx: @This(), cursor: *DB.Cursor(.read_write)) !void { │ │309 + var moment = try DB.HashMap(.read_write).init(cursor.*); │ │310 + const state = rp.Repo(.xit, repo_opts).State(.read_write){ .core = ctx.core, .extra = .{ .moment = &moment } }; │ │311 + if (!try refreshBranchesInTransaction(ctx.host_kind, repo_opts, state, &moment, ctx.io, ctx.allocator, ctx.updates, ctx.first_parent)) return error.CancelTransaction; │ │312 + │ │313 + // record the user action for undo │ │314 + try xit.undo.write(repo_opts, state, std.Io.Timestamp.now(ctx.io, .real).toSeconds(), .{ .custom = .{ .action = evt.undo_action } }); │ │315 + } │ │316 + }; │ │317 + │ │318 + const first_parent = try eventsFirstParent(repo_kind, repo_opts, io, allocator, repo); │ │319 + │ │320 + try repo.core.db_file.lock(io, .exclusive); │ │321 + defer repo.core.db_file.unlock(io); │ │322 + │ │323 + const history = try DB.ArrayList(.read_write).init(repo.core.db.rootCursor()); │ │324 + history.appendContext(.{ .slot = try history.getSlot(-1) }, Ctx{ │ │325 + .core = &repo.core, │ │326 + .io = io, │ │327 + .allocator = allocator, │ │328 + .host_kind = host_kind, │ │329 + .updates = updates, │ │330 + .first_parent = first_parent, │ │331 + }) catch |err| switch (err) { │ │332 + error.CancelTransaction => {}, │ │333 + else => return err, │ │334 + }; │ │335 + } │ │336 + │ │337 + // every open patch is checked after the transaction, since a fork's │ │338 + // lock has to be taken before the target's │ │339 + if (host_kind == .server) refreshOpenMergeability(repo_opts, io, allocator, repo, progress_ctx_maybe); │ │340 + return; │ │341 + } │ │342 + │ │343 + // git repos have no transaction to share, so each revision is written on │ │344 + // its own through the local event database. they report no progress, │ │345 + // since only a push has a progress context to report to │ │346 var arena = std.heap.ArenaAllocator.init(allocator); │ │347 defer arena.deinit(); │ │281 - var patches: std.AutoArrayHashMapUnmanaged([evt.event_id_size]u8, ?evt.Patch.Record) = .empty; │ │348 + var patches: std.AutoArrayHashMapUnmanaged([evt.event_id_size]u8, evt.Patch.Record) = .empty; │ │349 defer patches.deinit(allocator); │ │283 - var outdated: std.AutoArrayHashMapUnmanaged([evt.event_id_size]u8, evt.Patch.Record) = .empty; │ │284 - defer outdated.deinit(allocator); │ │350 │ │286 - // collect patches whose source or mergeability may have changed │ │351 collect: { │ │288 - var local_db: ?evt.LocalEventDB(repo_opts.hash) = if (repo_kind == .git) try evt.LocalEventDB(repo_opts.hash).openReadOnly(io, allocator, repo.core.repo_dir) else null; │ │289 - defer if (local_db) |*db| db.deinit(io, allocator); │ │290 - const moment = (if (local_db) |*db| evt.currentMomentFromDb(repo_opts.hash, db.db) else if (repo_kind == .git) break :collect else evt.currentMoment(repo_opts, repo)) catch |err| switch (err) {│ │352 + var local_db = (try evt.LocalEventDB(repo_opts.hash).openReadOnly(io, allocator, repo.core.repo_dir)) orelse break :collect; │ │353 + defer local_db.deinit(io, allocator); │ │354 + const moment = evt.currentMomentFromDb(repo_opts.hash, local_db.db) catch |err| switch (err) { │ │355 error.NotFound => break :collect, │ │356 else => return err, │ │357 }; │ ╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │294 - │ │295 - // include all open server patches for mergeability checks │ │296 - if (repo_kind == .xit and host_kind == .server) open: { │ │297 - const statuses_cursor = try moment.getCursor(hash.hashInt(repo_opts.hash, evt.Patch.status_to_id_set_key)) orelse break :open; │ │298 - const statuses = try DB.SortedMap(.read_only).init(statuses_cursor); │ │299 - const open_cursor = try statuses.getCursor("open") orelse break :open; │ │300 - const open_patches = try DB.SortedSet(.read_only).init(open_cursor); │ │301 - var iter = try open_patches.iteratorFromIndex(0); │ │302 - while (try iter.next()) |cursor| try patches.put(allocator, try evt.readOrderKeyId(DB, cursor), null); │ │303 - } │ │304 - │ │305 - switch (repo_kind) { │ │306 - .git => try collectOutdatedBranchPatches(repo_kind, repo_opts, io, allocator, &arena, .{ .core = &repo.core, .extra = .{} }, moment, updates, host_kind, &outdated), │ │307 - .xit => { │ │308 - var repo_moment = try repo.core.latestMoment(); │ │309 - try collectOutdatedBranchPatches(repo_kind, repo_opts, io, allocator, &arena, .{ .core = &repo.core, .extra = .{ .moment = &repo_moment } }, moment, updates, host_kind, &outdated);│ │310 - }, │ │311 - } │ │312 - for (outdated.keys(), outdated.values()) |id, record| try patches.put(allocator, id, record); │ │358 + try collectOutdatedBranchPatches(repo_kind, repo_opts, io, allocator, &arena, .{ .core = &repo.core, .extra = .{} }, moment, updates, host_kind, &patches); │ │359 } │ │360 │ │315 - // start progress reporting after the patch count is known │ │316 - if (repo_opts.ProgressCtx != void) { │ │317 - if (progress_ctx_maybe) |progress_ctx| progress_ctx.run(io, .{ .start = .{ .kind = .writing_patch, .estimated_total_items = patches.count() } }) catch {}; │ │361 + for (patches.keys(), patches.values()) |id, record| { │ │362 + const author = evt.CommitAuthor{ .name = "haxy", .email = record.author_email orelse "user@haxy" }; │ │363 + writeBranchPatch(host_kind, repo_kind, repo_opts, io, allocator, repo, std.fmt.bytesToHex(id, .lower), record.event, record.event, author) catch |err| { │ │364 + if (host_kind == .server) std.log.warn("failed to refresh branch patch: {s}", .{@errorName(err)}); │ │365 + }; │ │366 } │ │319 - │ │320 - // refresh each patch and report its completion │ │321 - for (patches.keys(), patches.values()) |id, record_maybe| { │ │322 - if (record_maybe) |record| { │ │323 - const author = evt.CommitAuthor{ .name = "haxy", .email = record.author_email orelse "user@haxy" }; │ │324 - writeBranchPatch(host_kind, repo_kind, repo_opts, io, allocator, repo, std.fmt.bytesToHex(id, .lower), record.event, record.event, author) catch |err| { │ │325 - if (host_kind == .server) std.log.warn("failed to refresh branch patch: {s}", .{@errorName(err)}); │ │326 - }; │ │327 - } │ │328 - if (repo_kind == .xit and host_kind == .server) refreshMergeability(repo_opts, io, allocator, repo, id); │ │329 - if (repo_opts.ProgressCtx != void) { │ │330 - if (progress_ctx_maybe) |progress_ctx| progress_ctx.run(io, .{ .complete_one = .writing_patch }) catch {}; │ │331 - } │ │332 - } │ │333 - │ │334 - // finish progress reporting after all patches have been visited │ │335 - if (repo_opts.ProgressCtx != void) { │ │336 - if (progress_ctx_maybe) |progress_ctx| progress_ctx.run(io, .{ .end = .writing_patch }) catch {}; │ │337 - } │ │367 } │ │368 │ │369 pub const PublishInput = struct { │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │468 io: std.Io, │ │469 allocator: std.mem.Allocator, │ │470 target_repo: *rp.Repo(.xit, repo_opts), │ │442 - id_maybe: ?[evt.event_id_size]u8, │ │471 + id: [evt.event_id_size]u8, │ │472 +) void { │ │473 + refreshMergeCheck(repo_opts, io, allocator, target_repo, &id) catch |err| { │ │474 + std.log.warn("failed to refresh mergeability: {s}", .{@errorName(err)}); │ │475 + }; │ │476 +} │ │477 + │ │478 +// check every open patch, reporting progress as each one finishes │ │479 +pub fn refreshOpenMergeability( │ │480 + comptime repo_opts: rp.RepoOpts(.xit), │ │481 + io: std.Io, │ │482 + allocator: std.mem.Allocator, │ │483 + target_repo: *rp.Repo(.xit, repo_opts), │ │484 + progress_ctx_maybe: ?repo_opts.ProgressCtx, │ │485 ) void { │ │444 - refreshMergeChecks(repo_opts, io, allocator, target_repo, id_maybe) catch |err| { │ │486 + refreshOpenMergeChecks(repo_opts, io, allocator, target_repo, progress_ctx_maybe) catch |err| {│ │487 std.log.warn("failed to refresh mergeability: {s}", .{@errorName(err)}); │ │488 }; │ │489 } │ ╰────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭──────╮ │next →│ ╰──────╯