HAXY

▄▀▄ ▄▀▀▀▀▄ ▀ ▄▄▀▀▀▀▀▀▄ ▀ ▀ ▀▀▀▀▀▀▀▄▀▄ ▄▄ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀ ▄▀▄ ▀▀▄ ▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀▄ ▀▀▀ ▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄ ▀▀▀▀ ▀▀▀▄▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀▀▀▄▀▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀▀▀▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▄▀▀▀▀▀▀▀▀▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀ ▄▀▀▀▀▀▀▀▄▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▄▄▄▄▄▄▀▀▀▀▀▀▀▀▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▄▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▄▄ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀▀▀ ▀▀▀▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▄▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀ ▀▀▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▄▀▀▀ ▀▀▀▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀▀▄▀▀ ▀▀▀▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▄▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀▀▀▀▀▀▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀▀▀▀▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▀▀▀▀▀▀▀▀▀▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀ ▀▀▀▀▀▀▀▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▄▀▀ ▀▀▀▀▀▀▀▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀ ▄▀▀▀▀▀▀▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▄ ▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▄▀▄ ▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀ ▄▄▄▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▀▄▀▀▀▀▄

╭──────────╮ │← previous│ ╰──────────╯ ╭ changed ───╮ │src/push.zig│ ╰────────────╯ ╭───────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │260 var revision_id_maybe: ?[evt.event_id_size]u8 = null; │ │261 var response = xit.net_server_receive_pack.Response.init(allocator); │ │262 defer response.deinit(); │ │319 - var progress = PushProgress{ .response = &response, .writer = writer }; │ │263 + var progress = serve_common.PushProgress{ .response = &response, .writer = writer, .sess = sess };│ │264 │ │265 // execute a transaction that receives the push and materializes its revision │ │266 const result = blk: { │ ╰───────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭──────────────────────────────────────────────────────────────────────────────────╮ │283 timestamp: u64, │ │284 newest: ?evt.PatchRev.WithId, │ │285 revision_id_maybe: *?[evt.event_id_size]u8, │ │342 - progress: *PushProgress, │ │286 + progress: *serve_common.PushProgress, │ │287 │ │288 pub fn run(ctx: @This(), cursor: *DB.Cursor(.read_write)) !void {│ │289 // receive the branch update │ ╰──────────────────────────────────────────────────────────────────────────────────╯ ╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │291 const state = State(.read_write){ .core = ctx.core, .extra = .{ .moment = &moment } }; │ │292 try xit.net_server_receive_pack.run(.xit, repo_opts, state, ctx.io, ctx.allocator, ctx.reader, ctx.writer, .{ .allowed_ref = "refs/heads/" ++ fork.ref.name, .deferred_response = ctx.response }, ctx.progress);│ │293 try writeReceivedPatches(repo_opts, state, ctx.io, ctx.allocator, ctx.response, ctx.writer); │ │294 + if (ctx.progress.gone) return error.ClientGone; │ │295 │ │296 // copy the target history needed to preserve the merge base │ │297 const source_oid = (try rf.readRecur(.xit, repo_opts, state.readOnly(), ctx.io, .{ .ref = fork.ref })) orelse return error.CancelTransaction; │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭ changed ────────────╮ │src/search_commit.zig│ ╰─────────────────────╯ ╭───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │259 serve_common.reportProgress(repo_opts, io, progress_ctx_maybe, .{ .text = "Indexing commits" }); │ │260 serve_common.reportProgress(repo_opts, io, progress_ctx_maybe, .{ .start = .{ .kind = .writing_patch, .estimated_total_items = diff.removed.len + diff.added.len } });│ │261 for (diff.removed) |commit| { │ │262 + if (serve_common.progressCancelled(repo_opts, progress_ctx_maybe)) return error.ClientGone; │ │263 defer serve_common.reportProgress(repo_opts, io, progress_ctx_maybe, .{ .complete_one = .writing_patch }); │ │264 const key = try readPosting(repo_opts, state.readOnly(), io, allocator, &message, commit); │ │265 try srch.remove(DB, version, allocator, &key, message.items); │ ╰───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │266 } │ │267 for (diff.added) |commit| { │ │268 + if (serve_common.progressCancelled(repo_opts, progress_ctx_maybe)) return error.ClientGone; │ │269 defer serve_common.reportProgress(repo_opts, io, progress_ctx_maybe, .{ .complete_one = .writing_patch });│ │270 const key = try readPosting(repo_opts, state.readOnly(), io, allocator, &message, commit); │ │271 try srch.add(DB, version, allocator, &key, message.items); │ ╰───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭ changed ───────────╮ │src/serve_common.zig│ ╰────────────────────╯ ╭───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │1 const std = @import("std"); │ │2 const evt = @import("./event.zig"); │ │3 -const rp = @import("xit").repo; │ │3 +const xit = @import("xit"); │ │4 +const rp = xit.repo; │ │5 +const ssh = @import("./serve_ssh_protocol.zig"); │ │6 │ │7 // progress is optional and best effort: a failed report must not fail the work │ │8 pub fn reportProgress(comptime repo_opts: rp.RepoOpts(.xit), io: std.Io, progress_ctx_maybe: ?repo_opts.ProgressCtx, event: rp.ProgressEvent) void {│ ╰───────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │10 if (progress_ctx_maybe) |progress_ctx| progress_ctx.run(io, event) catch {}; │ │11 } │ │12 } │ │13 + │ │14 +// progress for a push, sent on the pack stream's sideband │ │15 +pub const PushProgress = struct { │ │16 + response: *xit.net_server_receive_pack.Response, │ │17 + writer: *std.Io.Writer, │ │18 + // the client to probe, absent when nothing is listening │ │19 + sess: ?*ssh.SessionCtx = null, │ │20 + // set once it has hung up, so the phase in flight can stop too │ │21 + gone: bool = false, │ │22 + label: []const u8 = "Processing commits", │ │23 + count: usize = 0, │ │24 + total: usize = 0, │ │25 + // the last reported step, so a line is sent only when it changes │ │26 + step: usize = 0, │ │27 │ │28 + // how many items a phase with no total advances between reports │ │29 + const report_every = 1000; │ │30 + │ │31 + // the phases a push reports: the objects unpacked from the pack it │ │32 + // received, then everything counted after it │ │33 + fn reported(kind: rp.ProgressKind) bool { │ │34 + return kind == .writing_object_from_pack or kind == .writing_patch; │ │35 + } │ │36 + │ │37 + pub fn run(self: *PushProgress, _: std.Io, event: rp.ProgressEvent) !void { │ │38 + switch (event) { │ │39 + .start => |start| { │ │40 + if (!reported(start.kind)) return; │ │41 + // the unpack phase sends no label of its own │ │42 + if (start.kind == .writing_object_from_pack) self.label = "Unpacking objects"; │ │43 + self.total = start.estimated_total_items; │ │44 + self.count = 0; │ │45 + self.step = 0; │ │46 + }, │ │47 + .complete_one => |kind| { │ │48 + if (!reported(kind)) return; │ │49 + self.count += 1; │ │50 + }, │ │51 + .end => |kind| { │ │52 + if (!reported(kind)) return; │ │53 + }, │ │54 + // each phase names itself before reporting its count │ │55 + .text => |text| { │ │56 + self.label = text; │ │57 + return; │ │58 + }, │ │59 + else => return, │ │60 + } │ │61 + // a phase that knows no total counts instead, and stays quiet until it │ │62 + // has something to count │ │63 + if (self.total == 0 and self.count == 0) return; │ │64 + │ │65 + // every report flushes, so one is sent per percent, or per block of │ │66 + // items when there is no total to turn into one │ │67 + const step: usize = if (self.total > 0) @intCast(@as(u128, self.count) * 100 / self.total) else self.count / report_every;│ │68 + if (event == .complete_one and step == self.step) return; │ │69 + self.step = step; │ │70 + │ │71 + // reporting is the only moment a transaction looks up from its work │ │72 + if (self.sess) |sess| if (sess.peerGone()) { │ │73 + self.gone = true; │ │74 + return error.ClientGone; │ │75 + }; │ │76 + │ │77 + var buffer: [128]u8 = undefined; │ │78 + const suffix = if (event == .end) "\n" else "\r"; │ │79 + const text = if (self.total > 0) │ │80 + try std.fmt.bufPrint(&buffer, "{s}: {d}% ({d}/{d}){s}", .{ self.label, step, self.count, self.total, suffix }) │ │81 + else │ │82 + try std.fmt.bufPrint(&buffer, "{s}: {d}{s}", .{ self.label, self.count, suffix }); │ │83 + try self.response.progress(self.writer, text); │ │84 + } │ │85 +}; │ │86 + │ │87 +// whether the client the work is reporting to has hung up. only a sideband │ │88 +// reporter probes, so any other context never cancels. │ │89 +pub fn progressCancelled(comptime repo_opts: rp.RepoOpts(.xit), progress_ctx_maybe: ?repo_opts.ProgressCtx) bool { │ │90 + if (repo_opts.ProgressCtx != *SidebandProgress) return false; │ │91 + const progress_ctx = progress_ctx_maybe orelse return false; │ │92 + return progress_ctx.gone; │ │93 +} │ │94 + │ │95 // the outcome of resolving a requested repo path to its on-disk directory. │ │96 // the http and ssh paths each map the cases to their own error responses. │ │97 pub const RepoPath = union(enum) { │ ╰──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭ changed ────────╮ │src/serve_ssh.zig│ ╰─────────────────╯ ╭───────────────────────────────────────────────────────────────────────────────────────────────╮ │23 │ │24 var active_connections: std.atomic.Value(u32) = .init(0); │ │25 │ │26 -const any_repo_opts: rp.AnyRepoOpts(.xit) = .{ .ProgressCtx = *push.PushProgress }; │ │26 +const any_repo_opts: rp.AnyRepoOpts(.xit) = .{ .ProgressCtx = *serve_common.PushProgress };│ │27 │ │28 pub const SessionHandler = struct { │ │29 admin_repo_path: []const u8, │ ╰───────────────────────────────────────────────────────────────────────────────────────────────╯ ╭────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │395 }; │ │396 defer allocator.free(repo_path); │ │397 │ │398 - if (try serveIfExists(repo_path, handler.repo_root_path, &reader.interface, &writer.interface, io, allocator, parsed.service, protocol_version, author, handler.err)) return; │ │398 + if (try serveIfExists(repo_path, handler.repo_root_path, &reader.interface, &writer.interface, io, allocator, parsed.service, protocol_version, author, handler.err, sess)) return;│ │399 │ │400 if (!create_if_missing) return writeError(sess, "repo not found"); │ │401 │ ╰────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╮ │402 // create the on-disk repo for the just-minted event and serve the push │ │403 var repo = try createRepo(any_repo_opts.toRepoOpts(), io, allocator, repo_path); │ │404 defer repo.deinit(io, allocator); │ │405 - try servePack(repo.self_repo_opts, &repo, handler.repo_root_path, &reader.interface, &writer.interface, io, allocator, parsed.service, protocol_version, author, handler.err); │ │405 + try servePack(repo.self_repo_opts, &repo, handler.repo_root_path, &reader.interface, &writer.interface, io, allocator, parsed.service, protocol_version, author, handler.err, sess);│ │406 } │ │407 │ │408 // serve a patch draft: anyone may fetch it, only its author may push to it │ ╰─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────╯ ╭──────╮ │next →│ ╰──────╯