sync.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 sync1213import os14import src.events15import src.history16import src.remote17import src.storage1819// Where the repo's name on the remote is remembered.20const repo_state_key = 'remote_repo'2122pub const no_saves = 'this repo has no saves yet — nothing to offer'2324pub struct OfferReport {25pub mut:26	repo    string27	saves   int28	objects int29	events  int30	head    string31}3233// remember_repo binds a repo to its name on the remote. A clone does it at34// once: its folder may be renamed before it ever offers.35pub fn remember_repo(store &storage.Store, name string) ! {36	store.set_state(repo_state_key, name)!37}3839// repo_name is the folder's name at first offer, remembered in the repo so40// renaming the folder later changes nothing.41pub fn repo_name(store &storage.Store, mut backend remote.Backend) !string {42	if remembered := store.state(repo_state_key) {43		return remembered44	}45	name := os.base(store.root)46	existing := backend.get(head_key(name))!47	if existing.found {48		theirs := existing.body.bytestr().trim_space()49		if history.by_id(store, theirs) == none {50			return error('the remote already has "${name}" with a history this repo does not share — rename the folder, or use another remote')51		}52	}53	store.set_state(repo_state_key, name)!54	return name55}5657// offer sends what the remote lacks and then swaps `head`. Nothing is ever58// queued: the next offer simply sends everything the remote does not have.59pub fn offer(store &storage.Store, mut backend remote.Backend) !OfferReport {60	name := repo_name(store, mut backend)!61	local := history.head(store) or { return error(no_saves) }62	mut report := OfferReport{63		repo: name64		head: local.id65	}66	remote_head := backend.get(head_key(name))!67	// Records offered by a machine that has not applied a strip would put the68	// stripped path back on the remote, where nothing would rewrite it again.69	// ponytail: an offer racing a strip on another machine inside the same few70	// seconds can still slip through; closing that needs a remote lock.71	if unapplied_strip(store, mut backend, name)! {72		return error('the remote has a strip this repo has not applied — run koh steal')73	}74	if remote_head.found {75		theirs := remote_head.body.bytestr().trim_space()76		newer := saves_after(store, local.id, theirs) or {77			return error('remote has new saves — run koh steal')78		}79		for save in newer {80			upload_save(store, mut backend, name, save, mut report)!81		}82	} else {83		oldest_first := history.line(store)!84		for index := oldest_first.len - 1; index >= 0; index-- {85			upload_save(store, mut backend, name, oldest_first[index], mut report)!86		}87	}88	upload_events(store, mut backend, name, mut report)!89	if !backend.put_if_match(head_key(name), local.id.bytes(), remote_head.tag)! {90		return error('remote has new saves — run koh steal')91	}92	return report93}9495// saves_after lists the local saves newer than `theirs`, oldest first. It fails96// when `theirs` is not on this line, which is the honest answer: we cannot97// offer on top of a save we do not have.98fn saves_after(store &storage.Store, ours string, theirs string) ![]history.Save {99	mut newer := []history.Save{}100	mut cursor := ours101	for cursor != '' && cursor != theirs {102		save := history.by_id(store, cursor) or { return error('save ${cursor} is missing') }103		newer << save104		cursor = save.parent105	}106	if cursor != theirs {107		return error('${theirs} is not on this line')108	}109	newer.reverse_in_place()110	return newer111}112113fn upload_save(store &storage.Store, mut backend remote.Backend, repo string, save history.Save, mut report OfferReport) ! {114	files := history.files_of(store, save.id)!115	for file in files {116		if backend.put_if_absent(object_key(repo, file.hash), store.object_blob(file.hash)!)! {117			report.objects++118		}119	}120	record := save_record(store, save)!121	if backend.put_if_absent(save_key(repo, save.id), encode_save(record).bytes())! {122		report.saves++123	}124}125126// upload_events sends every event the remote does not have. Events are tiny, so127// listing them costs less than tracking which ones were sent.128fn upload_events(store &storage.Store, mut backend remote.Backend, repo string, mut report OfferReport) ! {129	already := backend.list(events_prefix(repo))!130	mut already_sent := map[string]bool{}131	for key in already {132		already_sent[os.base(key)] = true133	}134	for event in events.all(store)! {135		if event.id in already_sent {136			continue137		}138		if backend.put_if_absent(event_key(repo, event.id), encode_event(event).bytes())! {139			report.events++140		}141	}142}143144// fetched holds everything a steal read from the remote, before any of it is145// written locally. Network work happens first; the database gets one146// transaction, so a failure leaves nothing half-applied.147pub struct Fetched {148pub mut:149	repo_objects map[string][]u8150	saves        []SaveRecord // oldest first151	events_out   []events.Event152	head         string153	base         string // the newest save both sides already shared, '' when none154}155156pub fn fetch_back_to_known(store &storage.Store, mut backend remote.Backend, repo string) !Fetched {157	remote_head := backend.get(head_key(repo))!158	if !remote_head.found {159		return error('the remote has no "${repo}" yet — run koh offer here first')160	}161	mut fetched := Fetched{162		head: remote_head.body.bytestr().trim_space()163	}164	mut cursor := fetched.head165	for cursor != '' && history.by_id(store, cursor) == none {166		record_object := backend.get(save_key(repo, cursor))!167		if !record_object.found {168			return error('the remote is missing save ${cursor} — run koh offer there')169		}170		record := decode_save(record_object.body)!171		for file in record.files {172			if !store.has_object(file.hash)! && file.hash !in fetched.repo_objects {173				blob := backend.get(object_key(repo, file.hash))!174				if !blob.found {175					return error('the remote is missing object ${file.hash} that save ${record.id} needs')176				}177				fetched.repo_objects[file.hash] = blob.body178			}179		}180		fetched.saves << record181		cursor = record.parent182	}183	fetched.saves.reverse_in_place()184	fetched.base = cursor185	fetched.events_out = fetch_events(store, mut backend, repo)!186	return fetched187}188189fn fetch_events(store &storage.Store, mut backend remote.Backend, repo string) ![]events.Event {190	have := events.all(store)!191	mut known := map[string]bool{}192	for event in have {193		known[event.id] = true194	}195	mut incoming := []events.Event{}196	for key in backend.list(events_prefix(repo))! {197		id := os.base(key)198		if id in known {199			continue200		}201		body := backend.get(event_key(repo, id))!202		if !body.found {203			continue204		}205		incoming << decode_event(body.body)!206	}207	return incoming208}209210// write_fetched lands a fetch in one transaction: objects first (verified), then211// the saves they belong to, then the events.212pub fn write_fetched(store &storage.Store, fetched Fetched) ! {213	for hash, blob in fetched.repo_objects {214		store.put_object_blob(hash, blob)!215	}216	for record in fetched.saves {217		history.import_save(store, to_save(record), to_files(record))!218	}219	for event in fetched.events_out {220		if !events.receive(store, event)! {221			continue222		}223		if event.kind == events.kind_strip {224			// Somebody stripped a path: remove it here too, so the two225			// histories agree from now on.226			apply_strip(store, event)!227		}228	}229}