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 and the last offered head are remembered.20const repo_state_key = 'remote_repo'21const offered_head_key = 'remote_head'2223pub const no_saves = 'this repo has no saves yet — nothing to offer'2425pub struct OfferReport {26pub mut:27	repo    string28	saves   int29	objects int30	events  int31	head    string32}3334// repo_name is the folder's name at first offer, remembered in the repo so35// renaming the folder later changes nothing.36pub fn repo_name(store &storage.Store, mut backend remote.Backend) !string {37	if remembered := store.state(repo_state_key) {38		return remembered39	}40	name := os.base(store.root)41	existing := backend.get(head_key(name))!42	if existing.found {43		theirs := existing.body.bytestr().trim_space()44		if history.by_id(store, theirs) == none {45			return error('the remote already has "${name}" with a history this repo does not share — rename the folder, or use another remote')46		}47	}48	store.set_state(repo_state_key, name)!49	return name50}5152// offer sends what the remote lacks and then swaps `head`. Nothing is ever53// queued: the next offer simply sends everything the remote does not have.54pub fn offer(store &storage.Store, mut backend remote.Backend) !OfferReport {55	name := repo_name(store, mut backend)!56	local := history.head(store) or { return error(no_saves) }57	mut report := OfferReport{58		repo: name59		head: local.id60	}61	remote_head := backend.get(head_key(name))!62	// Records offered by a machine that has not applied a strip would put the63	// stripped path back on the remote, where nothing would rewrite it again.64	// ponytail: an offer racing a strip on another machine inside the same few65	// seconds can still slip through; closing that needs a remote lock.66	if unapplied_strip(store, mut backend, name)! {67		return error('the remote has a strip this repo has not applied — run koh steal')68	}69	if remote_head.found {70		theirs := remote_head.body.bytestr().trim_space()71		newer := saves_after(store, local.id, theirs) or {72			return error('remote has new saves — run koh steal')73		}74		for save in newer {75			upload_save(store, mut backend, name, save, mut report)!76		}77	} else {78		oldest_first := history.line(store)!79		for index := oldest_first.len - 1; index >= 0; index-- {80			upload_save(store, mut backend, name, oldest_first[index], mut report)!81		}82	}83	upload_events(store, mut backend, name, mut report)!84	if !backend.put_if_match(head_key(name), local.id.bytes(), remote_head.tag)! {85		return error('remote has new saves — run koh steal')86	}87	store.set_state(offered_head_key, local.id)!88	return report89}9091// saves_after lists the local saves newer than `theirs`, oldest first. It fails92// when `theirs` is not on this line, which is the honest answer: we cannot93// offer on top of a save we do not have.94fn saves_after(store &storage.Store, ours string, theirs string) ![]history.Save {95	mut newer := []history.Save{}96	mut cursor := ours97	for cursor != '' && cursor != theirs {98		save := history.by_id(store, cursor) or { return error('save ${cursor} is missing') }99		newer << save100		cursor = save.parent101	}102	if cursor != theirs {103		return error('${theirs} is not on this line')104	}105	newer.reverse_in_place()106	return newer107}108109fn upload_save(store &storage.Store, mut backend remote.Backend, repo string, save history.Save, mut report OfferReport) ! {110	files := history.files_of(store, save.id)!111	for file in files {112		if backend.put_if_absent(object_key(repo, file.hash), store.object_blob(file.hash)!)! {113			report.objects++114		}115	}116	record := save_record(store, save)!117	if backend.put_if_absent(save_key(repo, save.id), encode_save(record).bytes())! {118		report.saves++119	}120}121122// upload_events sends every event the remote does not have. Events are tiny, so123// listing them costs less than tracking which ones were sent.124fn upload_events(store &storage.Store, mut backend remote.Backend, repo string, mut report OfferReport) ! {125	already := backend.list(events_prefix(repo))!126	mut already_sent := map[string]bool{}127	for key in already {128		already_sent[os.base(key)] = true129	}130	for event in events.all(store)! {131		if event.id in already_sent {132			continue133		}134		if backend.put_if_absent(event_key(repo, event.id), encode_event(event).bytes())! {135			report.events++136		}137	}138}139140// fetched holds everything a steal read from the remote, before any of it is141// written locally. Network work happens first; the database gets one142// transaction, so a failure leaves nothing half-applied.143pub struct Fetched {144pub mut:145	repo_objects    map[string][]u8146	saves           []SaveRecord // oldest first147	events_out      []events.Event148	head            string149	base            string // the newest save both sides already shared, '' when none150	remote_save_ids map[string]bool151}152153pub fn fetch_back_to_known(store &storage.Store, mut backend remote.Backend, repo string) !Fetched {154	remote_head := backend.get(head_key(repo))!155	if !remote_head.found {156		return error('the remote has no "${repo}" yet — run koh offer here first')157	}158	mut fetched := Fetched{159		head:            remote_head.body.bytestr().trim_space()160		remote_save_ids: map[string]bool{}161	}162	mut cursor := fetched.head163	for cursor != '' && history.by_id(store, cursor) == none {164		record_object := backend.get(save_key(repo, cursor))!165		if !record_object.found {166			return error('the remote is missing save ${cursor} — run koh offer there')167		}168		record := decode_save(record_object.body)!169		for file in record.files {170			if !store.has_object(file.hash)! && file.hash !in fetched.repo_objects {171				blob := backend.get(object_key(repo, file.hash))!172				if !blob.found {173					return error('the remote is missing object ${file.hash} that save ${record.id} needs')174				}175				fetched.repo_objects[file.hash] = blob.body176			}177		}178		fetched.remote_save_ids[record.id] = true179		fetched.saves << record180		cursor = record.parent181	}182	fetched.saves.reverse_in_place()183	fetched.base = cursor184	fetched.events_out = fetch_events(store, mut backend, repo)!185	return fetched186}187188fn fetch_events(store &storage.Store, mut backend remote.Backend, repo string) ![]events.Event {189	have := events.all(store)!190	mut known := map[string]bool{}191	for event in have {192		known[event.id] = true193	}194	mut incoming := []events.Event{}195	for key in backend.list(events_prefix(repo))! {196		id := os.base(key)197		if id in known {198			continue199		}200		body := backend.get(event_key(repo, id))!201		if !body.found {202			continue203		}204		incoming << decode_event(body.body)!205	}206	return incoming207}208209// write_fetched lands a fetch in one transaction: objects first (verified), then210// the saves they belong to, then the events.211pub fn write_fetched(store &storage.Store, fetched Fetched) ! {212	for hash, blob in fetched.repo_objects {213		store.put_object_blob(hash, blob)!214	}215	for record in fetched.saves {216		history.import_save(store, to_save(record), to_files(record))!217	}218	for event in fetched.events_out {219		if !events.receive(store, event)! {220			continue221		}222		if event.kind == events.kind_strip {223			// Somebody stripped a path: remove it here too, so the two224			// histories agree from now on.225			apply_strip(store, event)!226		}227	}228}229230// can_fast_forward reports whether our line sits behind the remote's, which is231// the case when every local save is an ancestor of the remote head.232pub fn can_fast_forward(store &storage.Store, remote_head string, remote_ids map[string]bool) bool {233	mut cursor := remote_head234	for cursor != '' {235		if history.by_id(store, cursor) == none {236			return false237		}238		save := history.by_id(store, cursor) or { return false }239		if save.parent == '' {240			// Reached the first save without leaving the line we have.241			return true242		}243		cursor = save.parent244	}245	return false246}