steal.v

1//! Stealing: bringing another machine's saves in, and combining when both2//! machines moved.3//!4//! The order is always the same, and it is what makes an interrupted steal5//! harmless: read everything from the remote first, then land it in one6//! transaction, then write the working tree. Divergence is resolved per file —7//! a clash is never blended line by line, it is chosen.89module exchange1011import compress.zlib12import src.apply13import src.events14import src.fmt15import src.history16import src.remote17import src.snapshot18import src.storage19import strconv2021pub enum Side {22	mine23	theirs24}2526// Clash is one file both machines changed differently. `stat` is the short27// diff stat the terminal prompt shows, so the choice is informed.28pub struct Clash {29pub mut:30	path         string31	mine         snapshot.FileEntry32	theirs       snapshot.FileEntry33	base         snapshot.FileEntry34	mine_stat    string35	theirs_stat  string36	mine_patch   string37	theirs_patch string38	mine_save    history.Save39	theirs_save  history.Save40}4142pub struct StealPreview {43pub:44	repo      string45	base      history.Save46	local     []history.Save47	remote    []SaveRecord48	changed   int49	automatic int50	identical int51	clashes   []Clash52	yes       bool53	all_side  string54}5556pub struct StealOptions {57pub mut:58	mine   bool // resolve every clash as mine59	theirs bool // resolve every clash as theirs60	// Shown the whole combine and asked about every clash. Nil means "nobody to ask".61	guide   fn (preview StealPreview) !map[string]Side = unsafe { nil }62	yes     bool63	json    bool64	approve fn (saves int, events int, yes bool, json bool) ! = unsafe { nil }65}6667pub struct StealReport {68pub mut:69	repo            string70	forward         bool71	fetched_saves   int72	fetched_objects int73	fetched_events  int74	clashes         []string75	kept_mine       int76	replayed        int77	removed         int78	head            string79	flagged         []events.Flag // incoming saves that are flagged80	// Set when a combine took this machine's saves off the line: loading it81	// brings back the tree from before the steal, less anything stripped.82	previous_head string83	kept_versions []string84	unsafe_paths  []string85}8687pub const dirty_tree = 'save or checkpoint first — steal needs a clean tree'8889// ClashError is a clash nobody was there to decide. It carries the paths because the advice90// depends on the command: `koh steal` takes --mine or --theirs, `koh offer` does not.91pub struct ClashError {92	Error93pub:94	paths []string95}9697pub fn (e ClashError) msg() string {98	return '${e.paths.join(', ')} changed on both machines — run on a terminal to choose, or pass --mine or --theirs with --yes'99}100101pub const no_saves_here = 'this folder has files but no saves — steal a repo into an empty folder with "koh steal <name>"'102103// steal brings the remote's new saves in. Inside a repo with its own work this104// may have to combine; a repo with nothing of its own simply fast-forwards.105pub fn steal(store &storage.Store, mut backend remote.Backend, opts StealOptions) !StealReport {106	name := repo_name(store, mut backend)!107	if snapshot.restore_pending(store) {108		return error(snapshot.restore_unfinished)109	}110	head_before := history.head(store)111	// Combining rewrites the working tree, so unsaved work would be lost.112	if dirty(store, head_before)! {113		if head_before == none {114			return error(no_saves_here)115		}116		return error(dirty_tree)117	}118	fetched := fetch_back_to_known(store, mut backend, name)!119	mut report := StealReport{120		repo:            name121		fetched_saves:   fetched.saves.len122		fetched_objects: fetched.repo_objects.len123		fetched_events:  fetched.events_out.len124		head:            fetched.head125		unsafe_paths:    fetched.unsafe_paths126	}127	mut local_only := []history.Save{}128	if head := head_before {129		// The remote builds on `base`. If that save is here but no longer on130		// the line, neither line contains the other; jumping to the remote131		// would silently leave this line behind.132		local_only = saves_after(store, head.id, fetched.base) or {133			return error('the remote builds on save ${fetched.base}, which is not on this line — load the save you want, then offer')134		}135	}136	if fetched.base == '' && local_only.len > 0 {137		return error('the remote keeps a different history — it shares no save with this repo')138	}139	if (fetched.saves.len > 0 || fetched.events_out.len > 0)140		&& !(fetched.saves.len > 0 && local_only.len > 0 && opts.guide != unsafe { nil })141		&& opts.approve != unsafe { nil } {142		opts.approve(fetched.saves.len, fetched.events_out.len, opts.yes, opts.json)!143	}144	if fetched.saves.len > 0 {145		store.begin()!146		write_fetched_records(store, fetched) or {147			store.rollback()148			return err149		}150		store.end()!151	}152153	strips_before := events.strip_events(store)!.len154	mut clashes := []Clash{}155	mut side_of := map[string]Side{}156	if fetched.saves.len > 0 && local_only.len > 0 {157		clashes = combine_clashes(store, fetched, local_only)!158		report.clashes = clashes.map(it.path)159		if opts.guide != unsafe { nil } {160			side_of = opts.guide(combine_preview(store, name, fetched, local_only, clashes,161				opts.yes, if opts.mine {162					'mine'163				} else if opts.theirs { 'theirs' } else { '' })!)!164		} else {165			// Asked before the write lock: a person thinking must not stall every166			// agent's save on this repo.167			side_of = choose_sides(clashes, opts)!168		}169	}170	for clash in clashes {171		if side_of[clash.path] == .theirs {172			report.kept_versions << '${clash.path}:${local_only.last().id}'173		}174	}175176	store.begin() or { return err }177	// The questions ran without the lock. If a save or an edit landed178	// meanwhile, the answers were about a line that no longer exists.179	if moved := changed_since(store, head_before, strips_before) {180		if moved {181			store.rollback()182			return error('this repo changed while steal was asking — run koh steal again')183		}184	} else {185		store.rollback()186		return err187	}188	final := (write_and_combine(store, fetched, local_only, clashes, side_of, mut report) or {189		store.rollback()190		return err191	})192	files := history.files_of(store, final.id) or {193		store.rollback()194		return err195	}196	snapshot.check_links(store.root, files) or {197		store.rollback()198		return err199	}200	// The files follow the transaction; the mark lands with it, in one step.201	// No head before means a fresh steal, whose whole tree is new.202	moved := if head := head_before { final.id != head.id } else { true }203	if moved {204		snapshot.mark_restore(store) or {205			store.rollback()206			return err207		}208	}209	store.end() or { return err }210	report.head = final.id211	if local_only.len > 0 && report.fetched_saves > 0 {212		report.previous_head = local_only.last().id213	}214	flags := events.latest_flags(store)!215	for record in fetched.saves {216		if flag := flags[record.id] {217			if flag.flagged {218				report.flagged << flag219			}220		}221	}222223	if head := head_before {224		if final.id == head.id {225			// The line did not move, so neither does the working tree.226			return report227		}228		mut wanted := map[string]bool{}229		for file in files {230			wanted[file.path] = true231		}232		mut remove := []string{}233		for file in history.files_of(store, head.id)! {234			if file.path !in wanted {235				remove << file.path236			}237		}238		result := snapshot.restore_files(store, store.root, snapshot.load(store.root), files,239			remove)!240		report.removed = result.removed241	} else {242		snapshot.restore_files(store, store.root, snapshot.load(store.root), files, [])!243	}244	history.refresh_log_md(store)245	return report246}247248// write_and_combine lands a fetch and, when the local line has saves the remote249// lacks, replays them on top of the remote head. It runs inside the caller's250// transaction and returns the new head.251fn write_and_combine(store &storage.Store, fetched Fetched, local_only []history.Save, clashes []Clash, side_of map[string]Side, mut report StealReport) !history.Save {252	write_fetched(store, fetched)!253	if fetched.saves.len == 0 {254		// The remote head is already on this line: level, or this machine is255		// ahead and the offer that follows a steal sends the rest.256		return history.head(store) or { return error('this repo has no head') }257	}258	if local_only.len == 0 {259		// Nothing of ours is missing from the remote: the line simply moves on.260		history.set_head(store, fetched.head)!261		report.forward = true262		return history.head(store) or { return error('the remote head did not land') }263	}264	remote_files := history.files_of(store, fetched.head)!265	mut mine_files := []snapshot.FileEntry{}266	if our_head := history.head(store) {267		mine_files = history.files_of(store, our_head.id)!268	}269270	// A file the remote changed is a clash, decided below, or one this machine271	// left as it was or changed the same way, so the remote's version stands.272	// Replaying this machine's steps on it could only conflict: an edit it273	// later undid would no longer apply.274	base_files := history.files_of(store, fetched.base)!275	settled := changed_paths(base_files, remote_files)276277	// Take the remote's line first, so the replayed saves chain onto it.278	history.set_head(store, fetched.head)!279	mut current := remote_files.clone()280	for save in local_only {281		parent_files := if save.parent != '' {282			history.files_of(store, save.parent)!283		} else {284			[]snapshot.FileEntry{}285		}286		target_files := history.files_of(store, save.id)!287		plan := apply.three_way(current, parent_files, target_files, [], settled)288		if apply.conflicts(plan) {289			return error('combine could not apply ${plan.conflict} — it is not a clash file')290		}291		if plan.applied.len == 0 {292			// Everything this save touched was a file the remote changed.293			continue294		}295		history.record_save(store, '', plan.files, history.SaveInput{296			message:       save.message297			created_at:    save.created_at298			machine:       save.machine299			agent:         save.agent300			session:       save.session301			is_checkpoint: save.is_checkpoint302			replayed_from: save.id303		})!304		current = plan.files.clone()305		report.replayed++306	}307308	mut mine_paths := []string{}309	for clash in clashes {310		if side_of[clash.path] == .mine {311			mine_paths << clash.path312		}313	}314	if mine_paths.len > 0 {315		// One last save puts my versions of the chosen files back.316		final_plan := apply.three_way(current, current, mine_files, mine_paths, [])317		if final_plan.applied.len > 0 {318			history.record_save(store, '', final_plan.files, history.SaveInput{319				message: 'combine: kept mine for ${mine_paths.len} ${if mine_paths.len == 1 {320					'file'321				} else {322					'files'323				}}'324			})!325			report.kept_mine = mine_paths.len326		}327	}328	return history.head(store) or { return error('the combined line has no head') }329}330331// find_clashes lists the files both sides changed, differently. Files changed332// identically on both sides are not clashes: there is nothing to decide.333pub fn find_clashes(base []snapshot.FileEntry, mine []snapshot.FileEntry, theirs []snapshot.FileEntry) []Clash {334	base_map := snapshot.index_files(base)335	mine_map := snapshot.index_files(mine)336	theirs_map := snapshot.index_files(theirs)337	mut paths := mine_map.keys()338	for path in theirs_map.keys() {339		if path !in mine_map {340			paths << path341		}342	}343	paths.sort()344	mut clashes := []Clash{}345	for path in paths {346		here := mine_map[path] or {347			snapshot.FileEntry{348				path: path349			}350		}351		there := theirs_map[path] or {352			snapshot.FileEntry{353				path: path354			}355		}356		was := base_map[path] or {357			snapshot.FileEntry{358				path: path359			}360		}361		if same(here, there) {362			continue363		}364		if same(here, was) || same(there, was) {365			continue366		}367		clashes << Clash{368			path:   path369			mine:   here370			theirs: there371			base:   was372		}373	}374	return clashes375}376377// changed_paths lists the paths whose entry differs between two trees.378fn changed_paths(before []snapshot.FileEntry, after []snapshot.FileEntry) []string {379	was := snapshot.index_files(before)380	now := snapshot.index_files(after)381	mut paths := []string{}382	for path, entry in now {383		if !same(entry, was[path] or { snapshot.FileEntry{} }) {384			paths << path385		}386	}387	for path, _ in was {388		if path !in now {389			paths << path390		}391	}392	return paths393}394395// combine_clashes finds the files both lines changed, with the diff stat each396// prompt shows. It reads the remote head from the fetch, not the store, so it397// can run before anything is written.398fn combine_clashes(store &storage.Store, fetched Fetched, local_only []history.Save) ![]Clash {399	base_files := if fetched.base != '' {400		history.files_of(store, fetched.base)!401	} else {402		[]snapshot.FileEntry{}403	}404	head := history.head(store) or { return []Clash{} }405	// The transaction applies every known strip before it replays, so the406	// clashes are found on the trees as they will be then, or the answers407	// would be about files that are no longer there.408	strips := known_strips(store, fetched)!409	remote_head := fetched.saves.last()410	mut clashes := find_clashes(after_strips(base_files, fetched.base, strips), after_strips(history.files_of(store,411		head.id)!, head.id, strips), after_strips(to_files(remote_head), remote_head.id, strips))412	for mut clash in clashes {413		base := fetched_bytes(store, fetched, clash.base)!414		mine := fetched_bytes(store, fetched, clash.mine)!415		theirs := fetched_bytes(store, fetched, clash.theirs)!416		mine_delta := snapshot.diff_bytes(clash.path, 'modified', base, mine)!417		theirs_delta := snapshot.diff_bytes(clash.path, 'modified', base, theirs)!418		clash.mine_stat = delta_stat(mine_delta, clash.mine.size)419		clash.theirs_stat = delta_stat(theirs_delta, clash.theirs.size)420		clash.mine_patch = mine_delta.patch421		clash.theirs_patch = theirs_delta.patch422		clash.mine_save = last_touch(store, local_only, clash.path)!423		clash.theirs_save = last_touch(store, fetched.saves.map(to_save(it)), clash.path)!424	}425	return clashes426}427428fn delta_stat(delta snapshot.Delta, size i64) string {429	if delta.binary {430		return fmt.human_bytes(size)431	}432	return '+${delta.added} -${delta.removed}'433}434435fn last_touch(store &storage.Store, saves []history.Save, path string) !history.Save {436	for index := saves.len - 1; index >= 0; index-- {437		save := saves[index]438		current := snapshot.index_files(history.files_of(store, save.id)!)[path] or {439			snapshot.FileEntry{}440		}441		parent := if save.parent != '' {442			snapshot.index_files(history.files_of(store, save.parent)!)[path] or {443				snapshot.FileEntry{}444			}445		} else {446			snapshot.FileEntry{}447		}448		if !same(current, parent) {449			return save450		}451	}452	return saves.last()453}454455// combine_preview counts what a combine would do to the paths either side changed since `base`:456// `identical` were changed the same way on both sides, `clashes` need a choice, the rest go in by themselves.457fn combine_preview(store &storage.Store, name string, fetched Fetched, local []history.Save, clashes []Clash, yes bool, all_side string) !StealPreview {458	base := history.files_of(store, fetched.base)!459	mine := history.files_of(store, local.last().id)!460	theirs := history.files_of(store, fetched.head)!461	mine_changed := changed_paths(base, mine)462	theirs_changed := changed_paths(base, theirs)463	mut changed := map[string]bool{}464	for path in mine_changed {465		changed[path] = true466	}467	for path in theirs_changed {468		changed[path] = true469	}470	mut identical := 0471	base_map := snapshot.index_files(base)472	mine_map := snapshot.index_files(mine)473	theirs_map := snapshot.index_files(theirs)474	for path, _ in changed {475		if !same(mine_map[path] or { snapshot.FileEntry{} }, base_map[path] or { snapshot.FileEntry{} })476			&& same(mine_map[path] or { snapshot.FileEntry{} }, theirs_map[path] or { snapshot.FileEntry{} }) {477			identical++478		}479	}480	return StealPreview{481		repo:      name482		base:      history.by_id(store, fetched.base) or { history.Save{} }483		local:     local484		remote:    fetched.saves485		changed:   changed.len486		automatic: changed.len - identical - clashes.len487		identical: identical488		clashes:   clashes489		yes:       yes490		all_side:  all_side491	}492}493494// after_strips is a save's file list once the given strip events are applied,495// by the same rule write_fetched uses.496fn after_strips(files []snapshot.FileEntry, save_id string, strips []events.Event) []snapshot.FileEntry {497	return files.filter(fn [strips, save_id] (file snapshot.FileEntry) bool {498		return !strips.any(reaches(it, save_id) && covers(it.path, file.path))499	})500}501502// fetched_bytes reads one side of a clash: from the fetch when it has just503// arrived, else from the store. An absent side is empty.504fn fetched_bytes(store &storage.Store, fetched Fetched, entry snapshot.FileEntry) ![]u8 {505	if entry.hash == '' {506		return []u8{}507	}508	if blob := fetched.repo_objects[entry.hash] {509		return zlib.decompress(blob)!510	}511	return store.get_object(entry.hash)!512}513514fn choose_sides(clashes []Clash, opts StealOptions) !map[string]Side {515	mut sides := map[string]Side{}516	if clashes.len == 0 {517		return sides518	}519	for clash in clashes {520		if opts.mine {521			sides[clash.path] = .mine522			continue523		}524		if opts.theirs {525			sides[clash.path] = .theirs526			continue527		}528		return ClashError{529			paths: clashes.map(it.path)530		}531	}532	return sides533}534535fn same(a snapshot.FileEntry, b snapshot.FileEntry) bool {536	if a.hash == '' && b.hash == '' {537		return true538	}539	return a.hash == b.hash && a.executable == b.executable540}541542fn changed_since(store &storage.Store, before ?history.Save, strips_before int) !bool {543	if events.strip_events(store)!.len != strips_before {544		return true545	}546	now := history.head(store)547	if now or { history.Save{} }.id != before or { history.Save{} }.id {548		return true549	}550	return dirty(store, now)!551}552553// dirty reports whether the working tree holds anything a steal would554// overwrite. With no saves yet, that is every file but the untouched ignore555// list `koh init` wrote.556pub fn dirty(store &storage.Store, head ?history.Save) !bool {557	files := snapshot.walk(store, store.root, snapshot.load(store.root))!558	if saved := head {559		return !snapshot.same_tree(history.files_of(store, saved.id)!, files)560	}561	seed := storage.hash_hex(snapshot.default_ignore.bytes())562	return files.any(it.path != snapshot.ignore_file || it.hash != seed)563}564565// fetch_repo takes a whole repo — head, saves, objects and events — for566// `koh steal <name>`, which lands a copy in a new folder.567pub fn fetch_repo(mut backend remote.Backend, name string) !Fetched {568	check_format(mut backend, name)!569	remote_head := backend.get(head_key(name))!570	if !remote_head.found {571		return error('the remote has no repo called "${name}"')572	}573	mut fetched := Fetched{574		head: remote_head.body.bytestr().trim_space()575	}576	mut bundle := &Bundle{577		fetched: fetched578	}579	if fetch_bundle(mut backend, bundle_key(name), mut bundle)! {580		fetched = bundle.fetched581	} else {582		for kind in ['objects', 'saves', 'events'] {583			for key in backend.list('${name}/${kind}')! {584				body := backend.get(key)!585				if body.found {586					take(mut fetched, key.all_after('${name}/'), body.body)!587				}588			}589		}590	}591	fetched.saves = parent_first(fetched.saves)!592	fetched.events_out.sort(a.id < b.id)593	return fetched594}595596// fetch_bundle reads §22.7's bundle, when the remote has one. Over HTTPS it is read as it arrives:597// held whole, a 282 MB bundle cost 2.5 GB, mostly copies the HTTP client makes.598fn fetch_bundle(mut backend remote.Backend, key string, mut bundle Bundle) !bool {599	found := if mut backend is remote.Web {600		backend.stream(key, fn [mut bundle] (piece []u8) ! {601			bundle.feed(piece)!602		})!603	} else {604		whole := backend.get(key)!605		if whole.found {606			bundle.feed(whole.body)!607		}608		whole.found609	}610	if found {611		bundle.finish()!612	}613	return found614}615616// Bundle takes §22.7's bundle — `<key> <length>\n` and then that many bytes, over and over, until617// `end 0` — a piece at a time, into what one GET of each key would have returned. A connection618// that drops between two entries still looks like a whole bundle, so only the end line proves it619// arrived.620struct Bundle {621mut:622	fetched Fetched623	pending []u8 // the start of an entry still arriving624	ended   bool625}626627fn (mut b Bundle) feed(piece []u8) ! {628	b.pending << piece629	mut at := 0630	for at < b.pending.len {631		// Scanned in place: a slice copies on this V, and copying the rest per entry is quadratic.632		mut newline := at633		for newline < b.pending.len && b.pending[newline] != `\n` {634			newline++635		}636		if newline == b.pending.len {637			break638		}639		header := b.pending[at..newline].bytestr()640		if b.ended {641			return error('the bundle goes on after its end')642		}643		if header == 'end 0' {644			b.ended = true645			at = newline + 1646			continue647		}648		length := strconv.atoi(header.all_after(' '))!649		start := newline + 1650		if length < 0 {651			return error('the bundle has a negative length for ${header.all_before(' ')}')652		}653		if length > b.pending.len - start {654			break655		}656		take(mut b.fetched, header.all_before(' '), b.pending[start..start + length])!657		at = start + length658	}659	if at > 0 {660		b.pending = b.pending[at..]661	}662}663664fn (b &Bundle) finish() ! {665	if !b.ended || b.pending.len > 0 {666		return error('the bundle was cut short — try the steal again')667	}668}669670// take files one fetched key, relative to the repo: an object, a save or an event.671fn take(mut fetched Fetched, key string, body []u8) ! {672	match key.all_before('/') {673		'objects' {674			fetched.repo_objects[key.all_after('/')] = body675		}676		'saves' {677			mut record := decode_save(body)!678			fetched.unsafe_paths << drop_unsafe_paths(mut record)679			fetched.saves << record680		}681		'events' {682			event := decode_event(body)!683			if !ignored_strip(event) {684				fetched.events_out << event685			}686		}687		else {688			return error('the remote sent "${key}", which is not part of a repo')689		}690	}691}692693// parent_first orders saves so every parent lands before its children, as the694// foreign key on `parent` demands. Time cannot do it: a replayed save keeps its695// original, older time but sits on a newer parent.696fn parent_first(records []SaveRecord) ![]SaveRecord {697	mut by_id := map[string]SaveRecord{}698	for record in records {699		by_id[record.id] = record700	}701	mut placed := map[string]bool{}702	mut ordered := []SaveRecord{}703	for record in records {704		mut chain := []SaveRecord{}705		mut cursor := record.id706		for cursor != '' && cursor !in placed {707			if chain.len > records.len {708				return error('the save records on the remote form a loop')709			}710			current := by_id[cursor] or { return error('the remote is missing save ${cursor}') }711			chain << current712			cursor = current.parent713		}714		chain.reverse_in_place()715		for current in chain {716			placed[current.id] = true717			ordered << current718		}719	}720	return ordered721}