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.fmt16import src.history17import src.remote18import src.snapshot19import src.storage2021// Where the repo's name on the remote is remembered.22const repo_state_key = 'remote_repo'2324// The remote head as of this machine's last successful offer: a record for the25// person or a future feature. Nothing depends on it.26const remote_head_key = 'remote_head'2728pub const no_saves = 'this repo has no saves yet — nothing to offer'2930// The only remote layout this koh reads or writes. A repo offered before the31// key existed has no `format` key and counts as this version.32const format_value = 'koh-remote 1'3334// put_format records the layout on a repo that has none. put_if_absent never35// overwrites, so a machine that wrote one while this one read wins the race:36// its value is read back and checked, never assumed.37fn put_format(mut backend remote.Backend, repo string) ! {38	if backend.put_if_absent(format_key(repo), format_value.bytes())! {39		return40	}41	check_format(mut backend, repo)!42}4344// check_format reads <repo>/format and refuses a remote written by a koh that45// speaks another layout, so two versions never half-understand one bucket. An46// empty answer means the key is missing: version 1, which `offer` records.47fn check_format(mut backend remote.Backend, repo string) !string {48	existing := backend.get(format_key(repo))!49	if !existing.found {50		return ''51	}52	value := existing.body.bytestr().trim_space()53	if value != format_value {54		return error('the remote "${repo}" is "${value}" — this koh speaks "${format_value}"; update koh')55	}56	return value57}5859pub struct OfferReport {60pub mut:61	repo    string62	saves   int63	objects int64	events  int65	head    string66}6768// remember_repo binds a repo to its name on the remote. A steal into a new folder does it at69// once: its folder may be renamed before it ever offers.70pub fn remember_repo(store &storage.Store, name string) ! {71	store.set_state(repo_state_key, name)!72}7374// repo_name is the folder's name at first offer, lowercased so it makes the same75// Kepr URL whatever the folder's case (June, 2026-10-01), and remembered in the76// repo so renaming the folder later changes nothing.77pub fn repo_name(store &storage.Store, mut backend remote.Backend) !string {78	if remembered := store.state(repo_state_key) {79		return follow_rename(store, mut backend, remembered)!80	}81	name := os.base(store.root).to_lower()82	// Before the clash check below: a remote this koh cannot read is a "update83	// koh" answer, not a "rename the folder" one.84	check_format(mut backend, name)!85	existing := backend.get(head_key(name))!86	if existing.found {87		theirs := existing.body.bytestr().trim_space()88		if history.by_id(store, theirs) == none {89			return error('the remote already has "${name}" with a history this repo does not share — rename the folder, or use another remote')90		}91	}92	store.set_state(repo_state_key, name)!93	return name94}9596// follow_rename follows the marker Kepr leaves when it renames a repo, `.kepr/renamed/<old>`97// holding the new name (§30, 2026-10-01), for as long as the remembered name has no `head`, and98// remembers where it led. A few hops cover a repo renamed more than once.99fn follow_rename(store &storage.Store, mut backend remote.Backend, remembered string) !string {100	mut name := remembered101	for _ in 0 .. 5 {102		if backend.get(head_key(name))!.found {103			break104		}105		marker := backend.get('.kepr/renamed/${name}')!106		next := marker.body.bytestr().trim_space()107		if !marker.found || next == '' || next.contains('/') || next.contains('\\') || next in [108			'.',109			'..',110		] {111			break112		}113		name = next114	}115	if name != remembered {116		store.set_state(repo_state_key, name)!117	}118	return name119}120121// offer sends what the remote lacks and then swaps `head`. Nothing is ever122// queued: the next offer simply sends everything the remote does not have.123pub fn offer(store &storage.Store, mut backend remote.Backend) !OfferReport {124	name := repo_name(store, mut backend)!125	local := history.head(store) or { return error(no_saves) }126	mut report := OfferReport{127		repo: name128		head: local.id129	}130	if check_format(mut backend, name)! == '' {131		// Offered before this key existed: record the layout now.132		put_format(mut backend, name)!133	}134	remote_head := backend.get(head_key(name))!135	// Records offered by a machine that has not applied a strip would put the136	// stripped path back on the remote, where nothing would rewrite it again.137	if unapplied_strips(store, mut backend, name)!.len > 0 {138		return error(strip_not_applied)139	}140	mut uploaded := Uploaded{}141	upload_line(store, mut backend, name, local.id, remote_head, mut report, mut uploaded) or {142		return after_failed_upload(store, mut backend, name, uploaded, err)143	}144	// A strip sends its event before it lists the saves to rewrite, so one145	// missed here lists them after these records landed and rewrites them. One146	// seen here may have rewritten before they landed: take back what it reaches.147	if undo_late_strips(store, mut backend, name, uploaded)! {148		return error(strip_not_applied)149	}150	if !backend.put_if_match(head_key(name), local.id.bytes(), remote_head.tag)! {151		return error('remote has new saves — run koh steal')152	}153	store.set_state(remote_head_key, local.id)!154	return report155}156157// saves_after lists the local saves newer than `theirs`, oldest first. It fails158// when `theirs` is not on this line, which is the honest answer: we cannot159// offer on top of a save we do not have.160fn saves_after(store &storage.Store, ours string, theirs string) ![]history.Save {161	mut newer := []history.Save{}162	mut cursor := ours163	for cursor != '' && cursor != theirs {164		save := history.by_id(store, cursor) or { return error('save ${cursor} is missing') }165		newer << save166		cursor = save.parent167	}168	if cursor != theirs {169		return error('${theirs} is not on this line')170	}171	newer.reverse_in_place()172	return newer173}174175const strip_not_applied = 'the remote has a strip this repo has not applied — run koh steal'176177// Uploaded is what one upload put on the remote, so a late strip can take it back.178struct Uploaded {179mut:180	saves   []string181	objects map[string]bool182}183184fn (u Uploaded) done() []string {185	return if u.saves.len + u.objects.len == 0 {186		[]string{}187	} else {188		['uploaded ${fmt.plural(u.objects.len, 'object')} and ${fmt.plural(u.saves.len, 'save record')}']189	}190}191192// upload_line sends every save after the remote head, then the events.193fn upload_line(store &storage.Store, mut backend remote.Backend, name string, local string, remote_head remote.Object, mut report OfferReport, mut uploaded Uploaded) ! {194	if remote_head.found {195		theirs := remote_head.body.bytestr().trim_space()196		newer := saves_after(store, local, theirs) or {197			return error('remote has new saves — run koh steal')198		}199		for save in newer {200			upload_save(store, mut backend, name, save, mut report, mut uploaded)!201		}202	} else {203		oldest_first := history.line(store)!204		for index := oldest_first.len - 1; index >= 0; index-- {205			upload_save(store, mut backend, name, oldest_first[index], mut report, mut uploaded)!206		}207	}208	upload_events(store, mut backend, name, mut report)!209}210211// after_failed_upload still applies a strip that landed before the failure212// to what the upload had put.213fn after_failed_upload(store &storage.Store, mut backend remote.Backend, repo string, uploaded Uploaded, failure IError) IError {214	undo_late_strips(store, mut backend, repo, uploaded) or {215		return error('${failure.msg()}; removing what a strip on the remote covers also failed: ${err.msg()}')216	}217	return failure218}219220// undo_late_strips applies strips that reached the remote during an upload to221// what the upload put, and reports whether there were any. Records stay, less222// the stripped paths: another offer, or head itself, may already name them.223// Each strip's note is written first, so if this stops part way, Doctor sends224// the strip again once a steal brings it here. Objects go last, so no record225// is left naming one that is gone.226fn undo_late_strips(store &storage.Store, mut backend remote.Backend, repo string, uploaded Uploaded) !bool {227	strips := unapplied_strips(store, mut backend, repo)!228	if strips.len == 0 {229		return false230	}231	for strip in strips {232		queue_strip(store, strip.path, [strip.save_id], uploaded.objects.keys())!233	}234	for id in uploaded.saves {235		strip_record(mut backend, save_key(repo, id), id, strips)!236	}237	drop_stripped_objects(store, mut backend, repo, uploaded, strips)!238	return true239}240241// strip_record rewrites one remote record without what `strips` cover. A242// record that moved meanwhile is read again, since the other writer may have243// kept a path these strips name.244fn strip_record(mut backend remote.Backend, key string, id string, strips []events.Event) ! {245	for _ in 0 .. 5 {246		object := backend.get(key)!247		if !object.found {248			return249		}250		record := decode_save(object.body)!251		kept := record.files.filter(fn [strips, id] (file FileRecord) bool {252			return !strips.any(reaches(it, id) && covers(it.path, file.path))253		})254		if kept.len == record.files.len255			|| backend.put_if_match(key, encode_save(rewrite(record, kept)).bytes(), object.tag)! {256			return257		}258	}259	return error('the remote copy of save ${id} kept changing while a stripped path was removed; run koh doctor again')260}261262// Partway is a failure after some work had already landed. `done` says what,263// so Doctor can report both.264pub struct Partway {265	Error266pub:267	done   string268	reason string269}270271pub fn (e Partway) msg() string {272	return e.reason273}274275// partway adds what was `done` before `failure` to it, including what a276// failing step of its own had done.277pub fn partway(done []string, failure IError) IError {278	mut all := done.clone()279	if failure is Partway {280		all << failure.done281	}282	if all.len == 0 {283		return failure284	}285	return Partway{286		done:   all.join('; ')287		reason: failure.msg()288	}289}290291// drop_stripped_objects deletes what an upload put that a strip reaches in a292// save here, unless a remote record lists it under a path no strip covers:293// another machine may hold the same bytes there.294fn drop_stripped_objects(store &storage.Store, mut backend remote.Backend, repo string, uploaded Uploaded, strips []events.Event) ! {295	mut reached := map[string]bool{}296	for save in history.all_saves(store)! {297		for file in history.files_of(store, save.id)! {298			if file.hash in uploaded.objects && strips.any(reaches(it, save.id) && covers(it.path, file.path)) {299				reached[file.hash] = true300			}301		}302	}303	if reached.len == 0 {304		return305	}306	for key in backend.list('${repo}/saves')! {307		object := backend.get(key)!308		if object.found {309			record := decode_save(object.body)!310			for file in record.files {311				if !strips.any(reaches(it, record.id) && covers(it.path, file.path)) {312					reached.delete(file.hash)313				}314			}315		}316	}317	for hash, _ in reached {318		backend.delete(object_key(repo, hash))!319	}320}321322// upload_save puts a save's new objects, then its record, so a record on the remote always finds its323// objects there. Both are put_if_absent, so a retry after a partial offer is safe; `uploaded` keeps324// what this call created, which is all undo_late_strips may delete.325fn upload_save(store &storage.Store, mut backend remote.Backend, repo string, save history.Save, mut report OfferReport, mut uploaded Uploaded) ! {326	files := history.files_of(store, save.id)!327	// Saves go up oldest first and head moves only once all of them landed, so328	// the parent and its objects are on the remote already: only content new329	// in this save travels, not a round trip per file in the repo.330	mut sent := map[string]bool{}331	if save.parent != '' {332		for file in history.files_of(store, save.parent)! {333			sent[file.hash] = true334		}335	}336	for file in files {337		if file.hash in sent {338			continue339		}340		sent[file.hash] = true341		if backend.put_if_absent(object_key(repo, file.hash), store.object_blob(file.hash)!)! {342			report.objects++343			uploaded.objects[file.hash] = true344		}345	}346	record := save_record(store, save)!347	if backend.put_if_absent(save_key(repo, save.id), encode_save(record).bytes())! {348		report.saves++349		uploaded.saves << save.id350	}351}352353// upload_events sends every event the remote does not have. Events are tiny, so354// listing them costs less than tracking which ones were sent.355fn upload_events(store &storage.Store, mut backend remote.Backend, repo string, mut report OfferReport) ! {356	already := backend.list(events_prefix(repo))!357	mut already_sent := map[string]bool{}358	for key in already {359		already_sent[os.base(key)] = true360	}361	for event in events.all(store)! {362		if event.id in already_sent {363			continue364		}365		if backend.put_if_absent(event_key(repo, event.id), encode_event(event).bytes())! {366			report.events++367		}368	}369}370371// fetched holds everything a steal read from the remote, before any of it is372// written locally. Network work happens first; the database gets one373// transaction, so a failure leaves nothing half-applied.374pub struct Fetched {375pub mut:376	repo_objects map[string][]u8377	saves        []SaveRecord // oldest first378	events_out   []events.Event379	head         string380	base         string // the newest save both sides already shared, '' when none381	unsafe_paths []string382}383384// fetch_back_to_known walks from the remote head along parents until it meets a save on this line385// (`base`), reading only the objects this repo lacks. Nothing is written here: write_fetched lands it.386// A save held locally but off the line, from an earlier steal, is taken from the local copy.387pub fn fetch_back_to_known(store &storage.Store, mut backend remote.Backend, repo string) !Fetched {388	check_format(mut backend, repo)!389	remote_head := backend.get(head_key(repo))!390	if !remote_head.found {391		return error('the remote has no "${repo}" yet — run koh offer here first')392	}393	mut fetched := Fetched{394		head: remote_head.body.bytestr().trim_space()395	}396	mut on_line := map[string]bool{}397	for save in history.line(store)! {398		on_line[save.id] = true399	}400	// Strips come first: an object only a stripped path named may be gone.401	fetched.events_out = fetch_events(store, mut backend, repo)!402	strips := known_strips(store, fetched)!403	mut cursor := fetched.head404	for cursor != '' && cursor !in on_line {405		mut record := if saved := history.by_id(store, cursor) {406			save_record(store, saved)!407		} else {408			record_object := backend.get(save_key(repo, cursor))!409			if !record_object.found {410				return error('the remote is missing save ${cursor} — run koh offer there')411			}412			decode_save(record_object.body)!413		}414		fetched.unsafe_paths << drop_unsafe_paths(mut record)415		for file in record.files {416			if !store.has_object(file.hash)! && file.hash !in fetched.repo_objects417				&& !strips.any(reaches(it, record.id) && covers(it.path, file.path)) {418				blob := backend.get(object_key(repo, file.hash))!419				if !blob.found {420					return error('the remote is missing object ${file.hash} that save ${record.id} needs')421				}422				fetched.repo_objects[file.hash] = blob.body423			}424		}425		fetched.saves << record426		cursor = record.parent427	}428	fetched.saves.reverse_in_place()429	fetched.base = cursor430	return fetched431}432433fn fetch_events(store &storage.Store, mut backend remote.Backend, repo string) ![]events.Event {434	have := events.all(store)!435	mut known := map[string]bool{}436	for event in have {437		known[event.id] = true438	}439	mut incoming := []events.Event{}440	for key in backend.list(events_prefix(repo))! {441		id := os.base(key)442		if id in known {443			continue444		}445		body := backend.get(event_key(repo, id))!446		if !body.found {447			continue448		}449		event := decode_event(body.body)!450		if !ignored_strip(event) {451			incoming << event452		}453	}454	return incoming455}456457// write_fetched lands a fetch in one transaction: objects first (verified),458// then the events, then the saves they belong to. A strip bans its path from459// all of history, so it reaches every save, here or arriving, whenever made.460pub fn write_fetched(store &storage.Store, fetched Fetched) ! {461	for hash, blob in fetched.repo_objects {462		store.put_object_blob(hash, blob)!463	}464	mut stripped := false465	for event in fetched.events_out {466		if !events.receive(store, event)! {467			continue468		}469		if events.valid_strip(event) {470			remove_path(store, event.path, event.save_id)!471			stripped = true472		}473	}474	strips := known_strips(store, fetched)!475	for record in fetched.saves {476		import_arriving(store, record, strips)!477	}478	if stripped {479		store.drop_unreferenced_objects()!480	}481}482483// write_fetched_records makes network work durable without changing the line.484// Events wait for the combine transaction because a strip can rewrite history.485pub fn write_fetched_records(store &storage.Store, fetched Fetched) ! {486	for hash, blob in fetched.repo_objects {487		store.put_object_blob(hash, blob)!488	}489	strips := known_strips(store, fetched)!490	for record in fetched.saves {491		import_arriving(store, record, strips)!492	}493}494495// known_strips is every valid strip held here and every one the fetch brings:496// an arriving save loses what any of them reaches, not only what arrived with it.497fn known_strips(store &storage.Store, fetched Fetched) ![]events.Event {498	mut strips := events.strip_events(store)!499	strips << fetched.events_out500	return strips.filter(events.valid_strip(it))501}502503// import_arriving lands one save from the remote, unless it is already here, less every path a504// strip that `reaches` it removed.505fn import_arriving(store &storage.Store, record SaveRecord, strips []events.Event) ! {506	if history.by_id(store, record.id) != none {507		return508	}509	files := to_files(record).filter(fn [strips, record] (file snapshot.FileEntry) bool {510		return snapshot.legal_path(file.path)511			&& !strips.any(reaches(it, record.id) && covers(it.path, file.path))512	})513	history.import_save(store, to_save(record), files)!514}515516// reaches reports whether a strip event applies to a save: its one save, or517// every save there is, whenever it was made.518fn reaches(strip events.Event, save_id string) bool {519	return strip.save_id == '' || strip.save_id == save_id520}