╭ serve_common.zig ─────────────────────────────────────────────╮ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ╰───────────────────────────────────────────────────────────────╯
🬐 🬠 🬞🬂🬏 🬟 🬑 🬟 🬑 ╭26a37d20..╮ 26a37d20.. see-em: 🬋🬋🬋 🬋🬋🬋 🬋 🬋 │files │ commits refs discuss issues new login 🬐 🬠 🬐 🬠 🬑 🬟 🬰 ╰──────────╯ (555) user
╭────╮ ╭ clone ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ╮ │http│ ssh │ │ ╰────╯ ╰ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ─ ╯
.. embed/ event/ test/ ui/ command.zig diff3.zig event.zig find.zig fork.zig lib.zig main.zig main_wasm.zig patch.zig push.zig search.zig search_commit.zig search_thread.zig serve.zig ╭─────────────────────────────────╮ │serve_common.zig │ ╰─────────────────────────────────╯ serve_http.zig serve_ssh.zig serve_ssh_protocol.zig test.zig testnet.zig try.zig ui.zig web.zig
1 const std = @import("std"); 2 const evt = @import("./event.zig"); 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 { 9 if (repo_opts.ProgressCtx != void) { 10 if (progress_ctx_maybe) |progress_ctx| progress_ctx.run(io, event) catch {}; 11 } 12 } 13 14 // progress for a push or a clone, sent on the pack stream's sideband 15 pub const SidebandProgress = struct { 16 // a push reports through its deferred response, which knows whether the 17 // client wanted sideband at all. a clone has none and writes band 2 itself. 18 response: ?*xit.net_server_receive_pack.Response = null, 19 writer: *std.Io.Writer, 20 // the client to probe, absent when nothing is listening 21 sess: ?*ssh.SessionCtx = null, 22 // set once it has hung up, so the phase in flight can stop too 23 gone: bool = false, 24 label: []const u8 = "Processing commits", 25 count: usize = 0, 26 total: usize = 0, 27 // the last reported step, so a line is sent only when it changes 28 step: usize = 0, 29 30 // how many items a phase with no total advances between reports 31 const report_every = 1000; 32 33 // the phases a client is told about, listed so a new kind has to be 34 // decided on rather than inherit whatever label was last set 35 fn reported(kind: rp.ProgressKind) bool { 36 return switch (kind) { 37 .writing_object_from_pack, .enumerating_object, .compressing_object, .writing_patch => true, 38 .writing_object, .sending_bytes, .receiving_bytes => false, 39 }; 40 } 41 42 pub fn run(self: *SidebandProgress, _: std.Io, event: rp.ProgressEvent) !void { 43 switch (event) { 44 .start => |start| { 45 if (!reported(start.kind)) return; 46 // a pack phase carries no label of its own; the rest send a 47 // .text first 48 switch (start.kind) { 49 .writing_object_from_pack => self.label = "Unpacking objects", 50 .enumerating_object => self.label = "Enumerating objects", 51 .compressing_object => self.label = "Compressing objects", 52 else => {}, 53 } 54 self.total = start.estimated_total_items; 55 self.count = 0; 56 self.step = 0; 57 }, 58 .complete_one => |kind| { 59 if (!reported(kind)) return; 60 self.count += 1; 61 }, 62 .end => |kind| { 63 if (!reported(kind)) return; 64 }, 65 // each phase names itself before reporting its count 66 .text => |text| { 67 self.label = text; 68 return; 69 }, 70 else => return, 71 } 72 // a phase that knows no total counts instead, and stays quiet until it 73 // has something to count 74 if (self.total == 0 and self.count == 0) return; 75 76 // every report flushes, so one is sent per percent, or per block of 77 // items when there is no total to turn into one 78 const step: usize = if (self.total > 0) @intCast(@as(u128, self.count) * 100 / self.total) else self.count / report_every; 79 if (event == .complete_one and step == self.step) return; 80 self.step = step; 81 82 // reporting is the only moment a transaction looks up from its work 83 if (self.sess) |sess| if (sess.peerGone()) { 84 self.gone = true; 85 return error.ClientGone; 86 }; 87 88 var buffer: [128]u8 = undefined; 89 const suffix = if (event == .end) "\n" else "\r"; 90 const text = if (self.total > 0) 91 try std.fmt.bufPrint(&buffer, "{s}: {d}% ({d}/{d}){s}", .{ self.label, step, self.count, self.total, suffix }) 92 else 93 try std.fmt.bufPrint(&buffer, "{s}: {d}{s}", .{ self.label, self.count, suffix }); 94 if (self.response) |response| { 95 try response.progress(self.writer, text); 96 } else { 97 try xit.net_server_pkt.sendSideband(self.writer, 2, text); 98 try self.writer.flush(); 99 } 100 } 101 }; 102 103 // whether the client the work is reporting to has hung up. only a sideband 104 // reporter probes, so any other context never cancels. 105 pub fn progressCancelled(comptime repo_opts: rp.RepoOpts(.xit), progress_ctx_maybe: ?repo_opts.ProgressCtx) bool { 106 if (repo_opts.ProgressCtx != *SidebandProgress) return false; 107 const progress_ctx = progress_ctx_maybe orelse return false; 108 return progress_ctx.gone; 109 } 110 111 // the outcome of resolving a requested repo path to its on-disk directory. 112 // the http and ssh paths each map the cases to their own error responses. 113 pub const RepoPath = union(enum) { 114 ok: []const u8, // resolved on-disk path; the caller owns and frees it 115 invalid, // a <owner>/<repo> path was expected but not given 116 not_found, // unknown owner, or the repo doesn't exist and we aren't creating 117 }; 118 119 // resolve a requested repo path to its on-disk directory, parsed as 120 // <owner>/<repo> through the event store, minting a repo event for a fresh push. 121 pub fn resolveRepoPath( 122 io: std.Io, 123 allocator: std.mem.Allocator, 124 repo_root_path: []const u8, 125 admin_repo_path: []const u8, 126 requested: []const u8, 127 create_if_missing: bool, 128 ) !RepoPath { 129 const owner_repo = evt.parseOwnerRepoPath(requested) orelse return .invalid; 130 const event_id_hex = (try evt.resolveOrCreateRepo( 131 io, 132 allocator, 133 admin_repo_path, 134 owner_repo.owner, 135 owner_repo.name, 136 if (create_if_missing) .{} else null, 137 )) orelse return .not_found; 138 return .{ .ok = try std.fs.path.join(allocator, &.{ repo_root_path, &event_id_hex }) }; 139 } 140 141 // every connection task shares one error writer, so writing to it takes a 142 // lock. uncancelable, since a log line shouldn't be a cancelation point. 143 var log_mutex: std.Io.Mutex = .init; 144 145 pub fn logError(io: std.Io, err: *std.Io.Writer, comptime fmt: []const u8, args: anytype) void { 146 log_mutex.lockUncancelable(io); 147 defer log_mutex.unlock(io); 148 err.print(fmt, args) catch return; 149 err.flush() catch {}; 150 } 151 152 // accept connections forever, spawning `handleConn(context, stream)` as a task 153 // for each. the handler owns the stream (it closes it). `name` labels accept 154 // errors in the log. shared by the http, ssh, and web ui listeners. 155 pub fn runListener( 156 io: std.Io, 157 net_server: *std.Io.net.Server, 158 tasks: *std.Io.Group, 159 err: *std.Io.Writer, 160 name: []const u8, 161 context: anytype, 162 comptime handleConn: fn (@TypeOf(context), std.Io.net.Stream) void, 163 ) void { 164 const Context = @TypeOf(context); 165 166 const Conn = struct { 167 context: Context, 168 stream: std.Io.net.Stream, 169 170 fn run(c: @This()) void { 171 handleConn(c.context, c.stream); 172 } 173 }; 174 175 const Listener = struct { 176 io: std.Io, 177 net_server: *std.Io.net.Server, 178 tasks: *std.Io.Group, 179 err: *std.Io.Writer, 180 name: []const u8, 181 context: Context, 182 183 fn run(self: @This()) void { 184 while (true) { 185 const stream = self.net_server.accept(self.io) catch |accept_err| { 186 if (accept_err == error.Canceled) return; 187 logError(self.io, self.err, "{s} accept failed: {s}\n", .{ self.name, @errorName(accept_err) }); 188 continue; 189 }; 190 self.tasks.async(self.io, Conn.run, .{Conn{ .context = self.context, .stream = stream }}); 191 } 192 } 193 }; 194 195 tasks.async(io, Listener.run, .{Listener{ 196 .io = io, 197 .net_server = net_server, 198 .tasks = tasks, 199 .err = err, 200 .name = name, 201 .context = context, 202 }}); 203 } 204