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 json214import os15import src.events16import src.fmt17import src.history18import src.remote19import src.snapshot20import src.storage2122// Where the repo's name on the remote is remembered.23const repo_state_key = 'remote_repo'2425// The remote head as of this machine's last successful offer: a record for the26// person or a future feature. Nothing depends on it.27const remote_head_key = 'remote_head'2829pub const no_saves = 'this repo has no saves yet — nothing to offer'3031// The only remote layout this koh reads or writes. A repo offered before the32// key existed has no `format` key and counts as this version.33const format_value = 'koh-remote 1'3435// put_format records the layout on a repo that has none. put_if_absent never36// overwrites, so a machine that wrote one while this one read wins the race:37// its value is read back and checked, never assumed.38fn put_format(mut backend remote.Backend, repo string) ! {39	if backend.put_if_absent(format_key(repo), format_value.bytes())! {40		return41	}42	check_format(mut backend, repo)!43}4445// check_format reads <repo>/format and refuses a remote written by a koh that46// speaks another layout, so two versions never half-understand one bucket. An47// empty answer means the key is missing: version 1, which `offer` records.48fn check_format(mut backend remote.Backend, repo string) !string {49	existing := backend.get(format_key(repo))!50	if !existing.found {51		return ''52	}53	value := existing.body.bytestr().trim_space()54	if value != format_value {55		return error('the remote "${repo}" is "${value}" — this koh speaks "${format_value}"; update koh')56	}57	return value58}5960pub struct OfferReport {61pub mut:62	repo    string63	saves   int64	objects int65	events  int66	head    string67}6869// remember_repo binds a repo to its name on the remote. A steal into a new folder does it at70// once: its folder may be renamed before it ever offers.71pub fn remember_repo(store &storage.Store, name string) ! {72	store.set_state(repo_state_key, name)!73}7475// repo_name is the folder's name at first offer, lowercased so it makes the same76// Kepr URL whatever the folder's case (June, 2026-10-01), and remembered in the77// repo so renaming the folder later changes nothing.78pub fn repo_name(store &storage.Store, mut backend remote.Backend) !string {79	if remembered := store.state(repo_state_key) {80		return follow_rename(store, mut backend, remembered)!81	}82	name := os.base(store.root).to_lower()83	// Before the clash check below: a remote this koh cannot read is a "update84	// koh" answer, not a "rename the folder" one.85	check_format(mut backend, name)!86	existing := backend.get(head_key(name))!87	if existing.found {88		theirs := existing.body.bytestr().trim_space()89		if history.by_id(store, theirs) == none {90			return error('the remote already has "${name}" with a history this repo does not share — rename the folder, or use another remote')91		}92	}93	store.set_state(repo_state_key, name)!94	return name95}9697// follow_rename follows the marker Kepr leaves when it renames a repo, `.kepr/renamed/<old>`98// holding the new name (§30, 2026-10-01), for as long as the remembered name has no `head`, and99// remembers where it led. A few hops cover a repo renamed more than once.100fn follow_rename(store &storage.Store, mut backend remote.Backend, remembered string) !string {101	mut name := remembered102	for _ in 0 .. 5 {103		if backend.get(head_key(name))!.found {104			break105		}106		marker := backend.get('.kepr/renamed/${name}')!107		next := marker.body.bytestr().trim_space()108		if !marker.found || next == '' || next.contains('/') || next.contains('\\') || next in [109			'.',110			'..',111		] {112			break113		}114		name = next115	}116	if name != remembered {117		store.set_state(repo_state_key, name)!118	}119	return name120}121122// offer sends what the remote lacks and then swaps `head`. Nothing is ever123// queued: the next offer simply sends everything the remote does not have.124pub fn offer(store &storage.Store, mut backend remote.Backend) !OfferReport {125	name := repo_name(store, mut backend)!126	local := history.head(store) or { return error(no_saves) }127	mut report := OfferReport{128		repo: name129		head: local.id130	}131	if check_format(mut backend, name)! == '' {132		// Offered before this key existed: record the layout now.133		put_format(mut backend, name)!134	}135	remote_head := backend.get(head_key(name))!136	// Records offered by a machine that has not applied a strip would put the137	// stripped path back on the remote, where nothing would rewrite it again.138	if unapplied_strips(store, mut backend, name)!.len > 0 {139		return error(strip_not_applied)140	}141	mut uploaded := Uploaded{}142	upload_line(store, mut backend, name, local.id, remote_head, mut report, mut uploaded) or {143		return after_failed_upload(store, mut backend, name, uploaded, err)144	}145	// A strip sends its event before it lists the saves to rewrite, so one146	// missed here lists them after these records landed and rewrites them. One147	// seen here may have rewritten before they landed: take back what it reaches.148	if undo_late_strips(store, mut backend, name, uploaded)! {149		return error(strip_not_applied)150	}151	if !backend.put_if_match(head_key(name), local.id.bytes(), remote_head.tag)! {152		return error('remote has new saves — run koh steal')153	}154	store.set_state(remote_head_key, local.id)!155	return report156}157158// saves_after lists the local saves newer than `theirs`, oldest first. It fails159// when `theirs` is not on this line, which is the honest answer: we cannot160// offer on top of a save we do not have.161fn saves_after(store &storage.Store, ours string, theirs string) ![]history.Save {162	mut newer := []history.Save{}163	mut cursor := ours164	for cursor != '' && cursor != theirs {165		save := history.by_id(store, cursor) or { return error('save ${cursor} is missing') }166		newer << save167		cursor = save.parent168	}169	if cursor != theirs {170		return error('${theirs} is not on this line')171	}172	newer.reverse_in_place()173	return newer174}175176const strip_not_applied = 'the remote has a strip this repo has not applied — run koh steal'177178// Uploaded is what one upload put on the remote, so a late strip can take it back.179struct Uploaded {180mut:181	saves   []string182	objects map[string]bool183}184185fn (u Uploaded) done() []string {186	return if u.saves.len + u.objects.len == 0 {187		[]string{}188	} else {189		['uploaded ${fmt.plural(u.objects.len, 'object')} and ${fmt.plural(u.saves.len, 'save record')}']190	}191}192193// upload_line sends every save after the remote head, then the events.194fn upload_line(store &storage.Store, mut backend remote.Backend, name string, local string, remote_head remote.Object, mut report OfferReport, mut uploaded Uploaded) ! {195	if remote_head.found {196		theirs := remote_head.body.bytestr().trim_space()197		newer := saves_after(store, local, theirs) or {198			return error('remote has new saves — run koh steal')199		}200		for save in newer {201			upload_save(store, mut backend, name, save, mut report, mut uploaded)!202		}203	} else {204		oldest_first := history.line(store)!205		for index := oldest_first.len - 1; index >= 0; index-- {206			upload_save(store, mut backend, name, oldest_first[index], mut report, mut uploaded)!207		}208	}209	upload_events(store, mut backend, name, mut report)!210}211212// after_failed_upload still applies a strip that landed before the failure213// to what the upload had put.214fn after_failed_upload(store &storage.Store, mut backend remote.Backend, repo string, uploaded Uploaded, failure IError) IError {215	undo_late_strips(store, mut backend, repo, uploaded) or {216		return error('${failure.msg()}; removing what a strip on the remote covers also failed: ${err.msg()}')217	}218	return failure219}220221// unsettled_key keeps what an upload sent when settling a late strip failed,222// for Doctor to finish once a steal brings the strip here.223pub const unsettled_key = 'upload_unsettled'224225pub fn unsettled(store &storage.Store) ![]string {226	text := store.state(unsettled_key) or { return []string{} }227	return json2.decode[[]string](text) or {228		return error('the note of an unsettled upload is unreadable; it is kept for recovery')229	}230}231232fn undo_late_strips(store &storage.Store, mut backend remote.Backend, repo string, uploaded Uploaded) !bool {233	return settle_late_strips(store, mut backend, repo, uploaded) or {234		failure := err235		mut kept := unsettled(store) or { return error('${failure.msg()}; ${err.msg()}') }236		kept << uploaded.objects.keys().filter(it !in kept)237		store.set_state(unsettled_key, json2.encode(kept)) or {238			return error('${failure.msg()}; ${err.msg()}')239		}240		return failure241	}242}243244// settle_late_strips applies strips that reached the remote during an upload245// to what it put, and reports whether there were any. Records stay, less the246// stripped paths, as another offer or head may name them. Notes go first and247// objects last, so a stop part way leaves a note and no record without its object.248fn settle_late_strips(store &storage.Store, mut backend remote.Backend, repo string, uploaded Uploaded) !bool {249	strips := unapplied_strips(store, mut backend, repo)!250	if strips.len == 0 {251		return false252	}253	for strip in strips {254		queue_strip(store, strip.path, [strip.save_id], uploaded.objects.keys())!255	}256	for id in uploaded.saves {257		strip_record(mut backend, save_key(repo, id), id, strips)!258	}259	drop_stripped_objects(store, mut backend, repo, uploaded, strips)!260	return true261}262263// strip_record rewrites one remote record without what `strips` cover. A264// record that moved meanwhile is read again, since the other writer may have265// kept a path these strips name.266fn strip_record(mut backend remote.Backend, key string, id string, strips []events.Event) ! {267	for _ in 0 .. 5 {268		object := backend.get(key)!269		if !object.found {270			return271		}272		record := decode_save(object.body)!273		kept := record.files.filter(fn [strips, id] (file FileRecord) bool {274			return !strips.any(reaches(it, id) && covers(it.path, file.path))275		})276		if kept.len == record.files.len277			|| backend.put_if_match(key, encode_save(rewrite(record, kept)).bytes(), object.tag)! {278			return279		}280	}281	return error('the remote copy of save ${id} kept changing while a stripped path was removed; run koh doctor again')282}283284// Partway is a failure after some work had already landed. `done` says what,285// so Doctor can report both.286pub struct Partway {287	Error288pub:289	done   string290	reason string291}292293pub fn (e Partway) msg() string {294	return e.reason295}296297// done_text is what a step finished before it failed, if anything.298pub fn done_text(verb string, done int, noun string) []string {299	return if done == 0 { []string{} } else { ['${verb} ${fmt.plural(done, noun)}'] }300}301302// partway adds what was `done` before `failure` to it, including what a303// failing step of its own had done.304pub fn partway(done []string, failure IError) IError {305	mut all := done.clone()306	if failure is Partway {307		all << failure.done308	}309	if all.len == 0 {310		return failure311	}312	return Partway{313		done:   all.join('; ')314		reason: failure.msg()315	}316}317318// drop_stripped_objects deletes what an upload put that a strip reaches in a319// save here, unless a remote record lists it under a path no strip covers:320// another machine may hold the same bytes there.321fn drop_stripped_objects(store &storage.Store, mut backend remote.Backend, repo string, uploaded Uploaded, strips []events.Event) ! {322	mut reached := map[string]bool{}323	// A local save that still names it under another path may have no other copy.324	mut kept := map[string]bool{}325	for save in history.all_saves(store)! {326		for file in history.files_of(store, save.id)!.filter(it.hash in uploaded.objects) {327			if strips.any(reaches(it, save.id) && covers(it.path, file.path)) {328				reached[file.hash] = true329			} else {330				kept[file.hash] = true331			}332		}333	}334	if reached.len == 0 {335		return336	}337	for key in backend.list('${repo}/saves')! {338		object := backend.get(key)!339		if object.found {340			record := decode_save(object.body)!341			for file in record.files {342				if !strips.any(reaches(it, record.id) && covers(it.path, file.path)) {343					reached.delete(file.hash)344				}345			}346		}347	}348	for hash, _ in reached {349		if hash !in kept {350			backend.delete(object_key(repo, hash))!351		}352	}353}354355// upload_save puts a save's new objects, then its record, so a record on the remote always finds its356// objects there. Both are put_if_absent, so a retry after a partial offer is safe; `uploaded` keeps357// what this call created, which is all undo_late_strips may delete.358fn upload_save(store &storage.Store, mut backend remote.Backend, repo string, save history.Save, mut report OfferReport, mut uploaded Uploaded) ! {359	files := history.files_of(store, save.id)!360	// Saves go up oldest first and head moves only once all of them landed, so361	// the parent and its objects are on the remote already: only content new362	// in this save travels, not a round trip per file in the repo.363	mut sent := map[string]bool{}364	if save.parent != '' {365		for file in history.files_of(store, save.parent)! {366			sent[file.hash] = true367		}368	}369	for file in files {370		if file.hash in sent {371			continue372		}373		sent[file.hash] = true374		if backend.put_if_absent(object_key(repo, file.hash), store.object_blob(file.hash)!)! {375			report.objects++376			uploaded.objects[file.hash] = true377		}378	}379	record := save_record(store, save)!380	if backend.put_if_absent(save_key(repo, save.id), encode_save(record).bytes())! {381		report.saves++382		uploaded.saves << save.id383	}384}385386// upload_events sends every event the remote does not have. Events are tiny, so387// listing them costs less than tracking which ones were sent.388fn upload_events(store &storage.Store, mut backend remote.Backend, repo string, mut report OfferReport) ! {389	already := backend.list(events_prefix(repo))!390	mut already_sent := map[string]bool{}391	for key in already {392		already_sent[os.base(key)] = true393	}394	for event in events.all(store)! {395		if event.id in already_sent {396			continue397		}398		if backend.put_if_absent(event_key(repo, event.id), encode_event(event).bytes())! {399			report.events++400		}401	}402}403404// fetched holds everything a steal read from the remote, before any of it is405// written locally. Network work happens first; the database gets one406// transaction, so a failure leaves nothing half-applied.407pub struct Fetched {408pub mut:409	repo_objects map[string][]u8410	saves        []SaveRecord // oldest first411	events_out   []events.Event412	head         string413	base         string // the newest save both sides already shared, '' when none414	unsafe_paths []string415}416417// fetch_back_to_known walks from the remote head along parents until it meets a save on this line418// (`base`), reading only the objects this repo lacks. Nothing is written here: write_fetched lands it.419// A save held locally but off the line, from an earlier steal, is taken from the local copy.420pub fn fetch_back_to_known(store &storage.Store, mut backend remote.Backend, repo string) !Fetched {421	check_format(mut backend, repo)!422	remote_head := backend.get(head_key(repo))!423	if !remote_head.found {424		return error('the remote has no "${repo}" yet — run koh offer here first')425	}426	mut fetched := Fetched{427		head: remote_head.body.bytestr().trim_space()428	}429	mut on_line := map[string]bool{}430	for save in history.line(store)! {431		on_line[save.id] = true432	}433	// Strips come first: an object only a stripped path named may be gone.434	fetched.events_out = fetch_events(store, mut backend, repo)!435	strips := known_strips(store, fetched)!436	mut cursor := fetched.head437	for cursor != '' && cursor !in on_line {438		mut record := if saved := history.by_id(store, cursor) {439			save_record(store, saved)!440		} else {441			record_object := backend.get(save_key(repo, cursor))!442			if !record_object.found {443				return error('the remote is missing save ${cursor} — run koh offer there')444			}445			decode_save(record_object.body)!446		}447		fetched.unsafe_paths << drop_unsafe_paths(mut record)448		for file in record.files {449			if !store.has_object(file.hash)! && file.hash !in fetched.repo_objects450				&& !strips.any(reaches(it, record.id) && covers(it.path, file.path)) {451				blob := backend.get(object_key(repo, file.hash))!452				if !blob.found {453					return error('the remote is missing object ${file.hash} that save ${record.id} needs')454				}455				fetched.repo_objects[file.hash] = blob.body456			}457		}458		fetched.saves << record459		cursor = record.parent460	}461	fetched.saves.reverse_in_place()462	fetched.base = cursor463	return fetched464}465466fn fetch_events(store &storage.Store, mut backend remote.Backend, repo string) ![]events.Event {467	have := events.all(store)!468	mut known := map[string]bool{}469	for event in have {470		known[event.id] = true471	}472	mut incoming := []events.Event{}473	for key in backend.list(events_prefix(repo))! {474		id := os.base(key)475		if id in known {476			continue477		}478		body := backend.get(event_key(repo, id))!479		if !body.found {480			continue481		}482		event := decode_event(body.body)!483		if !ignored_strip(event) {484			incoming << event485		}486	}487	return incoming488}489490// write_fetched lands a fetch in one transaction: objects first (verified),491// then the events, then the saves they belong to. A strip bans its path from492// all of history, so it reaches every save, here or arriving, whenever made.493pub fn write_fetched(store &storage.Store, fetched Fetched) ! {494	for hash, blob in fetched.repo_objects {495		store.put_object_blob(hash, blob)!496	}497	mut stripped := false498	for event in fetched.events_out {499		if !events.receive(store, event)! {500			continue501		}502		if events.valid_strip(event) {503			remove_path(store, event.path, event.save_id)!504			stripped = true505		}506	}507	strips := known_strips(store, fetched)!508	for record in fetched.saves {509		import_arriving(store, record, strips)!510	}511	if stripped {512		store.drop_unreferenced_objects()!513	}514}515516// write_fetched_records makes network work durable without changing the line.517// Events wait for the combine transaction because a strip can rewrite history.518pub fn write_fetched_records(store &storage.Store, fetched Fetched) ! {519	for hash, blob in fetched.repo_objects {520		store.put_object_blob(hash, blob)!521	}522	strips := known_strips(store, fetched)!523	for record in fetched.saves {524		import_arriving(store, record, strips)!525	}526}527528// known_strips is every valid strip held here and every one the fetch brings:529// an arriving save loses what any of them reaches, not only what arrived with it.530fn known_strips(store &storage.Store, fetched Fetched) ![]events.Event {531	mut strips := events.strip_events(store)!532	strips << fetched.events_out533	return strips.filter(events.valid_strip(it))534}535536// import_arriving lands one save from the remote, unless it is already here, less every path a537// strip that `reaches` it removed.538fn import_arriving(store &storage.Store, record SaveRecord, strips []events.Event) ! {539	if history.by_id(store, record.id) != none {540		return541	}542	files := to_files(record).filter(fn [strips, record] (file snapshot.FileEntry) bool {543		return snapshot.legal_path(file.path)544			&& !strips.any(reaches(it, record.id) && covers(it.path, file.path))545	})546	history.import_save(store, to_save(record), files)!547}548549// reaches reports whether a strip event applies to a save: its one save, or550// every save there is, whenever it was made.551fn reaches(strip events.Event, save_id string) bool {552	return strip.save_id == '' || strip.save_id == save_id553}