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 and the last offered head are remembered.21const repo_state_key = 'remote_repo'22const offered_head_key = 'remote_head'2324pub const no_saves = 'this repo has no saves yet — nothing to offer'2526pub struct OfferReport {27pub mut:28	repo    string29	saves   int30	objects int31	events  int32	head    string33}3435// repo_name is the folder's name at first offer, remembered in the repo so36// renaming the folder later changes nothing.37pub fn repo_name(store &storage.Store, mut backend remote.Backend) !string {38	if remembered := store.state(repo_state_key) {39		return remembered40	}41	name := os.base(store.root)42	existing := backend.get(head_key(name))!43	if existing.found {44		theirs := existing.body.bytestr().trim_space()45		if history.by_id(store, theirs) == none {46			return error('the remote already has "${name}" with a history this repo does not share — rename the folder, or use another remote')47		}48	}49	store.set_state(repo_state_key, name)!50	return name51}5253// offer sends what the remote lacks and then swaps `head`. Nothing is ever54// queued: the next offer simply sends everything the remote does not have.55pub fn offer(store &storage.Store, mut backend remote.Backend) !OfferReport {56	name := repo_name(store, mut backend)!57	local := history.head(store) or { return error(no_saves) }58	mut report := OfferReport{59		repo: name60		head: local.id61	}62	remote_head := backend.get(head_key(name))!63	if remote_head.found {64		theirs := remote_head.body.bytestr().trim_space()65		newer := saves_after(store, local.id, theirs) or {66			return error('remote has new saves — run koh steal')67		}68		for save in newer {69			upload_save(store, mut backend, name, save, mut report)!70		}71	} else {72		oldest_first := history.line(store)!73		for index := oldest_first.len - 1; index >= 0; index-- {74			upload_save(store, mut backend, name, oldest_first[index], mut report)!75		}76	}77	upload_events(store, mut backend, name, mut report)!78	if !backend.put_if_match(head_key(name), local.id.bytes(), remote_head.tag)! {79		return error('remote has new saves — run koh steal')80	}81	store.set_state(offered_head_key, local.id)!82	return report83}8485// saves_after lists the local saves newer than `theirs`, oldest first. It fails86// when `theirs` is not on this line, which is the honest answer: we cannot87// offer on top of a save we do not have.88fn saves_after(store &storage.Store, ours string, theirs string) ![]history.Save {89	mut newer := []history.Save{}90	mut cursor := ours91	for cursor != '' && cursor != theirs {92		save := history.by_id(store, cursor) or { return error('save ${cursor} is missing') }93		newer << save94		cursor = save.parent95	}96	if cursor != theirs {97		return error('${theirs} is not on this line')98	}99	newer.reverse_in_place()100	return newer101}102103fn upload_save(store &storage.Store, mut backend remote.Backend, repo string, save history.Save, mut report OfferReport) ! {104	files := history.files_of(store, save.id)!105	for file in files {106		if backend.put_if_absent(object_key(repo, file.hash), store.object_blob(file.hash)!)! {107			report.objects++108		}109	}110	record := save_record(store, save)!111	if backend.put_if_absent(save_key(repo, save.id), encode_save(record).bytes())! {112		report.saves++113	}114}115116// upload_events sends every event the remote does not have. Events are tiny, so117// listing them costs less than tracking which ones were sent.118fn upload_events(store &storage.Store, mut backend remote.Backend, repo string, mut report OfferReport) ! {119	already := backend.list(events_prefix(repo))!120	mut already_sent := map[string]bool{}121	for key in already {122		already_sent[os.base(key)] = true123	}124	for event in events.all(store)! {125		if event.id in already_sent {126			continue127		}128		if backend.put_if_absent(event_key(repo, event.id), encode_event(event).bytes())! {129			report.events++130		}131	}132}133134// fetched holds everything a steal read from the remote, before any of it is135// written locally. Network work happens first; the database gets one136// transaction, so a failure leaves nothing half-applied.137pub struct Fetched {138pub mut:139	repo_objects    map[string][]u8140	saves           []SaveRecord // oldest first141	events_out      []events.Event142	head            string143	base            string // the newest save both sides already shared, '' when none144	remote_save_ids map[string]bool145}146147pub fn fetch_back_to_known(store &storage.Store, mut backend remote.Backend, repo string) !Fetched {148	remote_head := backend.get(head_key(repo))!149	if !remote_head.found {150		return error('the remote has no "${repo}" yet — run koh offer here first')151	}152	mut fetched := Fetched{153		head:            remote_head.body.bytestr().trim_space()154		remote_save_ids: map[string]bool{}155	}156	mut cursor := fetched.head157	for cursor != '' && history.by_id(store, cursor) == none {158		record_object := backend.get(save_key(repo, cursor))!159		if !record_object.found {160			return error('the remote is missing save ${cursor} — run koh offer there')161		}162		record := decode_save(record_object.body)!163		for file in record.files {164			if !store.has_object(file.hash)! && file.hash !in fetched.repo_objects {165				blob := backend.get(object_key(repo, file.hash))!166				if !blob.found {167					return error('the remote is missing object ${file.hash} that save ${record.id} needs')168				}169				fetched.repo_objects[file.hash] = blob.body170			}171		}172		fetched.remote_save_ids[record.id] = true173		fetched.saves << record174		cursor = record.parent175	}176	fetched.saves.reverse_in_place()177	fetched.base = cursor178	fetched.events_out = fetch_events(store, mut backend, repo)!179	return fetched180}181182fn fetch_events(store &storage.Store, mut backend remote.Backend, repo string) ![]events.Event {183	have := events.all(store)!184	mut known := map[string]bool{}185	for event in have {186		known[event.id] = true187	}188	mut incoming := []events.Event{}189	for key in backend.list(events_prefix(repo))! {190		id := os.base(key)191		if id in known {192			continue193		}194		body := backend.get(event_key(repo, id))!195		if !body.found {196			continue197		}198		incoming << decode_event(body.body)!199	}200	return incoming201}202203// write_fetched lands a fetch in one transaction: objects first (verified), then204// the saves they belong to, then the events.205pub fn write_fetched(store &storage.Store, fetched Fetched) ! {206	for hash, blob in fetched.repo_objects {207		store.put_object_blob(hash, blob)!208	}209	for record in fetched.saves {210		history.import_save(store, to_save(record), to_files(record))!211	}212	for event in fetched.events_out {213		events.write(store, event)!214		if event.kind == events.kind_strip {215			// Somebody stripped a path: remove it here too, so the two216			// histories agree from now on.217			apply_strip(store, event)!218		}219	}220}221222// can_fast_forward reports whether our line sits behind the remote's, which is223// the case when every local save is an ancestor of the remote head.224pub fn can_fast_forward(store &storage.Store, remote_head string, remote_ids map[string]bool) bool {225	mut cursor := remote_head226	for cursor != '' {227		if history.by_id(store, cursor) == none {228			return false229		}230		save := history.by_id(store, cursor) or { return false }231		if save.parent == '' {232			// Reached the first save without leaving the line we have.233			return true234		}235		cursor = save.parent236	}237	return false238}239240// PushReport says what a strip managed to make visible on the remote.241pub struct PushReport {242pub mut:243	records    int244	objects    int245	event_sent bool246}247248// push_strip makes a strip visible on the remote. The event itself travels like249// any other event; on top of that, every remote save record that listed the250// path is rewritten in place, and — when the strip covered the whole local251// history — the objects nobody references any more are deleted.252//253// The caller passes what it stripped, because the local rows are already gone254// by the time the remote hears about it. Every step is safe to repeat, so an255// interrupted strip finishes on the next run.256pub fn push_strip(store &storage.Store, mut backend remote.Backend, path string, save_ids []string, object_hashes []string, drop_objects bool) !PushReport {257	name := repo_name(store, mut backend)!258	mut report := PushReport{}259	mut upload := OfferReport{260		repo: name261	}262	upload_events(store, mut backend, name, mut upload)!263	report.event_sent = upload.events > 0264265	for save_id in save_ids {266		remote_object := backend.get(save_key(name, save_id))!267		if !remote_object.found {268			continue269		}270		record := decode_save(remote_object.body)!271		mut kept := []FileRecord{}272		for file in record.files {273			if file.path == path || file.path.starts_with('${path.trim_right('/')}/') {274				continue275			}276			kept << file277		}278		if kept.len == record.files.len {279			continue280		}281		if !backend.put_if_match(save_key(name, save_id),282			encode_save(rewrite(record, kept)).bytes(), remote_object.tag)! {283			return error('the remote copy of save ${save_id} moved while stripping — run koh steal, then strip again')284		}285		report.records++286	}287	if drop_objects {288		for hash in object_hashes {289			// Only objects no local save refers to any more: with `--save`, the290			// other saves still need them.291			if !store.has_object(hash)! {292				backend.delete(object_key(name, hash))!293				report.objects++294			}295		}296	}297	return report298}299300// rewrite returns the save record with a new file list.301fn rewrite(record SaveRecord, files []FileRecord) SaveRecord {302	return SaveRecord{303		id:            record.id304		parent:        record.parent305		message:       record.message306		created_at:    record.created_at307		machine:       record.machine308		is_checkpoint: record.is_checkpoint309		size_bytes:    record.size_bytes310		agent:         record.agent311		session:       record.session312		restores:      record.restores313		replayed_from: record.replayed_from314		files:         files315	}316}317318// apply_strip is what a machine does when it hears about somebody else's strip:319// the same removal, so both histories agree from then on. It returns how many320// saved files it removed.321pub fn apply_strip(store &storage.Store, event events.Event) !int {322	mut query := "DELETE FROM save_files WHERE (path = ? OR path LIKE ? ESCAPE '\\')"323	if event.save_id != '' {324		query += ' AND save_id = ?'325	}326	mut stmt := store.db.prepare(query)!327	stmt.bind_text(1, event.path)!328	stmt.bind_text(2, like_under(event.path))!329	if event.save_id != '' {330		stmt.bind_text(3, event.save_id)!331	}332	stmt.step()!333	removed := store.db.changes()334	stmt.finalize()335	if removed > 0 {336		store.drop_unreferenced_objects()!337	}338	return removed339}340341// like_under escapes LIKE's own wildcards so a path containing `%` or `_`342// matches itself and nothing else.343fn like_under(folder string) string {344	escaped := folder.replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_')345	return '${escaped}/%'346}