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.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'2829pub struct OfferReport {30pub mut:31	repo    string32	saves   int33	objects int34	events  int35	head    string36}3738// remember_repo binds a repo to its name on the remote. A clone does it at39// once: its folder may be renamed before it ever offers.40pub fn remember_repo(store &storage.Store, name string) ! {41	store.set_state(repo_state_key, name)!42}4344// repo_name is the folder's name at first offer, remembered in the repo so45// renaming the folder later changes nothing.46pub fn repo_name(store &storage.Store, mut backend remote.Backend) !string {47	if remembered := store.state(repo_state_key) {48		return remembered49	}50	name := os.base(store.root)51	existing := backend.get(head_key(name))!52	if existing.found {53		theirs := existing.body.bytestr().trim_space()54		if history.by_id(store, theirs) == none {55			return error('the remote already has "${name}" with a history this repo does not share — rename the folder, or use another remote')56		}57	}58	store.set_state(repo_state_key, name)!59	return name60}6162// offer sends what the remote lacks and then swaps `head`. Nothing is ever63// queued: the next offer simply sends everything the remote does not have.64pub fn offer(store &storage.Store, mut backend remote.Backend) !OfferReport {65	name := repo_name(store, mut backend)!66	local := history.head(store) or { return error(no_saves) }67	mut report := OfferReport{68		repo: name69		head: local.id70	}71	remote_head := backend.get(head_key(name))!72	// Records offered by a machine that has not applied a strip would put the73	// stripped path back on the remote, where nothing would rewrite it again.74	// ponytail: an offer racing a strip on another machine inside the same few75	// seconds can still slip through; closing that needs a remote lock.76	if unapplied_strip(store, mut backend, name)! {77		return error('the remote has a strip this repo has not applied — run koh steal')78	}79	if remote_head.found {80		theirs := remote_head.body.bytestr().trim_space()81		newer := saves_after(store, local.id, theirs) or {82			return error('remote has new saves — run koh steal')83		}84		for save in newer {85			upload_save(store, mut backend, name, save, mut report)!86		}87	} else {88		oldest_first := history.line(store)!89		for index := oldest_first.len - 1; index >= 0; index-- {90			upload_save(store, mut backend, name, oldest_first[index], mut report)!91		}92	}93	upload_events(store, mut backend, name, mut report)!94	if !backend.put_if_match(head_key(name), local.id.bytes(), remote_head.tag)! {95		return error('remote has new saves — run koh steal')96	}97	store.set_state(remote_head_key, local.id)!98	return report99}100101// saves_after lists the local saves newer than `theirs`, oldest first. It fails102// when `theirs` is not on this line, which is the honest answer: we cannot103// offer on top of a save we do not have.104fn saves_after(store &storage.Store, ours string, theirs string) ![]history.Save {105	mut newer := []history.Save{}106	mut cursor := ours107	for cursor != '' && cursor != theirs {108		save := history.by_id(store, cursor) or { return error('save ${cursor} is missing') }109		newer << save110		cursor = save.parent111	}112	if cursor != theirs {113		return error('${theirs} is not on this line')114	}115	newer.reverse_in_place()116	return newer117}118119fn upload_save(store &storage.Store, mut backend remote.Backend, repo string, save history.Save, mut report OfferReport) ! {120	files := history.files_of(store, save.id)!121	// Saves go up oldest first and head moves only once all of them landed, so122	// the parent and its objects are on the remote already: only content new123	// in this save travels, not a round trip per file in the repo.124	mut sent := map[string]bool{}125	if save.parent != '' {126		for file in history.files_of(store, save.parent)! {127			sent[file.hash] = true128		}129	}130	for file in files {131		if file.hash in sent {132			continue133		}134		sent[file.hash] = true135		if backend.put_if_absent(object_key(repo, file.hash), store.object_blob(file.hash)!)! {136			report.objects++137		}138	}139	record := save_record(store, save)!140	if backend.put_if_absent(save_key(repo, save.id), encode_save(record).bytes())! {141		report.saves++142	}143}144145// upload_events sends every event the remote does not have. Events are tiny, so146// listing them costs less than tracking which ones were sent.147fn upload_events(store &storage.Store, mut backend remote.Backend, repo string, mut report OfferReport) ! {148	already := backend.list(events_prefix(repo))!149	mut already_sent := map[string]bool{}150	for key in already {151		already_sent[os.base(key)] = true152	}153	for event in events.all(store)! {154		if event.id in already_sent {155			continue156		}157		if backend.put_if_absent(event_key(repo, event.id), encode_event(event).bytes())! {158			report.events++159		}160	}161}162163// fetched holds everything a steal read from the remote, before any of it is164// written locally. Network work happens first; the database gets one165// transaction, so a failure leaves nothing half-applied.166pub struct Fetched {167pub mut:168	repo_objects map[string][]u8169	saves        []SaveRecord // oldest first170	events_out   []events.Event171	head         string172	base         string // the newest save both sides already shared, '' when none173}174175pub fn fetch_back_to_known(store &storage.Store, mut backend remote.Backend, repo string) !Fetched {176	remote_head := backend.get(head_key(repo))!177	if !remote_head.found {178		return error('the remote has no "${repo}" yet — run koh offer here first')179	}180	mut fetched := Fetched{181		head: remote_head.body.bytestr().trim_space()182	}183	mut cursor := fetched.head184	for cursor != '' && history.by_id(store, cursor) == none {185		record_object := backend.get(save_key(repo, cursor))!186		if !record_object.found {187			return error('the remote is missing save ${cursor} — run koh offer there')188		}189		record := decode_save(record_object.body)!190		for file in record.files {191			if !store.has_object(file.hash)! && file.hash !in fetched.repo_objects {192				blob := backend.get(object_key(repo, file.hash))!193				if !blob.found {194					return error('the remote is missing object ${file.hash} that save ${record.id} needs')195				}196				fetched.repo_objects[file.hash] = blob.body197			}198		}199		fetched.saves << record200		cursor = record.parent201	}202	fetched.saves.reverse_in_place()203	fetched.base = cursor204	fetched.events_out = fetch_events(store, mut backend, repo)!205	return fetched206}207208fn fetch_events(store &storage.Store, mut backend remote.Backend, repo string) ![]events.Event {209	have := events.all(store)!210	mut known := map[string]bool{}211	for event in have {212		known[event.id] = true213	}214	mut incoming := []events.Event{}215	for key in backend.list(events_prefix(repo))! {216		id := os.base(key)217		if id in known {218			continue219		}220		body := backend.get(event_key(repo, id))!221		if !body.found {222			continue223		}224		incoming << decode_event(body.body)!225	}226	return incoming227}228229// write_fetched lands a fetch in one transaction: objects first (verified),230// then the events, then the saves they belong to. A new strip reaches every231// save already here, but an arriving save only when `reaches` says so: one232// made after the strip is new work, not what the strip removed.233pub fn write_fetched(store &storage.Store, fetched Fetched) ! {234	for hash, blob in fetched.repo_objects {235		store.put_object_blob(hash, blob)!236	}237	mut strips := []events.Event{}238	for event in fetched.events_out {239		if !events.receive(store, event)! {240			continue241		}242		if event.kind == events.kind_strip {243			// Somebody stripped a path: remove it here too, so the two244			// histories agree from now on. Objects go once the saves are in.245			remove_path(store, event.path, event.save_id)!246			strips << event247		}248	}249	for record in fetched.saves {250		files := to_files(record).filter(fn [strips, record] (file snapshot.FileEntry) bool {251			return !strips.any(reaches(it, record.id, record.created_at)252				&& covers(it.path, file.path))253		})254		history.import_save(store, to_save(record), files)!255	}256	if strips.len > 0 {257		store.drop_unreferenced_objects()!258	}259}260261// reaches reports whether a strip event applies to a save arriving from the262// remote. A strip of every save covers the saves made up to its own time.263fn reaches(strip events.Event, save_id string, created_at i64) bool {264	if strip.save_id != '' {265		return strip.save_id == save_id266	}267	return created_at <= strip.created_at268}