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//! pushes local saves the remote lacks; `steal` pulls 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, as the document24// asks; nothing reads it back yet.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// check_format reads <repo>/format and refuses a remote written by a koh that34// speaks another layout, so two versions never half-understand one bucket. An35// empty answer means the key is missing: version 1, which `offer` records.36fn check_format(mut backend remote.Backend, repo string) !string {37	existing := backend.get(format_key(repo))!38	if !existing.found {39		return ''40	}41	value := existing.body.bytestr().trim_space()42	if value != format_value {43		return error('the remote "${repo}" is "${value}" — this koh speaks "${format_value}"; update koh')44	}45	return value46}4748pub struct OfferReport {49pub mut:50	repo    string51	saves   int52	objects int53	events  int54	head    string55}5657// remember_repo binds a repo to its name on the remote. A clone does it at58// once: its folder may be renamed before it ever offers.59pub fn remember_repo(store &storage.Store, name string) ! {60	store.set_state(repo_state_key, name)!61}6263// repo_name is the folder's name at first offer, remembered in the repo so64// renaming the folder later changes nothing.65pub fn repo_name(store &storage.Store, mut backend remote.Backend) !string {66	if remembered := store.state(repo_state_key) {67		return remembered68	}69	name := os.base(store.root)70	existing := backend.get(head_key(name))!71	if existing.found {72		theirs := existing.body.bytestr().trim_space()73		if history.by_id(store, theirs) == none {74			return error('the remote already has "${name}" with a history this repo does not share — rename the folder, or use another remote')75		}76	}77	store.set_state(repo_state_key, name)!78	return name79}8081// offer sends what the remote lacks and then swaps `head`. Nothing is ever82// queued: the next offer simply sends everything the remote does not have.83pub fn offer(store &storage.Store, mut backend remote.Backend) !OfferReport {84	name := repo_name(store, mut backend)!85	local := history.head(store) or { return error(no_saves) }86	mut report := OfferReport{87		repo: name88		head: local.id89	}90	if check_format(mut backend, name)! == '' {91		// Offered before this key existed: record the layout now, without92		// overwriting one another machine wrote first.93		backend.put_if_absent(format_key(name), format_value.bytes())!94	}95	remote_head := backend.get(head_key(name))!96	// Records offered by a machine that has not applied a strip would put the97	// stripped path back on the remote, where nothing would rewrite it again.98	// ponytail: an offer racing a strip on another machine inside the same few99	// seconds can still slip through; closing that needs a remote lock.100	if unapplied_strip(store, mut backend, name)! {101		return error('the remote has a strip this repo has not applied — run koh steal')102	}103	if remote_head.found {104		theirs := remote_head.body.bytestr().trim_space()105		newer := saves_after(store, local.id, theirs) or {106			return error('remote has new saves — run koh steal')107		}108		for save in newer {109			upload_save(store, mut backend, name, save, mut report)!110		}111	} else {112		oldest_first := history.line(store)!113		for index := oldest_first.len - 1; index >= 0; index-- {114			upload_save(store, mut backend, name, oldest_first[index], mut report)!115		}116	}117	upload_events(store, mut backend, name, mut report)!118	if !backend.put_if_match(head_key(name), local.id.bytes(), remote_head.tag)! {119		return error('remote has new saves — run koh steal')120	}121	store.set_state(remote_head_key, local.id)!122	return report123}124125// saves_after lists the local saves newer than `theirs`, oldest first. It fails126// when `theirs` is not on this line, which is the honest answer: we cannot127// offer on top of a save we do not have.128fn saves_after(store &storage.Store, ours string, theirs string) ![]history.Save {129	mut newer := []history.Save{}130	mut cursor := ours131	for cursor != '' && cursor != theirs {132		save := history.by_id(store, cursor) or { return error('save ${cursor} is missing') }133		newer << save134		cursor = save.parent135	}136	if cursor != theirs {137		return error('${theirs} is not on this line')138	}139	newer.reverse_in_place()140	return newer141}142143fn upload_save(store &storage.Store, mut backend remote.Backend, repo string, save history.Save, mut report OfferReport) ! {144	files := history.files_of(store, save.id)!145	// Saves go up oldest first and head moves only once all of them landed, so146	// the parent and its objects are on the remote already: only content new147	// in this save travels, not a round trip per file in the repo.148	mut sent := map[string]bool{}149	if save.parent != '' {150		for file in history.files_of(store, save.parent)! {151			sent[file.hash] = true152		}153	}154	for file in files {155		if file.hash in sent {156			continue157		}158		sent[file.hash] = true159		if backend.put_if_absent(object_key(repo, file.hash), store.object_blob(file.hash)!)! {160			report.objects++161		}162	}163	record := save_record(store, save)!164	if backend.put_if_absent(save_key(repo, save.id), encode_save(record).bytes())! {165		report.saves++166	}167}168169// upload_events sends every event the remote does not have. Events are tiny, so170// listing them costs less than tracking which ones were sent.171fn upload_events(store &storage.Store, mut backend remote.Backend, repo string, mut report OfferReport) ! {172	already := backend.list(events_prefix(repo))!173	mut already_sent := map[string]bool{}174	for key in already {175		already_sent[os.base(key)] = true176	}177	for event in events.all(store)! {178		if event.id in already_sent {179			continue180		}181		if backend.put_if_absent(event_key(repo, event.id), encode_event(event).bytes())! {182			report.events++183		}184	}185}186187// fetched holds everything a steal read from the remote, before any of it is188// written locally. Network work happens first; the database gets one189// transaction, so a failure leaves nothing half-applied.190pub struct Fetched {191pub mut:192	repo_objects map[string][]u8193	saves        []SaveRecord // oldest first194	events_out   []events.Event195	head         string196	base         string // the newest save both sides already shared, '' when none197}198199pub fn fetch_back_to_known(store &storage.Store, mut backend remote.Backend, repo string) !Fetched {200	check_format(mut backend, repo)!201	remote_head := backend.get(head_key(repo))!202	if !remote_head.found {203		return error('the remote has no "${repo}" yet — run koh offer here first')204	}205	mut fetched := Fetched{206		head: remote_head.body.bytestr().trim_space()207	}208	mut cursor := fetched.head209	for cursor != '' && history.by_id(store, cursor) == none {210		record_object := backend.get(save_key(repo, cursor))!211		if !record_object.found {212			return error('the remote is missing save ${cursor} — run koh offer there')213		}214		record := decode_save(record_object.body)!215		for file in record.files {216			if !store.has_object(file.hash)! && file.hash !in fetched.repo_objects {217				blob := backend.get(object_key(repo, file.hash))!218				if !blob.found {219					return error('the remote is missing object ${file.hash} that save ${record.id} needs')220				}221				fetched.repo_objects[file.hash] = blob.body222			}223		}224		fetched.saves << record225		cursor = record.parent226	}227	fetched.saves.reverse_in_place()228	fetched.base = cursor229	fetched.events_out = fetch_events(store, mut backend, repo)!230	return fetched231}232233fn fetch_events(store &storage.Store, mut backend remote.Backend, repo string) ![]events.Event {234	have := events.all(store)!235	mut known := map[string]bool{}236	for event in have {237		known[event.id] = true238	}239	mut incoming := []events.Event{}240	for key in backend.list(events_prefix(repo))! {241		id := os.base(key)242		if id in known {243			continue244		}245		body := backend.get(event_key(repo, id))!246		if !body.found {247			continue248		}249		incoming << decode_event(body.body)!250	}251	return incoming252}253254// write_fetched lands a fetch in one transaction: objects first (verified),255// then the events, then the saves they belong to. A new strip reaches every256// save already here, but an arriving save only when `reaches` says so: one257// made after the strip is new work, not what the strip removed.258pub fn write_fetched(store &storage.Store, fetched Fetched) ! {259	for hash, blob in fetched.repo_objects {260		store.put_object_blob(hash, blob)!261	}262	mut strips := []events.Event{}263	for event in fetched.events_out {264		if !events.receive(store, event)! {265			continue266		}267		if event.kind == events.kind_strip {268			// Somebody stripped a path: remove it here too, so the two269			// histories agree from now on. Objects go once the saves are in.270			remove_path(store, event.path, event.save_id)!271			strips << event272		}273	}274	for record in fetched.saves {275		files := to_files(record).filter(fn [strips, record] (file snapshot.FileEntry) bool {276			return !strips.any(reaches(it, record.id, record.created_at)277				&& covers(it.path, file.path))278		})279		history.import_save(store, to_save(record), files)!280	}281	if strips.len > 0 {282		store.drop_unreferenced_objects()!283	}284}285286// reaches reports whether a strip event applies to a save arriving from the287// remote. A strip of every save covers the saves made up to its own time.288fn reaches(strip events.Event, save_id string, created_at i64) bool {289	if strip.save_id != '' {290		return strip.save_id == save_id291	}292	return created_at <= strip.created_at293}