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 merged line by line, it is chosen.89module sync1011import compress.zlib12import os13import src.apply14import src.events15import src.history16import src.remote17import src.snapshot18import src.storage1920pub enum Side {21	mine22	theirs23}2425// Clash is one file both machines changed differently. `stat` is the short26// diff stat the terminal prompt shows, so the choice is informed.27pub struct Clash {28pub mut:29	path   string30	mine   snapshot.FileEntry31	theirs snapshot.FileEntry32	base   snapshot.FileEntry33	stat   string34}3536pub struct StealOptions {37pub mut:38	mine   bool // resolve every clash as mine39	theirs bool // resolve every clash as theirs40	// Asked once per clash when neither flag is set. Nil means "nobody to ask".41	ask fn (clash Clash) Side = unsafe { nil }42}4344pub struct StealReport {45pub mut:46	repo            string47	forward         bool48	fetched_saves   int49	fetched_objects int50	fetched_events  int51	clashes         []string52	kept_mine       int53	replayed        int54	removed         int55	head            string56	flagged         []events.Flag // incoming saves that are flagged57}5859pub const dirty_tree = 'save or checkpoint first — steal needs a clean tree'60pub const no_saves_here = 'this folder has files but no saves — steal a repo into an empty folder with "koh steal <name>"'6162// steal brings the remote's new saves in. Inside a repo with its own work this63// may have to combine; a repo with nothing of its own simply fast-forwards.64pub fn steal(store &storage.Store, mut backend remote.Backend, opts StealOptions) !StealReport {65	name := repo_name(store, mut backend)!66	if snapshot.restore_pending(store) {67		return error(snapshot.restore_unfinished)68	}69	head_before := history.head(store)70	// Combining rewrites the working tree, so unsaved work would be lost.71	if dirty(store, head_before)! {72		if head_before == none {73			return error(no_saves_here)74		}75		return error(dirty_tree)76	}77	fetched := fetch_back_to_known(store, mut backend, name)!78	mut report := StealReport{79		repo:            name80		fetched_saves:   fetched.saves.len81		fetched_objects: fetched.repo_objects.len82		fetched_events:  fetched.events_out.len83		head:            fetched.head84	}85	mut local_only := []history.Save{}86	if head := head_before {87		// The remote builds on `base`. If that save is here but no longer on88		// the line, neither line contains the other; jumping to the remote89		// would silently leave this line behind.90		local_only = saves_after(store, head.id, fetched.base) or {91			return error('the remote builds on save ${fetched.base}, which is not on this line — load the save you want, then offer')92		}93	}94	if fetched.base == '' && local_only.len > 0 {95		return error('the remote keeps a different history — it shares no save with this repo')96	}9798	strips_before := events.strip_events(store)!.len99	mut clashes := []Clash{}100	mut side_of := map[string]Side{}101	if fetched.saves.len > 0 && local_only.len > 0 {102		clashes = combine_clashes(store, fetched)!103		report.clashes = clashes.map(it.path)104		// Asked before the write lock: a person thinking must not stall every105		// agent's save on this repo.106		side_of = choose_sides(clashes, opts)!107	}108109	store.begin() or { return err }110	// The questions ran without the lock. If a save or an edit landed111	// meanwhile, the answers were about a line that no longer exists.112	if moved := changed_since(store, head_before, strips_before) {113		if moved {114			store.rollback()115			return error('this repo changed while steal was asking — run koh steal again')116		}117	} else {118		store.rollback()119		return err120	}121	final := (write_and_combine(store, fetched, local_only, clashes, side_of, mut report) or {122		store.rollback()123		return err124	})125	files := history.files_of(store, final.id) or {126		store.rollback()127		return err128	}129	snapshot.check_links(store.root, files) or {130		store.rollback()131		return err132	}133	// The files follow the commit; the mark lands with it, in one step.134	if head_before == none || final.id != head_before or { history.Save{} }.id {135		snapshot.mark_restore(store) or {136			store.rollback()137			return err138		}139	}140	store.commit() or { return err }141	report.head = final.id142	flags := events.latest_flags(store)!143	for record in fetched.saves {144		if flag := flags[record.id] {145			if flag.flagged {146				report.flagged << flag147			}148		}149	}150151	if head := head_before {152		if final.id == head.id {153			// The line did not move, so neither does the working tree.154			return report155		}156		mut wanted := map[string]bool{}157		for file in files {158			wanted[file.path] = true159		}160		mut remove := []string{}161		for file in history.files_of(store, head.id)! {162			if file.path !in wanted {163				remove << file.path164			}165		}166		result := snapshot.restore_files(store, store.root, snapshot.load(store.root), files,167			remove)!168		report.removed = result.removed169	} else {170		snapshot.restore_files(store, store.root, snapshot.load(store.root), files, [])!171	}172	history.refresh_log_md(store)173	return report174}175176// write_and_combine lands a fetch and, when the local line has saves the remote177// lacks, replays them on top of the remote head. It runs inside the caller's178// transaction and returns the new head.179fn write_and_combine(store &storage.Store, fetched Fetched, local_only []history.Save, clashes []Clash, side_of map[string]Side, mut report StealReport) !history.Save {180	write_fetched(store, fetched)!181	if fetched.saves.len == 0 {182		// The remote head is already on this line: level, or this machine is183		// ahead and the offer that follows a steal sends the rest.184		return history.head(store) or { return error('this repo has no head') }185	}186	if local_only.len == 0 {187		// Nothing of ours is missing from the remote: the line simply moves on.188		history.set_head(store, fetched.head)!189		report.forward = true190		return history.head(store) or { return error('the remote head did not land') }191	}192	remote_files := history.files_of(store, fetched.head)!193	mut mine_files := []snapshot.FileEntry{}194	if our_head := history.head(store) {195		mine_files = history.files_of(store, our_head.id)!196	}197198	// A file the remote changed is a clash, decided below, or one this machine199	// left as it was or changed the same way, so the remote's version stands.200	// Replaying this machine's steps on it could only conflict: an edit it201	// later undid would no longer apply.202	base_files := history.files_of(store, fetched.base)!203	settled := changed_paths(base_files, remote_files)204205	// Take the remote's line first, so the replayed saves chain onto it.206	history.set_head(store, fetched.head)!207	mut current := remote_files.clone()208	for save in local_only {209		parent_files := if save.parent != '' {210			history.files_of(store, save.parent)!211		} else {212			[]snapshot.FileEntry{}213		}214		target_files := history.files_of(store, save.id)!215		plan := apply.three_way(current, parent_files, target_files, [], settled)216		if apply.conflicts(plan) {217			return error('combine could not apply ${plan.conflict} — it is not a clash file')218		}219		if plan.applied.len == 0 {220			// Everything this save touched was a file the remote changed.221			continue222		}223		history.record_save(store, '', plan.files, history.SaveInput{224			message:       save.message225			created_at:    save.created_at226			machine:       save.machine227			agent:         save.agent228			session:       save.session229			is_checkpoint: save.is_checkpoint230			replayed_from: save.id231		})!232		current = plan.files.clone()233		report.replayed++234	}235236	mut mine_paths := []string{}237	for clash in clashes {238		if side_of[clash.path] == .mine {239			mine_paths << clash.path240		}241	}242	if mine_paths.len > 0 {243		// One last save puts my versions of the chosen files back.244		final_plan := apply.three_way(current, current, mine_files, mine_paths, [])245		if final_plan.applied.len > 0 {246			history.record_save(store, '', final_plan.files, history.SaveInput{247				message: 'combine: kept mine for ${mine_paths.len} ${if mine_paths.len == 1 {248					'file'249				} else {250					'files'251				}}'252			})!253			report.kept_mine = mine_paths.len254		}255	}256	return history.head(store) or { return error('the combined line has no head') }257}258259// find_clashes lists the files both sides changed, differently. Files changed260// identically on both sides are not clashes: there is nothing to decide.261pub fn find_clashes(base []snapshot.FileEntry, mine []snapshot.FileEntry, theirs []snapshot.FileEntry) []Clash {262	base_map := snapshot.index_files(base)263	mine_map := snapshot.index_files(mine)264	theirs_map := snapshot.index_files(theirs)265	mut paths := mine_map.keys()266	for path in theirs_map.keys() {267		if path !in mine_map {268			paths << path269		}270	}271	paths.sort()272	mut clashes := []Clash{}273	for path in paths {274		here := mine_map[path] or {275			snapshot.FileEntry{276				path: path277			}278		}279		there := theirs_map[path] or {280			snapshot.FileEntry{281				path: path282			}283		}284		was := base_map[path] or {285			snapshot.FileEntry{286				path: path287			}288		}289		if same(here, there) {290			continue291		}292		if same(here, was) || same(there, was) {293			continue294		}295		clashes << Clash{296			path:   path297			mine:   here298			theirs: there299			base:   was300		}301	}302	return clashes303}304305// changed_paths lists the paths whose entry differs between two trees.306fn changed_paths(before []snapshot.FileEntry, after []snapshot.FileEntry) []string {307	was := snapshot.index_files(before)308	now := snapshot.index_files(after)309	mut paths := []string{}310	for path, entry in now {311		if !same(entry, was[path] or { snapshot.FileEntry{} }) {312			paths << path313		}314	}315	for path, _ in was {316		if path !in now {317			paths << path318		}319	}320	return paths321}322323// combine_clashes finds the files both lines changed, with the diff stat each324// prompt shows. It reads the remote head from the fetch, not the store, so it325// can run before anything is written.326fn combine_clashes(store &storage.Store, fetched Fetched) ![]Clash {327	base_files := if fetched.base != '' {328		history.files_of(store, fetched.base)!329	} else {330		[]snapshot.FileEntry{}331	}332	head := history.head(store) or { return []Clash{} }333	// The transaction applies the incoming strips before it replays, so the334	// clashes are found on the trees as they will be then, or the answers335	// would be about files that are no longer there.336	strips := fetched.events_out.filter(it.kind == events.kind_strip)337	// Base and mine are here already, so every strip reaches them (created 0).338	remote_head := fetched.saves.last()339	mut clashes := find_clashes(after_strips(base_files, fetched.base, 0, strips), after_strips(history.files_of(store,340		head.id)!, head.id, 0, strips), after_strips(to_files(remote_head), remote_head.id,341		remote_head.created_at, strips))342	for mut clash in clashes {343		mine := fetched_bytes(store, fetched, clash.mine)!344		theirs := fetched_bytes(store, fetched, clash.theirs)!345		delta := snapshot.diff_bytes(clash.path, 'modified', mine, theirs)!346		clash.stat = if delta.binary { 'binary' } else { '+${delta.added} -${delta.removed}' }347	}348	return clashes349}350351// after_strips is a save's file list once the given strip events are applied,352// by the same rule write_fetched uses.353fn after_strips(files []snapshot.FileEntry, save_id string, created_at i64, strips []events.Event) []snapshot.FileEntry {354	return files.filter(fn [strips, save_id, created_at] (file snapshot.FileEntry) bool {355		return !strips.any(reaches(it, save_id, created_at) && covers(it.path, file.path))356	})357}358359// fetched_bytes reads one side of a clash: from the fetch when it has just360// arrived, else from the store. An absent side is empty.361fn fetched_bytes(store &storage.Store, fetched Fetched, entry snapshot.FileEntry) ![]u8 {362	if entry.hash == '' {363		return []u8{}364	}365	if blob := fetched.repo_objects[entry.hash] {366		return zlib.decompress(blob)!367	}368	return store.get_object(entry.hash)!369}370371fn choose_sides(clashes []Clash, opts StealOptions) !map[string]Side {372	mut sides := map[string]Side{}373	if clashes.len == 0 {374		return sides375	}376	for clash in clashes {377		if opts.mine {378			sides[clash.path] = .mine379			continue380		}381		if opts.theirs {382			sides[clash.path] = .theirs383			continue384		}385		if opts.ask != unsafe { nil } {386			sides[clash.path] = opts.ask(clash)387			continue388		}389		return error('${clashes.len} files changed on both machines — run on a terminal to choose, or pass --mine or --theirs')390	}391	return sides392}393394fn same(a snapshot.FileEntry, b snapshot.FileEntry) bool {395	if a.hash == '' && b.hash == '' {396		return true397	}398	return a.hash == b.hash && a.executable == b.executable399}400401fn changed_since(store &storage.Store, before ?history.Save, strips_before int) !bool {402	if events.strip_events(store)!.len != strips_before {403		return true404	}405	now := history.head(store)406	if now or { history.Save{} }.id != before or { history.Save{} }.id {407		return true408	}409	return dirty(store, now)!410}411412// dirty reports whether the working tree holds anything a steal would413// overwrite. With no saves yet, that is every file but the untouched ignore414// list `koh init` wrote.415fn dirty(store &storage.Store, head ?history.Save) !bool {416	files := snapshot.walk(store, store.root, snapshot.load(store.root))!417	if saved := head {418		return !snapshot.same_tree(history.files_of(store, saved.id)!, files)419	}420	seed := storage.hash_hex(snapshot.default_ignore.bytes())421	return files.any(it.path != snapshot.ignore_file || it.hash != seed)422}423424// fetch_repo pulls a whole repo — head, saves, objects and events — for425// `koh steal <name>`, which lands a copy in a new folder.426pub fn fetch_repo(mut backend remote.Backend, name string) !Fetched {427	remote_head := backend.get(head_key(name))!428	if !remote_head.found {429		return error('the remote has no repo called "${name}"')430	}431	save_ids := backend.list('${name}/saves')!432	object_keys := backend.list('${name}/objects')!433	event_keys := backend.list('${name}/events')!434	mut fetched := Fetched{435		head: remote_head.body.bytestr().trim_space()436	}437	for key in object_keys {438		blob := backend.get(key)!439		if blob.found {440			fetched.repo_objects[os.base(key)] = blob.body441		}442	}443	for key in save_ids {444		body := backend.get(key)!445		if !body.found {446			continue447		}448		record := decode_save(body.body)!449		fetched.saves << record450	}451	fetched.saves = parent_first(fetched.saves)!452	for key in event_keys {453		body := backend.get(key)!454		if body.found {455			fetched.events_out << decode_event(body.body)!456		}457	}458	return fetched459}460461// parent_first orders saves so every parent lands before its children, as the462// foreign key on `parent` demands. Time cannot do it: a replayed save keeps its463// original, older time but sits on a newer parent.464fn parent_first(records []SaveRecord) ![]SaveRecord {465	mut by_id := map[string]SaveRecord{}466	for record in records {467		by_id[record.id] = record468	}469	mut placed := map[string]bool{}470	mut ordered := []SaveRecord{}471	for record in records {472		mut chain := []SaveRecord{}473		mut cursor := record.id474		for cursor != '' && cursor !in placed {475			if chain.len > records.len {476				return error('the save records on the remote form a loop')477			}478			current := by_id[cursor] or { return error('the remote is missing save ${cursor}') }479			chain << current480			cursor = current.parent481		}482		chain.reverse_in_place()483		for current in chain {484			placed[current.id] = true485			ordered << current486		}487	}488	return ordered489}