╭─────────────╮ │scroll to top│ ╰─────────────╯ ╭ push.zig ─────────────────────────────────────────────────────╮ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ │ ╰───────────────────────────────────────────────────────────────╯
🬐 🬠 🬞🬂🬏 🬟 🬑 🬟 🬑 ╭a6463714..╮ a6463714.. see-em: 🬋🬋🬋 🬋🬋🬋 🬋 🬋 │files │ commits refs discuss issues new login 🬐 🬠 🬐 🬠 🬑 🬟 🬰 ╰──────────╯ (695) 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 markdown.zig notify.zig patch.zig progress.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
124 const State = rp.Repo(.xit, repo_opts).State; 125 126 var response = xit.net_server_receive_pack.Response.init(allocator); 127 defer response.deinit(); 128 129 var updates = xit.net_server_receive_pack.AppliedRefUpdates.init(allocator); 130 defer updates.deinit(); 131 132 var sideband = progress.Sideband{ .response = &response, .writer = writer, .sess = sess }; 133 const Ctx = struct { 134 updates: *xit.net_server_receive_pack.AppliedRefUpdates, 135 core: *rp.Repo(.xit, repo_opts).Core, 136 io: std.Io, 137 allocator: std.mem.Allocator, 138 reader: *std.Io.Reader, 139 writer: *std.Io.Writer, 140 options: xit.net_server_receive_pack.Options, 141 author: evt.CommitAuthor, 142 response: *xit.net_server_receive_pack.Response, 143 server: evt.Host.Server, 144 error_writer: *std.Io.Writer, 145 sideband: *progress.Sideband, 146 147 pub fn run(ctx: @This(), cursor: *DB.Cursor(.read_write)) !void { 148 var moment = try DB.HashMap(.read_write).init(cursor.*); 149 const state = State(.read_write){ .core = ctx.core, .extra = .{ .moment = &moment } }; 150 151 var receive_options = ctx.options; 152 receive_options.applied_ref_updates = ctx.updates; 153 receive_options.deferred_response = ctx.response; 154 try xit.net_server_receive_pack.run(.xit, repo_opts, state, ctx.io, ctx.allocator, ctx.reader, ctx.writer, receive_options, ctx.sideband); 155 try writeReceivedPatches(repo_opts, state, ctx.io, ctx.allocator, ctx.response, ctx.writer); 156 if (ctx.sideband.gone) return error.ClientGone; 157 158 // a repo without an events branch has no events to consume. a 159 // no-op consume must not cancel here, since the push shares the 160 // transaction and must commit regardless. 161 if (null != try rf.readRecur(.xit, repo_opts, state.readOnly(), ctx.io, .{ .ref = evt.events_ref })) { 162 _ = try evt.consumeInTransaction(.repo, .xit, repo_opts, state, &ctx.core.db, &moment, ctx.io, ctx.allocator, evt.events_ref); 163 try pch.detectMerged(repo_opts, state, &ctx.core.db, &moment, ctx.io, ctx.allocator, ctx.server.users_dir, ctx.updates.items.items, ctx.error_writer); 164 165 // the revisions the pushed branches cause belong to the push 166 _ = try pch.refreshBranchesInTransaction(.{ .server = ctx.server }, repo_opts, state, &moment, ctx.io, ctx.allocator, ctx.updates.items.items, null, ctx.sideband); 167 } 168 169 if (ctx.sideband.gone) return error.ClientGone; 170 171 // the indexes the pushed refs invalidate belong to the push too 172 _ = try find.refreshInTransaction(repo_opts, state, &moment, ctx.io, ctx.allocator, ctx.sideband); 173 var moves: std.ArrayList(srch_cmmt.Move) = .empty; 174 defer moves.deinit(ctx.allocator); 175 for (ctx.updates.items.items) |update| try moves.append(ctx.allocator, .{ .old_oid = update.old_oid, .new_oid = update.new_oid }); 176 _ = try srch_cmmt.refreshInTransaction(repo_opts, state, &moment, ctx.io, ctx.allocator, moves.items, ctx.sideband); 177 178 try writeUndo(repo_opts, state, ctx.io, ctx.allocator, ctx.author, ctx.updates.items.items); 179 } 180 }; 181 182 const result = blk: { 183 try repo.core.db_file.lock(io, .exclusive); 184 defer repo.core.db_file.unlock(io); 185 186 const history = try DB.ArrayList(.read_write).init(repo.core.db.rootCursor()); 187 break :blk history.appendContext( 188 .{ .slot = try history.getSlot(-1) }, 189 Ctx{ 190 .updates = &updates, 191 .core = &repo.core, 192 .io = io, 193 .allocator = allocator, 194 .reader = reader, 195 .writer = writer, 196 .options = options, 197 .author = author, 198 .response = &response, 199 .server = server, 200 .error_writer = error_writer, 201 .sideband = &sideband, 202 }, 203 ); 204 }; 205 result catch |err| { 206 response.finish(writer, @errorName(err)) catch {}; 207 if (err == error.CancelTransaction) return; 208 return err; 209 }; 210 evt.bumpRepoActivityAt(io, allocator, server, repo.core.work_path); 211 212 pch.refreshBranches(.{ .server = server }, .xit, repo_opts, io, allocator, repo, updates.items.items, &sideband) catch |err| { 213 serve_common.logError(io, error_writer, "failed to refresh branch patches: {s}\n", .{@errorName(err)}); 214 pch.refreshOpenMergeability(repo_opts, io, allocator, repo, server.users_dir, null, &sideband); 215 }; 216 try response.finish(writer, null); 217 } 218 219 pub fn receiveFork( 220 comptime repo_opts: rp.RepoOpts(.xit), 221 io: std.Io, 222 allocator: std.mem.Allocator, 223 fork_repo: *rp.Repo(.xit, repo_opts), 224 target_repo: *rp.Repo(.xit, repo_opts), 225 server: evt.Host.Server, 226 id: *const [evt.event_id_size * 2]u8, 227 author: evt.CommitAuthor, 228 timestamp: u64, 229 reader: *std.Io.Reader, 230 writer: *std.Io.Writer, 231 error_writer: *std.Io.Writer, 232 sess: ?*ssh.SessionCtx, 233 ) !void { 234 const patch_id = try evt.parseEventId(id); 235 var arena = std.heap.ArenaAllocator.init(allocator); 236 defer arena.deinit(); 237 238 // load the fork and target patch state 239 const fork_moment = evt.currentMoment(repo_opts, fork_repo) catch return error.PatchDataUnavailable; 240 const fork_patch = (try evt.Patch.readById(evt.EventDB(repo_opts.hash), repo_opts.hash, fork_moment, &arena, &patch_id)) orelse return error.PatchDataUnavailable; 241 const target_moment = evt.currentMoment(repo_opts, target_repo) catch |err| switch (err) { 242 error.NotFound => null, 243 else => |other| return other, 244 }; 245 const target_patch = if (target_moment) |moment| 246 try evt.Patch.readById(evt.EventDB(repo_opts.hash), repo_opts.hash, moment, &arena, &patch_id) 247 else 248 null; 249 if (fork_patch.removed and target_patch == null) return error.PatchDataUnavailable; 250 if (target_patch) |patch| { 251 if (!std.mem.eql(u8, patch.author_email orelse "", author.email)) return error.PatchDataUnavailable; 252 if (patch.event.status.kind() == .merged) return error.PatchAlreadyMerged; 253 } 254 const published = target_patch != null; 255 const active_patch = target_patch orelse fork_patch; 256 const title = active_patch.event.title; 257 258 // resolve the target and the newest fork revision 259 const target_branch = active_patch.event.target_branch; 260 const target_oid = (try target_repo.readRef(io, .{ .kind = .head, .name = target_branch })) orelse return error.TargetNotFound; 261 const newest = try evt.PatchRev.readNewest(evt.EventDB(repo_opts.hash), repo_opts.hash, fork_moment, &arena); 262 var revision_id_maybe: ?[evt.event_id_size]u8 = null; 263 var response = xit.net_server_receive_pack.Response.init(allocator); 264 defer response.deinit(); 265 var sideband = progress.Sideband{ .response = &response, .writer = writer, .sess = sess }; 266 267 // execute a transaction that receives the push and materializes its revision 268 const result = blk: { 269 const DB = rp.Repo(.xit, repo_opts).DB; 270 const State = rp.Repo(.xit, repo_opts).State; 271 const Ctx = struct { 272 core: *rp.Repo(.xit, repo_opts).Core, 273 target_core: *rp.Repo(.xit, repo_opts).Core, 274 io: std.Io, 275 allocator: std.mem.Allocator, 276 reader: *std.Io.Reader, 277 writer: *std.Io.Writer, 278 response: *xit.net_server_receive_pack.Response, 279 patch_id: [evt.event_id_size * 2]u8, 280 patch: evt.Patch.Record, 281 published: bool, 282 target_oid: [hash.hexLen(repo_opts.hash)]u8, 283 title: []const u8, 284 author: evt.CommitAuthor, 285 timestamp: u64, 286 newest: ?evt.PatchRev.WithId, 287 revision_id_maybe: *?[evt.event_id_size]u8, 288 sideband: *progress.Sideband, 289 290 pub fn run(ctx: @This(), cursor: *DB.Cursor(.read_write)) !void { 291 // receive the branch update 292 var moment = try DB.HashMap(.read_write).init(cursor.*); 293 const state = State(.read_write){ .core = ctx.core, .extra = .{ .moment = &moment } }; 294 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.sideband); 295 try writeReceivedPatches(repo_opts, state, ctx.io, ctx.allocator, ctx.response, ctx.writer); 296 if (ctx.sideband.gone) return error.ClientGone; 297 298 // copy the target history needed to preserve the merge base 299 const source_oid = (try rf.readRecur(.xit, repo_opts, state.readOnly(), ctx.io, .{ .ref = fork.ref })) orelse return error.CancelTransaction; 300 var target_repo_moment = try ctx.target_core.latestMoment(); 301 const target_state = State(.read_only){ .core = ctx.target_core, .extra = .{ .moment = &target_repo_moment } }; 302 var objects = try obj.ObjectIterator(.xit, repo_opts).init(target_state, ctx.io, ctx.allocator, .{ .kind = .all }); 303 defer objects.deinit(); 304 try objects.include(&ctx.target_oid); 305 try obj.copyFromObjectIterator(.xit, repo_opts, state, .xit, repo_opts, &objects, ctx.io, ctx.sideband); 306 307 const base_oid = try mrg.commonAncestor(.xit, repo_opts, state.readOnly(), ctx.io, ctx.allocator, &ctx.target_oid, &source_oid); 308 const existing_revision = if (ctx.newest) |latest| 309 if (std.mem.eql(u8, latest.record.event.base_oid, &base_oid) and 310 std.mem.eql(u8, latest.record.event.source_oid, &source_oid)) latest else null 311 else 312 null; 313 314 // reuse an identical revision or record a new one 315 var revision_arena = std.heap.ArenaAllocator.init(ctx.allocator); 316 defer revision_arena.deinit(); 317 var events: [2]evt.EventWithId = undefined; 318 var event_count: usize = 0; 319 if (existing_revision) |latest| { 320 if (ctx.published) ctx.revision_id_maybe.* = latest.id; 321 } else { 322 const prepared = try evt.PatchRev.prepare(.xit, repo_opts, state, ctx.io, &revision_arena, &base_oid, &source_oid, ctx.title, ctx.author, ctx.timestamp); 323 events[event_count] = prepared.event; 324 event_count += 1; 325 ctx.revision_id_maybe.* = try evt.parseEventId(&prepared.event.id); 326 327 if (!ctx.published) { 328 var patch = ctx.patch.event; 329 patch.revision = prepared.revision; 330 events[event_count] = .{ 331 .id = ctx.patch_id, 332 .timestamp = ctx.timestamp, 333 .author = ctx.author, 334 .event = .{ .patch = patch }, 335 }; 336 event_count += 1; 337 } 338 } 339 340 // published patches keep only revisions in the fork 341 if (ctx.published and !ctx.patch.removed) { 342 events[event_count] = .{ 343 .id = ctx.patch_id, 344 .timestamp = ctx.timestamp, 345 .author = ctx.author, 346 .event = .{ .patch = null }, 347 }; 348 event_count += 1; 349 } 350 351 // commit and index any new events 352 if (event_count > 0) { 353 try evt.commitEvents(.xit, repo_opts, state, ctx.io, ctx.allocator, evt.events_ref, events[0..event_count], null); 354 if (!try evt.consumeInTransaction(.fork, .xit, repo_opts, state, &ctx.core.db, &moment, ctx.io, ctx.allocator, evt.events_ref)) return error.CancelTransaction; 355 } 356 // the patch branch's file index lands with the push 357 _ = try find.refreshInTransaction(repo_opts, state, &moment, ctx.io, ctx.allocator, ctx.sideband); 358 359 // a fork push tracks no ref updates, so it records no ranges 360 try writeUndo(repo_opts, state, ctx.io, ctx.allocator, ctx.author, &.{}); 361 } 362 }; 363 364 try fork_repo.core.db_file.lock(io, .exclusive); 365 defer fork_repo.core.db_file.unlock(io); 366 367 const history = try DB.ArrayList(.read_write).init(fork_repo.core.db.rootCursor()); 368 break :blk history.appendContext(.{ .slot = try history.getSlot(-1) }, Ctx{ 369 .core = &fork_repo.core, 370 .target_core = &target_repo.core, 371 .io = io, 372 .allocator = allocator, 373 .reader = reader, 374 .writer = writer, 375 .response = &response, 376 .patch_id = id.*, 377 .patch = fork_patch, 378 .published = published, 379 .target_oid = target_oid, 380 .title = title, 381 .author = author, 382 .timestamp = timestamp, 383 .newest = newest, 384 .revision_id_maybe = &revision_id_maybe, 385 .sideband = &sideband, 386 }); 387 }; 388 result catch |err| { 389 response.finish(writer, @errorName(err)) catch {}; 390 if (err == error.CancelTransaction) return; 391 return err; 392 }; 393 394 if (!published) return response.finish(writer, null); 395 396 sideband.run(io, .{ .text = "Updating patches" }) catch {}; 397 sideband.run(io, .{ .start = .{ .kind = .writing_patch, .estimated_total_items = 1 } }) catch {}; 398 399 // best-effort update the published patch so it has the new revision 400 var refreshed = false; 401 update: { 402 const revision_id = revision_id_maybe orelse break :update; 403 var update_arena = std.heap.ArenaAllocator.init(allocator); 404 defer update_arena.deinit(); 405 const target_update_moment = evt.currentMoment(repo_opts, target_repo) catch break :update; 406 const published_patch = (evt.Patch.readById(evt.EventDB(repo_opts.hash), repo_opts.hash, target_update_moment, &update_arena, &patch_id) catch break :update) orelse break :update; 407 if (published_patch.removed) break :update; 408 const fork_update_moment = evt.currentMoment(repo_opts, fork_repo) catch break :update; 409 const revision = (evt.PatchRev.readById(evt.EventDB(repo_opts.hash), repo_opts.hash, fork_update_moment, &update_arena, &revision_id) catch break :update) orelse break :update; 410 if (revision.removed) break :update; 411 const selected = evt.Patch.Revision.fromRecord(revision_id, revision); 412 if (published_patch.event.revision) |current| { 413 if (evt.fieldEqual(evt.Patch.Revision, current, selected)) break :update; 414 } 415 var patch = published_patch.event; 416 patch.revision = selected; 417 evt.consume(.{ .server = server }, .repo, .xit, repo_opts, io, allocator, target_repo, evt.events_ref, &.{.{ 418 .id = id.*, 419 .timestamp = timestamp, 420 .author = author, 421 .event = .{ .patch = patch }, 422 }}) catch |update_err| { 423 serve_common.logError(io, error_writer, "failed to update published patch {s}: {s}\n", .{ id, @errorName(update_err) }); 424 break :update; 425 }; 426 // consume also refreshed mergeability 427 refreshed = true; 428 } 429 430 if (!refreshed) pch.refreshMergeability(repo_opts, io, allocator, target_repo, server.users_dir, patch_id); 431 sideband.run(io, .{ .complete_one = .writing_patch }) catch {}; 432 sideband.run(io, .{ .end = .writing_patch }) catch {}; 433 434 // the client may read either repo as soon as it gets the final response. 435 try response.finish(writer, null); 436 } 437