exchange.v

1//! Sync: offer and steal.2//!3//! All the cleverness sits here, above the five-operation remote interface, so4//! the backends stay thin and this logic is written once and tested offline.5//!6//! The remote is a line of saves, the same shape as the local one. `offer`7//! sends local saves the remote lacks; `steal` takes remote saves and, when8//! both sides moved, replays the local work on top of the remote's. Divergence9//! is resolved per file, never per line.1011module exchange1213import os14import src.events15import src.history16import src.remote17import src.snapshot18import src.storage1920// Where the repo's name on the remote is remembered.21const repo_state_key = 'remote_repo'2223// The remote head as of this machine's last successful offer: a record for the24// person or a future feature. Nothing depends on it.25const remote_head_key = 'remote_head'2627pub const no_saves = 'this repo has no saves yet — nothing to offer'2829// The only remote layout this koh reads or writes. A repo offered before the30// key existed has no `format` key and counts as this version.31const format_value = 'koh-remote 1'3233// put_format records the layout on a repo that has none. put_if_absent never34// overwrites, so a machine that wrote one while this one read wins the race:35// its value is read back and checked, never assumed.36fn put_format(mut backend remote.Backend, repo string) ! {37	if backend.put_if_absent(format_key(repo), format_value.bytes())! {38		return39	}40	check_format(mut backend, repo)!41}4243// check_format reads <repo>/format and refuses a remote written by a koh that44// speaks another layout, so two versions never half-understand one bucket. An45// empty answer means the key is missing: version 1, which `offer` records.46fn check_format(mut backend remote.Backend, repo string) !string {47	existing := backend.get(format_key(repo))!48	if !existing.found {49		return ''50	}51	value := existing.body.bytestr().trim_space()52	if value != format_value {53		return error('the remote "${repo}" is "${value}" — this koh speaks "${format_value}"; update koh')54	}55	return value56}5758pub struct OfferReport {59pub mut:60	repo    string61	saves   int62	objects int63	events  int64	head    string65}6667// remember_repo binds a repo to its name on the remote. A steal into a new folder does it at68// once: its folder may be renamed before it ever offers.69pub fn remember_repo(store &storage.Store, name string) ! {70	store.set_state(repo_state_key, name)!71}7273// repo_name is the folder's name at first offer, lowercased so it makes the same74// Kepr URL whatever the folder's case (June, 2026-10-01), and remembered in the75// repo so renaming the folder later changes nothing.76pub fn repo_name(store &storage.Store, mut backend remote.Backend) !string {77	if remembered := store.state(repo_state_key) {78		return follow_rename(store, mut backend, remembered)!79	}80	name := os.base(store.root).to_lower()81	// Before the clash check below: a remote this koh cannot read is a "update82	// koh" answer, not a "rename the folder" one.83	check_format(mut backend, name)!84	existing := backend.get(head_key(name))!85	if existing.found {86		theirs := existing.body.bytestr().trim_space()87		if history.by_id(store, theirs) == none {88			return error('the remote already has "${name}" with a history this repo does not share — rename the folder, or use another remote')89		}90	}91	store.set_state(repo_state_key, name)!92	return name93}9495// follow_rename follows the marker Kepr leaves when it renames a repo, `.kepr/renamed/<old>`96// holding the new name (§30, 2026-10-01), for as long as the remembered name has no `head`, and97// remembers where it led. A few hops cover a repo renamed more than once.98fn follow_rename(store &storage.Store, mut backend remote.Backend, remembered string) !string {99	mut name := remembered100	for _ in 0 .. 5 {101		if backend.get(head_key(name))!.found {102			break103		}104		marker := backend.get('.kepr/renamed/${name}')!105		next := marker.body.bytestr().trim_space()106		if !marker.found || next == '' || next.contains('/') || next.contains('\\') || next in [107			'.',108			'..',109		] {110			break111		}112		name = next113	}114	if name != remembered {115		store.set_state(repo_state_key, name)!116	}117	return name118}119120// offer sends what the remote lacks and then swaps `head`. Nothing is ever121// queued: the next offer simply sends everything the remote does not have.122pub fn offer(store &storage.Store, mut backend remote.Backend) !OfferReport {123	name := repo_name(store, mut backend)!124	local := history.head(store) or { return error(no_saves) }125	mut report := OfferReport{126		repo: name127		head: local.id128	}129	if check_format(mut backend, name)! == '' {130		// Offered before this key existed: record the layout now.131		put_format(mut backend, name)!132	}133	remote_head := backend.get(head_key(name))!134	// Records offered by a machine that has not applied a strip would put the135	// stripped path back on the remote, where nothing would rewrite it again.136	if unapplied_strips(store, mut backend, name)!.len > 0 {137		return error(strip_not_applied)138	}139	mut uploaded := Uploaded{}140	upload_line(store, mut backend, name, local.id, remote_head, mut report, mut uploaded) or {141		return after_failed_upload(store, mut backend, name, uploaded, false, err)142	}143	// A strip sends its event before it lists the saves to rewrite, so one144	// missed here lists them after these records landed and rewrites them. One145	// seen here may have rewritten before they landed: take back what it reaches.146	if undo_late_strips(store, mut backend, name, uploaded, false)! {147		return error(strip_not_applied)148	}149	if !backend.put_if_match(head_key(name), local.id.bytes(), remote_head.tag)! {150		return error('remote has new saves — run koh steal')151	}152	store.set_state(remote_head_key, local.id)!153	return report154}155156// saves_after lists the local saves newer than `theirs`, oldest first. It fails157// when `theirs` is not on this line, which is the honest answer: we cannot158// offer on top of a save we do not have.159fn saves_after(store &storage.Store, ours string, theirs string) ![]history.Save {160	mut newer := []history.Save{}161	mut cursor := ours162	for cursor != '' && cursor != theirs {163		save := history.by_id(store, cursor) or { return error('save ${cursor} is missing') }164		newer << save165		cursor = save.parent166	}167	if cursor != theirs {168		return error('${theirs} is not on this line')169	}170	newer.reverse_in_place()171	return newer172}173174const strip_not_applied = 'the remote has a strip this repo has not applied — run koh steal'175176// Uploaded is what one upload put on the remote, so a late strip can take it back.177struct Uploaded {178mut:179	saves   []string180	objects map[string]bool181}182183// upload_line sends every save after the remote head, then the events.184fn upload_line(store &storage.Store, mut backend remote.Backend, name string, local string, remote_head remote.Object, mut report OfferReport, mut uploaded Uploaded) ! {185	if remote_head.found {186		theirs := remote_head.body.bytestr().trim_space()187		newer := saves_after(store, local, theirs) or {188			return error('remote has new saves — run koh steal')189		}190		for save in newer {191			upload_save(store, mut backend, name, save, mut report, mut uploaded)!192		}193	} else {194		oldest_first := history.line(store)!195		for index := oldest_first.len - 1; index >= 0; index-- {196			upload_save(store, mut backend, name, oldest_first[index], mut report, mut uploaded)!197		}198	}199	upload_events(store, mut backend, name, mut report)!200}201202// after_failed_upload still applies a strip that landed before the failure203// to what the upload had put.204fn after_failed_upload(store &storage.Store, mut backend remote.Backend, repo string, uploaded Uploaded, keep_records bool, failure IError) IError {205	undo_late_strips(store, mut backend, repo, uploaded, keep_records) or {206		return error('${failure.msg()}; removing what a strip on the remote covers also failed: ${err.msg()}')207	}208	return failure209}210211// undo_late_strips applies strips that reached the remote during an upload to212// what the upload put, and reports whether there were any. An offer's records213// go, since head never reached them. Doctor's resent records may already be214// named by head, so they keep everything but the stripped paths; one that215// moved meanwhile was rewritten by the strip itself.216fn undo_late_strips(store &storage.Store, mut backend remote.Backend, repo string, uploaded Uploaded, keep_records bool) !bool {217	strips := unapplied_strips(store, mut backend, repo)!218	if strips.len == 0 {219		return false220	}221	for id in uploaded.saves {222		key := save_key(repo, id)223		if !keep_records {224			backend.delete(key)!225			continue226		}227		object := backend.get(key)!228		if !object.found {229			continue230		}231		record := decode_save(object.body)!232		kept := record.files.filter(fn [strips, id] (file FileRecord) bool {233			return !strips.any(reaches(it, id) && covers(it.path, file.path))234		})235		if kept.len < record.files.len {236			backend.put_if_match(key, encode_save(rewrite(record, kept)).bytes(), object.tag)!237		}238	}239	drop_stripped_objects(store, mut backend, repo, uploaded, strips)!240	return true241}242243// drop_stripped_objects deletes what an upload put that a strip reaches in a244// save here, unless a remote record still lists it: another machine may hold245// the same bytes under a path no strip names.246fn drop_stripped_objects(store &storage.Store, mut backend remote.Backend, repo string, uploaded Uploaded, strips []events.Event) ! {247	mut reached := map[string]bool{}248	for save in history.all_saves(store)! {249		for file in history.files_of(store, save.id)! {250			if file.hash in uploaded.objects && strips.any(reaches(it, save.id) && covers(it.path, file.path)) {251				reached[file.hash] = true252			}253		}254	}255	if reached.len == 0 {256		return257	}258	for key in backend.list('${repo}/saves')! {259		object := backend.get(key)!260		if object.found {261			for file in decode_save(object.body)!.files {262				reached.delete(file.hash)263			}264		}265	}266	for hash, _ in reached {267		backend.delete(object_key(repo, hash))!268	}269}270271// upload_save puts a save's new objects, then its record, so a record on the remote always finds its272// objects there. Both are put_if_absent, so a retry after a partial offer is safe; `uploaded` keeps273// what this call created, which is all undo_late_strips may delete.274fn upload_save(store &storage.Store, mut backend remote.Backend, repo string, save history.Save, mut report OfferReport, mut uploaded Uploaded) ! {275	files := history.files_of(store, save.id)!276	// Saves go up oldest first and head moves only once all of them landed, so277	// the parent and its objects are on the remote already: only content new278	// in this save travels, not a round trip per file in the repo.279	mut sent := map[string]bool{}280	if save.parent != '' {281		for file in history.files_of(store, save.parent)! {282			sent[file.hash] = true283		}284	}285	for file in files {286		if file.hash in sent {287			continue288		}289		sent[file.hash] = true290		if backend.put_if_absent(object_key(repo, file.hash), store.object_blob(file.hash)!)! {291			report.objects++292			uploaded.objects[file.hash] = true293		}294	}295	record := save_record(store, save)!296	if backend.put_if_absent(save_key(repo, save.id), encode_save(record).bytes())! {297		report.saves++298		uploaded.saves << save.id299	}300}301302// upload_events sends every event the remote does not have. Events are tiny, so303// listing them costs less than tracking which ones were sent.304fn upload_events(store &storage.Store, mut backend remote.Backend, repo string, mut report OfferReport) ! {305	already := backend.list(events_prefix(repo))!306	mut already_sent := map[string]bool{}307	for key in already {308		already_sent[os.base(key)] = true309	}310	for event in events.all(store)! {311		if event.id in already_sent {312			continue313		}314		if backend.put_if_absent(event_key(repo, event.id), encode_event(event).bytes())! {315			report.events++316		}317	}318}319320// fetched holds everything a steal read from the remote, before any of it is321// written locally. Network work happens first; the database gets one322// transaction, so a failure leaves nothing half-applied.323pub struct Fetched {324pub mut:325	repo_objects map[string][]u8326	saves        []SaveRecord // oldest first327	events_out   []events.Event328	head         string329	base         string // the newest save both sides already shared, '' when none330	unsafe_paths []string331}332333// fetch_back_to_known walks from the remote head along parents until it meets a save on this line334// (`base`), reading only the objects this repo lacks. Nothing is written here: write_fetched lands it.335// A save held locally but off the line, from an earlier steal, is taken from the local copy.336pub fn fetch_back_to_known(store &storage.Store, mut backend remote.Backend, repo string) !Fetched {337	check_format(mut backend, repo)!338	remote_head := backend.get(head_key(repo))!339	if !remote_head.found {340		return error('the remote has no "${repo}" yet — run koh offer here first')341	}342	mut fetched := Fetched{343		head: remote_head.body.bytestr().trim_space()344	}345	mut on_line := map[string]bool{}346	for save in history.line(store)! {347		on_line[save.id] = true348	}349	mut cursor := fetched.head350	for cursor != '' && cursor !in on_line {351		mut record := if saved := history.by_id(store, cursor) {352			save_record(store, saved)!353		} else {354			record_object := backend.get(save_key(repo, cursor))!355			if !record_object.found {356				return error('the remote is missing save ${cursor} — run koh offer there')357			}358			decode_save(record_object.body)!359		}360		fetched.unsafe_paths << drop_unsafe_paths(mut record)361		for file in record.files {362			if !store.has_object(file.hash)! && file.hash !in fetched.repo_objects {363				blob := backend.get(object_key(repo, file.hash))!364				if !blob.found {365					return error('the remote is missing object ${file.hash} that save ${record.id} needs')366				}367				fetched.repo_objects[file.hash] = blob.body368			}369		}370		fetched.saves << record371		cursor = record.parent372	}373	fetched.saves.reverse_in_place()374	fetched.base = cursor375	fetched.events_out = fetch_events(store, mut backend, repo)!376	return fetched377}378379fn fetch_events(store &storage.Store, mut backend remote.Backend, repo string) ![]events.Event {380	have := events.all(store)!381	mut known := map[string]bool{}382	for event in have {383		known[event.id] = true384	}385	mut incoming := []events.Event{}386	for key in backend.list(events_prefix(repo))! {387		id := os.base(key)388		if id in known {389			continue390		}391		body := backend.get(event_key(repo, id))!392		if !body.found {393			continue394		}395		event := decode_event(body.body)!396		if !ignored_strip(event) {397			incoming << event398		}399	}400	return incoming401}402403// write_fetched lands a fetch in one transaction: objects first (verified),404// then the events, then the saves they belong to. A strip bans its path from405// all of history, so it reaches every save, here or arriving, whenever made.406pub fn write_fetched(store &storage.Store, fetched Fetched) ! {407	for hash, blob in fetched.repo_objects {408		store.put_object_blob(hash, blob)!409	}410	mut stripped := false411	for event in fetched.events_out {412		if !events.receive(store, event)! {413			continue414		}415		if events.valid_strip(event) {416			remove_path(store, event.path, event.save_id)!417			stripped = true418		}419	}420	strips := known_strips(store, fetched)!421	for record in fetched.saves {422		import_arriving(store, record, strips)!423	}424	if stripped {425		store.drop_unreferenced_objects()!426	}427}428429// write_fetched_records makes network work durable without changing the line.430// Events wait for the combine transaction because a strip can rewrite history.431pub fn write_fetched_records(store &storage.Store, fetched Fetched) ! {432	for hash, blob in fetched.repo_objects {433		store.put_object_blob(hash, blob)!434	}435	strips := known_strips(store, fetched)!436	for record in fetched.saves {437		import_arriving(store, record, strips)!438	}439}440441// known_strips is every valid strip held here and every one the fetch brings:442// an arriving save loses what any of them reaches, not only what arrived with it.443fn known_strips(store &storage.Store, fetched Fetched) ![]events.Event {444	mut strips := events.strip_events(store)!445	strips << fetched.events_out446	return strips.filter(events.valid_strip(it))447}448449// import_arriving lands one save from the remote, unless it is already here, less every path a450// strip that `reaches` it removed.451fn import_arriving(store &storage.Store, record SaveRecord, strips []events.Event) ! {452	if history.by_id(store, record.id) != none {453		return454	}455	files := to_files(record).filter(fn [strips, record] (file snapshot.FileEntry) bool {456		return snapshot.legal_path(file.path)457			&& !strips.any(reaches(it, record.id) && covers(it.path, file.path))458	})459	history.import_save(store, to_save(record), files)!460}461462// reaches reports whether a strip event applies to a save: its one save, or463// every save there is, whenever it was made.464fn reaches(strip events.Event, save_id string) bool {465	return strip.save_id == '' || strip.save_id == save_id466}