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// 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 clone 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, remembered in the repo so74// renaming the folder later changes nothing.75pub fn repo_name(store &storage.Store, mut backend remote.Backend) !string {76	if remembered := store.state(repo_state_key) {77		return remembered78	}79	name := os.base(store.root)80	// Before the clash check below: a remote this koh cannot read is a "update81	// koh" answer, not a "rename the folder" one.82	check_format(mut backend, name)!83	existing := backend.get(head_key(name))!84	if existing.found {85		theirs := existing.body.bytestr().trim_space()86		if history.by_id(store, theirs) == none {87			return error('the remote already has "${name}" with a history this repo does not share — rename the folder, or use another remote')88		}89	}90	store.set_state(repo_state_key, name)!91	return name92}9394// offer sends what the remote lacks and then swaps `head`. Nothing is ever95// queued: the next offer simply sends everything the remote does not have.96pub fn offer(store &storage.Store, mut backend remote.Backend) !OfferReport {97	name := repo_name(store, mut backend)!98	local := history.head(store) or { return error(no_saves) }99	mut report := OfferReport{100		repo: name101		head: local.id102	}103	if check_format(mut backend, name)! == '' {104		// Offered before this key existed: record the layout now.105		put_format(mut backend, name)!106	}107	remote_head := backend.get(head_key(name))!108	// Records offered by a machine that has not applied a strip would put the109	// stripped path back on the remote, where nothing would rewrite it again.110	// ponytail: an offer racing a strip on another machine inside the same few111	// seconds can still slip through; closing that needs a remote lock.112	if unapplied_strip(store, mut backend, name)! {113		return error('the remote has a strip this repo has not applied — run koh steal')114	}115	if remote_head.found {116		theirs := remote_head.body.bytestr().trim_space()117		newer := saves_after(store, local.id, theirs) or {118			return error('remote has new saves — run koh steal')119		}120		for save in newer {121			upload_save(store, mut backend, name, save, mut report)!122		}123	} else {124		oldest_first := history.line(store)!125		for index := oldest_first.len - 1; index >= 0; index-- {126			upload_save(store, mut backend, name, oldest_first[index], mut report)!127		}128	}129	upload_events(store, mut backend, name, mut report)!130	if !backend.put_if_match(head_key(name), local.id.bytes(), remote_head.tag)! {131		return error('remote has new saves — run koh steal')132	}133	store.set_state(remote_head_key, local.id)!134	return report135}136137// saves_after lists the local saves newer than `theirs`, oldest first. It fails138// when `theirs` is not on this line, which is the honest answer: we cannot139// offer on top of a save we do not have.140fn saves_after(store &storage.Store, ours string, theirs string) ![]history.Save {141	mut newer := []history.Save{}142	mut cursor := ours143	for cursor != '' && cursor != theirs {144		save := history.by_id(store, cursor) or { return error('save ${cursor} is missing') }145		newer << save146		cursor = save.parent147	}148	if cursor != theirs {149		return error('${theirs} is not on this line')150	}151	newer.reverse_in_place()152	return newer153}154155fn upload_save(store &storage.Store, mut backend remote.Backend, repo string, save history.Save, mut report OfferReport) ! {156	files := history.files_of(store, save.id)!157	// Saves go up oldest first and head moves only once all of them landed, so158	// the parent and its objects are on the remote already: only content new159	// in this save travels, not a round trip per file in the repo.160	mut sent := map[string]bool{}161	if save.parent != '' {162		for file in history.files_of(store, save.parent)! {163			sent[file.hash] = true164		}165	}166	for file in files {167		if file.hash in sent {168			continue169		}170		sent[file.hash] = true171		if backend.put_if_absent(object_key(repo, file.hash), store.object_blob(file.hash)!)! {172			report.objects++173		}174	}175	record := save_record(store, save)!176	if backend.put_if_absent(save_key(repo, save.id), encode_save(record).bytes())! {177		report.saves++178	}179}180181// upload_events sends every event the remote does not have. Events are tiny, so182// listing them costs less than tracking which ones were sent.183fn upload_events(store &storage.Store, mut backend remote.Backend, repo string, mut report OfferReport) ! {184	already := backend.list(events_prefix(repo))!185	mut already_sent := map[string]bool{}186	for key in already {187		already_sent[os.base(key)] = true188	}189	for event in events.all(store)! {190		if event.id in already_sent {191			continue192		}193		if backend.put_if_absent(event_key(repo, event.id), encode_event(event).bytes())! {194			report.events++195		}196	}197}198199// fetched holds everything a steal read from the remote, before any of it is200// written locally. Network work happens first; the database gets one201// transaction, so a failure leaves nothing half-applied.202pub struct Fetched {203pub mut:204	repo_objects map[string][]u8205	saves        []SaveRecord // oldest first206	events_out   []events.Event207	head         string208	base         string // the newest save both sides already shared, '' when none209}210211pub fn fetch_back_to_known(store &storage.Store, mut backend remote.Backend, repo string) !Fetched {212	check_format(mut backend, repo)!213	remote_head := backend.get(head_key(repo))!214	if !remote_head.found {215		return error('the remote has no "${repo}" yet — run koh offer here first')216	}217	mut fetched := Fetched{218		head: remote_head.body.bytestr().trim_space()219	}220	mut on_line := map[string]bool{}221	for save in history.line(store)! {222		on_line[save.id] = true223	}224	mut cursor := fetched.head225	for cursor != '' && cursor !in on_line {226		record := if saved := history.by_id(store, cursor) {227			save_record(store, saved)!228		} else {229			record_object := backend.get(save_key(repo, cursor))!230			if !record_object.found {231				return error('the remote is missing save ${cursor} — run koh offer there')232			}233			decode_save(record_object.body)!234		}235		for file in record.files {236			if !store.has_object(file.hash)! && file.hash !in fetched.repo_objects {237				blob := backend.get(object_key(repo, file.hash))!238				if !blob.found {239					return error('the remote is missing object ${file.hash} that save ${record.id} needs')240				}241				fetched.repo_objects[file.hash] = blob.body242			}243		}244		fetched.saves << record245		cursor = record.parent246	}247	fetched.saves.reverse_in_place()248	fetched.base = cursor249	fetched.events_out = fetch_events(store, mut backend, repo)!250	return fetched251}252253fn fetch_events(store &storage.Store, mut backend remote.Backend, repo string) ![]events.Event {254	have := events.all(store)!255	mut known := map[string]bool{}256	for event in have {257		known[event.id] = true258	}259	mut incoming := []events.Event{}260	for key in backend.list(events_prefix(repo))! {261		id := os.base(key)262		if id in known {263			continue264		}265		body := backend.get(event_key(repo, id))!266		if !body.found {267			continue268		}269		incoming << decode_event(body.body)!270	}271	return incoming272}273274// write_fetched lands a fetch in one transaction: objects first (verified),275// then the events, then the saves they belong to. A strip bans its path from276// all of history, so it reaches every save, here or arriving, whenever made.277pub fn write_fetched(store &storage.Store, fetched Fetched) ! {278	for hash, blob in fetched.repo_objects {279		store.put_object_blob(hash, blob)!280	}281	mut strips := []events.Event{}282	for event in fetched.events_out {283		if !events.receive(store, event)! {284			continue285		}286		if event.kind == events.kind_strip {287			remove_path(store, event.path, event.save_id)!288			strips << event289		}290	}291	for record in fetched.saves {292		import_arriving(store, record, strips)!293	}294	if strips.len > 0 {295		store.drop_unreferenced_objects()!296	}297}298299// write_fetched_records makes network work durable without changing the line.300// Events wait for the combine transaction because a strip can rewrite history.301pub fn write_fetched_records(store &storage.Store, fetched Fetched) ! {302	for hash, blob in fetched.repo_objects {303		store.put_object_blob(hash, blob)!304	}305	strips := fetched.events_out.filter(it.kind == events.kind_strip)306	for record in fetched.saves {307		import_arriving(store, record, strips)!308	}309}310311// import_arriving lands one save from the remote, unless it is already here, less every path a312// strip that `reaches` it removed.313fn import_arriving(store &storage.Store, record SaveRecord, strips []events.Event) ! {314	if history.by_id(store, record.id) != none {315		return316	}317	files := to_files(record).filter(fn [strips, record] (file snapshot.FileEntry) bool {318		return !strips.any(reaches(it, record.id) && covers(it.path, file.path))319	})320	history.import_save(store, to_save(record), files)!321}322323// reaches reports whether a strip event applies to a save: its one save, or324// every save there is, whenever it was made.325fn reaches(strip events.Event, save_id string) bool {326	return strip.save_id == '' || strip.save_id == save_id327}