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}5758pub const dirty_tree = 'save or checkpoint first — steal needs a clean tree'59pub const no_saves_here = 'this folder has files but no saves — steal a repo into an empty folder with "koh steal <name>"'6061// steal brings the remote's new saves in. Inside a repo with its own work this62// may have to combine; a repo with nothing of its own simply fast-forwards.63pub fn steal(store &storage.Store, mut backend remote.Backend, opts StealOptions) !StealReport {64	name := repo_name(store, mut backend)!65	head_before := history.head(store)66	// Combining rewrites the working tree, so unsaved work would be lost.67	if dirty(store, head_before)! {68		if head_before == none {69			return error(no_saves_here)70		}71		return error(dirty_tree)72	}73	fetched := fetch_back_to_known(store, mut backend, name)!74	mut report := StealReport{75		repo:            name76		fetched_saves:   fetched.saves.len77		fetched_objects: fetched.repo_objects.len78		fetched_events:  fetched.events_out.len79		head:            fetched.head80	}81	mut local_only := []history.Save{}82	if head := head_before {83		// The remote builds on `base`. If that save is here but no longer on84		// the line, neither line contains the other; jumping to the remote85		// would silently leave this line behind.86		local_only = saves_after(store, head.id, fetched.base) or {87			return error('the remote builds on save ${fetched.base}, which is not on this line — load the save you want, then offer')88		}89	}90	if fetched.base == '' && local_only.len > 0 {91		return error('the remote keeps a different history — it shares no save with this repo')92	}9394	strips_before := events.strip_events(store)!.len95	mut clashes := []Clash{}96	mut side_of := map[string]Side{}97	if fetched.saves.len > 0 && local_only.len > 0 {98		clashes = combine_clashes(store, fetched)!99		report.clashes = clashes.map(it.path)100		// Asked before the write lock: a person thinking must not stall every101		// agent's save on this repo.102		side_of = choose_sides(clashes, opts)!103	}104105	store.begin() or { return err }106	// The questions ran without the lock. If a save or an edit landed107	// meanwhile, the answers were about a line that no longer exists.108	if moved := changed_since(store, head_before, strips_before) {109		if moved {110			store.rollback()111			return error('this repo changed while steal was asking — run koh steal again')112		}113	} else {114		store.rollback()115		return err116	}117	final := (write_and_combine(store, fetched, local_only, clashes, side_of, mut report) or {118		store.rollback()119		return err120	})121	store.commit() or { return err }122	report.head = final.id123124	if head := head_before {125		if final.id == head.id {126			// The line did not move, so neither does the working tree.127			return report128		}129		files := history.files_of(store, final.id)!130		mut wanted := map[string]bool{}131		for file in files {132			wanted[file.path] = true133		}134		mut remove := []string{}135		for file in history.files_of(store, head.id)! {136			if file.path !in wanted {137				remove << file.path138			}139		}140		result := snapshot.restore_files(store, store.root, snapshot.load(store.root), files,141			remove)!142		report.removed = result.removed143	} else {144		files := history.files_of(store, final.id)!145		snapshot.restore_files(store, store.root, snapshot.load(store.root), files, [])!146	}147	history.refresh_log_md(store)148	return report149}150151// write_and_combine lands a fetch and, when the local line has saves the remote152// lacks, replays them on top of the remote head. It runs inside the caller's153// transaction and returns the new head.154fn write_and_combine(store &storage.Store, fetched Fetched, local_only []history.Save, clashes []Clash, side_of map[string]Side, mut report StealReport) !history.Save {155	write_fetched(store, fetched)!156	if fetched.saves.len == 0 {157		// The remote head is already on this line: level, or this machine is158		// ahead and the offer that follows a steal sends the rest.159		return history.head(store) or { return error('this repo has no head') }160	}161	if local_only.len == 0 {162		// Nothing of ours is missing from the remote: the line simply moves on.163		history.set_head(store, fetched.head)!164		report.forward = true165		return history.head(store) or { return error('the remote head did not land') }166	}167	remote_files := history.files_of(store, fetched.head)!168	mut mine_files := []snapshot.FileEntry{}169	if our_head := history.head(store) {170		mine_files = history.files_of(store, our_head.id)!171	}172173	// Take the remote's line first, so the replayed saves chain onto it.174	history.set_head(store, fetched.head)!175	mut current := remote_files.clone()176	for save in local_only {177		parent_files := if save.parent != '' {178			history.files_of(store, save.parent)!179		} else {180			[]snapshot.FileEntry{}181		}182		target_files := history.files_of(store, save.id)!183		plan := apply.three_way(current, parent_files, target_files, [], clashes.map(it.path))184		if apply.conflicts(plan) {185			return error('combine could not apply ${plan.conflict} — it is not a clash file')186		}187		if plan.applied.len == 0 {188			// Everything this save touched was a clash resolved as theirs.189			continue190		}191		history.record_save(store, '', plan.files, history.SaveInput{192			message:       save.message193			created_at:    save.created_at194			machine:       save.machine195			agent:         save.agent196			session:       save.session197			is_checkpoint: save.is_checkpoint198			replayed_from: save.id199		})!200		current = plan.files.clone()201		report.replayed++202	}203204	mut mine_paths := []string{}205	for clash in clashes {206		if side_of[clash.path] == .mine {207			mine_paths << clash.path208		}209	}210	if mine_paths.len > 0 {211		// One last save puts my versions of the chosen files back.212		final_plan := apply.three_way(current, current, mine_files, mine_paths, [])213		if final_plan.applied.len > 0 {214			history.record_save(store, '', final_plan.files, history.SaveInput{215				message: 'combine: kept mine for ${mine_paths.len} ${if mine_paths.len == 1 {216					'file'217				} else {218					'files'219				}}'220			})!221			report.kept_mine = mine_paths.len222		}223	}224	return history.head(store) or { return error('the combined line has no head') }225}226227// find_clashes lists the files both sides changed, differently. Files changed228// identically on both sides are not clashes: there is nothing to decide.229pub fn find_clashes(base []snapshot.FileEntry, mine []snapshot.FileEntry, theirs []snapshot.FileEntry) []Clash {230	base_map := snapshot.index_files(base)231	mine_map := snapshot.index_files(mine)232	theirs_map := snapshot.index_files(theirs)233	mut paths := mine_map.keys()234	for path in theirs_map.keys() {235		if path !in mine_map {236			paths << path237		}238	}239	paths.sort()240	mut clashes := []Clash{}241	for path in paths {242		here := mine_map[path] or {243			snapshot.FileEntry{244				path: path245			}246		}247		there := theirs_map[path] or {248			snapshot.FileEntry{249				path: path250			}251		}252		was := base_map[path] or {253			snapshot.FileEntry{254				path: path255			}256		}257		if same(here, there) {258			continue259		}260		if same(here, was) || same(there, was) {261			continue262		}263		clashes << Clash{264			path:   path265			mine:   here266			theirs: there267			base:   was268		}269	}270	return clashes271}272273// combine_clashes finds the files both lines changed, with the diff stat each274// prompt shows. It reads the remote head from the fetch, not the store, so it275// can run before anything is written.276fn combine_clashes(store &storage.Store, fetched Fetched) ![]Clash {277	base_files := if fetched.base != '' {278		history.files_of(store, fetched.base)!279	} else {280		[]snapshot.FileEntry{}281	}282	head := history.head(store) or { return []Clash{} }283	// The transaction applies the incoming strips before it replays, so the284	// clashes are found on the trees as they will be then, or the answers285	// would be about files that are no longer there.286	strips := fetched.events_out.filter(it.kind == events.kind_strip)287	mut clashes := find_clashes(after_strips(base_files, fetched.base, strips), after_strips(history.files_of(store,288		head.id)!, head.id, strips), after_strips(to_files(fetched.saves.last()), fetched.head,289		strips))290	for mut clash in clashes {291		mine := fetched_bytes(store, fetched, clash.mine)!292		theirs := fetched_bytes(store, fetched, clash.theirs)!293		delta := snapshot.diff_bytes(clash.path, 'modified', mine, theirs)!294		clash.stat = if delta.binary { 'binary' } else { '+${delta.added} -${delta.removed}' }295	}296	return clashes297}298299// after_strips is a save's file list once the given strip events are applied.300fn after_strips(files []snapshot.FileEntry, save_id string, strips []events.Event) []snapshot.FileEntry {301	return files.filter(fn [strips, save_id] (file snapshot.FileEntry) bool {302		for strip in strips {303			if (strip.save_id == '' || strip.save_id == save_id) && covers(strip.path, file.path) {304				return false305			}306		}307		return true308	})309}310311// fetched_bytes reads one side of a clash: from the fetch when it has just312// arrived, else from the store. An absent side is empty.313fn fetched_bytes(store &storage.Store, fetched Fetched, entry snapshot.FileEntry) ![]u8 {314	if entry.hash == '' {315		return []u8{}316	}317	if blob := fetched.repo_objects[entry.hash] {318		return zlib.decompress(blob)!319	}320	return store.get_object(entry.hash)!321}322323fn choose_sides(clashes []Clash, opts StealOptions) !map[string]Side {324	mut sides := map[string]Side{}325	if clashes.len == 0 {326		return sides327	}328	for clash in clashes {329		if opts.mine {330			sides[clash.path] = .mine331			continue332		}333		if opts.theirs {334			sides[clash.path] = .theirs335			continue336		}337		if opts.ask != unsafe { nil } {338			sides[clash.path] = opts.ask(clash)339			continue340		}341		return error('${clashes.len} files changed on both machines — run on a terminal to choose, or pass --mine or --theirs')342	}343	return sides344}345346fn same(a snapshot.FileEntry, b snapshot.FileEntry) bool {347	if a.hash == '' && b.hash == '' {348		return true349	}350	return a.hash == b.hash && a.executable == b.executable351}352353fn changed_since(store &storage.Store, before ?history.Save, strips_before int) !bool {354	if events.strip_events(store)!.len != strips_before {355		return true356	}357	now := history.head(store)358	if now or { history.Save{} }.id != before or { history.Save{} }.id {359		return true360	}361	return dirty(store, now)!362}363364// dirty reports whether the working tree holds anything a steal would365// overwrite. With no saves yet, that is every file but the untouched ignore366// list `koh init` wrote.367fn dirty(store &storage.Store, head ?history.Save) !bool {368	files := snapshot.walk(store, store.root, snapshot.load(store.root))!369	if saved := head {370		return !snapshot.same_tree(history.files_of(store, saved.id)!, files)371	}372	seed := storage.hash_hex(snapshot.default_ignore.bytes())373	return files.any(it.path != snapshot.ignore_file || it.hash != seed)374}375376// fetch_repo pulls a whole repo — head, saves, objects and events — for377// `koh steal <name>`, which lands a copy in a new folder.378pub fn fetch_repo(mut backend remote.Backend, name string) !Fetched {379	remote_head := backend.get(head_key(name))!380	if !remote_head.found {381		return error('the remote has no repo called "${name}"')382	}383	save_ids := backend.list('${name}/saves')!384	object_keys := backend.list('${name}/objects')!385	event_keys := backend.list('${name}/events')!386	mut fetched := Fetched{387		head:            remote_head.body.bytestr().trim_space()388		remote_save_ids: map[string]bool{}389	}390	for key in object_keys {391		blob := backend.get(key)!392		if blob.found {393			fetched.repo_objects[os.base(key)] = blob.body394		}395	}396	for key in save_ids {397		body := backend.get(key)!398		if !body.found {399			continue400		}401		record := decode_save(body.body)!402		fetched.remote_save_ids[record.id] = true403		fetched.saves << record404	}405	fetched.saves = parent_first(fetched.saves)!406	for key in event_keys {407		body := backend.get(key)!408		if body.found {409			fetched.events_out << decode_event(body.body)!410		}411	}412	return fetched413}414415// parent_first orders saves so every parent lands before its children, as the416// foreign key on `parent` demands. Time cannot do it: a replayed save keeps its417// original, older time but sits on a newer parent.418fn parent_first(records []SaveRecord) ![]SaveRecord {419	mut by_id := map[string]SaveRecord{}420	for record in records {421		by_id[record.id] = record422	}423	mut placed := map[string]bool{}424	mut ordered := []SaveRecord{}425	for record in records {426		mut chain := []SaveRecord{}427		mut cursor := record.id428		for cursor != '' && cursor !in placed {429			if chain.len > records.len {430				return error('the save records on the remote form a loop')431			}432			current := by_id[cursor] or { return error('the remote is missing save ${cursor}') }433			chain << current434			cursor = current.parent435		}436		chain.reverse_in_place()437		for current in chain {438			placed[current.id] = true439			ordered << current440		}441	}442	return ordered443}