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	if remote_head.found {63		theirs := remote_head.body.bytestr().trim_space()64		newer := saves_after(store, local.id, theirs) or {65			return error('remote has new saves — run koh steal')66		}67		for save in newer {68			upload_save(store, mut backend, name, save, mut report)!69		}70	} else {71		oldest_first := history.line(store)!72		for index := oldest_first.len - 1; index >= 0; index-- {73			upload_save(store, mut backend, name, oldest_first[index], mut report)!74		}75	}76	upload_events(store, mut backend, name, mut report)!77	if !backend.put_if_match(head_key(name), local.id.bytes(), remote_head.tag)! {78		return error('remote has new saves — run koh steal')79	}80	store.set_state(offered_head_key, local.id)!81	return report82}8384// saves_after lists the local saves newer than `theirs`, oldest first. It fails85// when `theirs` is not on this line, which is the honest answer: we cannot86// offer on top of a save we do not have.87fn saves_after(store &storage.Store, ours string, theirs string) ![]history.Save {88	mut newer := []history.Save{}89	mut cursor := ours90	for cursor != '' && cursor != theirs {91		save := history.by_id(store, cursor) or { return error('save ${cursor} is missing') }92		newer << save93		cursor = save.parent94	}95	if cursor != theirs {96		return error('${theirs} is not on this line')97	}98	newer.reverse_in_place()99	return newer100}101102fn upload_save(store &storage.Store, mut backend remote.Backend, repo string, save history.Save, mut report OfferReport) ! {103	files := history.files_of(store, save.id)!104	for file in files {105		if backend.put_if_absent(object_key(repo, file.hash), store.object_blob(file.hash)!)! {106			report.objects++107		}108	}109	record := save_record(store, save)!110	if backend.put_if_absent(save_key(repo, save.id), encode_save(record).bytes())! {111		report.saves++112	}113}114115// upload_events sends every event the remote does not have. Events are tiny, so116// listing them costs less than tracking which ones were sent.117fn upload_events(store &storage.Store, mut backend remote.Backend, repo string, mut report OfferReport) ! {118	already := backend.list(events_prefix(repo))!119	mut already_sent := map[string]bool{}120	for key in already {121		already_sent[os.base(key)] = true122	}123	for event in events.all(store)! {124		if event.id in already_sent {125			continue126		}127		if backend.put_if_absent(event_key(repo, event.id), encode_event(event).bytes())! {128			report.events++129		}130	}131}132133// fetched holds everything a steal read from the remote, before any of it is134// written locally. Network work happens first; the database gets one135// transaction, so a failure leaves nothing half-applied.136pub struct Fetched {137pub mut:138	repo_objects    map[string][]u8139	saves           []SaveRecord // oldest first140	events_out      []events.Event141	head            string142	base            string // the newest save both sides already shared, '' when none143	remote_save_ids map[string]bool144}145146pub fn fetch_back_to_known(store &storage.Store, mut backend remote.Backend, repo string) !Fetched {147	remote_head := backend.get(head_key(repo))!148	if !remote_head.found {149		return error('the remote has no "${repo}" yet — run koh offer here first')150	}151	mut fetched := Fetched{152		head:            remote_head.body.bytestr().trim_space()153		remote_save_ids: map[string]bool{}154	}155	mut cursor := fetched.head156	for cursor != '' && history.by_id(store, cursor) == none {157		record_object := backend.get(save_key(repo, cursor))!158		if !record_object.found {159			return error('the remote is missing save ${cursor} — run koh offer there')160		}161		record := decode_save(record_object.body)!162		for file in record.files {163			if !store.has_object(file.hash)! && file.hash !in fetched.repo_objects {164				blob := backend.get(object_key(repo, file.hash))!165				if !blob.found {166					return error('the remote is missing object ${file.hash} that save ${record.id} needs')167				}168				fetched.repo_objects[file.hash] = blob.body169			}170		}171		fetched.remote_save_ids[record.id] = true172		fetched.saves << record173		cursor = record.parent174	}175	fetched.saves.reverse_in_place()176	fetched.base = cursor177	fetched.events_out = fetch_events(store, mut backend, repo)!178	return fetched179}180181fn fetch_events(store &storage.Store, mut backend remote.Backend, repo string) ![]events.Event {182	have := events.all(store)!183	mut known := map[string]bool{}184	for event in have {185		known[event.id] = true186	}187	mut incoming := []events.Event{}188	for key in backend.list(events_prefix(repo))! {189		id := os.base(key)190		if id in known {191			continue192		}193		body := backend.get(event_key(repo, id))!194		if !body.found {195			continue196		}197		incoming << decode_event(body.body)!198	}199	return incoming200}201202// write_fetched lands a fetch in one transaction: objects first (verified), then203// the saves they belong to, then the events.204pub fn write_fetched(store &storage.Store, fetched Fetched) ! {205	for hash, blob in fetched.repo_objects {206		store.put_object_blob(hash, blob)!207	}208	for record in fetched.saves {209		history.import_save(store, to_save(record), to_files(record))!210	}211	for event in fetched.events_out {212		if !events.receive(store, event)! {213			continue214		}215		if event.kind == events.kind_strip {216			// Somebody stripped a path: remove it here too, so the two217			// histories agree from now on.218			apply_strip(store, event)!219		}220	}221}222223// can_fast_forward reports whether our line sits behind the remote's, which is224// the case when every local save is an ancestor of the remote head.225pub fn can_fast_forward(store &storage.Store, remote_head string, remote_ids map[string]bool) bool {226	mut cursor := remote_head227	for cursor != '' {228		if history.by_id(store, cursor) == none {229			return false230		}231		save := history.by_id(store, cursor) or { return false }232		if save.parent == '' {233			// Reached the first save without leaving the line we have.234			return true235		}236		cursor = save.parent237	}238	return false239}240241// PushReport says what a strip managed to make visible on the remote.242pub struct PushReport {243pub mut:244	records    int245	objects    int246	event_sent bool247}248249// push_strip makes a strip visible on the remote. The event itself travels like250// any other event; on top of that, every remote save record that listed the251// path is rewritten in place, and — when the strip covered the whole local252// history — the objects nobody references any more are deleted.253//254// The caller passes what it stripped, because the local rows are already gone255// by the time the remote hears about it. Every step is safe to repeat, so an256// interrupted strip finishes on the next run.257pub fn push_strip(store &storage.Store, mut backend remote.Backend, path string, save_ids []string, object_hashes []string, drop_objects bool) !PushReport {258	name := repo_name(store, mut backend)!259	mut report := PushReport{}260	mut upload := OfferReport{261		repo: name262	}263	upload_events(store, mut backend, name, mut upload)!264	report.event_sent = upload.events > 0265266	for save_id in save_ids {267		remote_object := backend.get(save_key(name, save_id))!268		if !remote_object.found {269			continue270		}271		record := decode_save(remote_object.body)!272		mut kept := []FileRecord{}273		for file in record.files {274			if file.path == path || file.path.starts_with('${path.trim_right('/')}/') {275				continue276			}277			kept << file278		}279		if kept.len == record.files.len {280			continue281		}282		if !backend.put_if_match(save_key(name, save_id),283			encode_save(rewrite(record, kept)).bytes(), remote_object.tag)! {284			return error('the remote copy of save ${save_id} moved while stripping — run koh steal, then strip again')285		}286		report.records++287	}288	if drop_objects {289		for hash in object_hashes {290			// Only objects no local save refers to any more: with `--save`, the291			// other saves still need them.292			if !store.has_object(hash)! {293				backend.delete(object_key(name, hash))!294				report.objects++295			}296		}297	}298	return report299}300301// rewrite returns the save record with a new file list.302fn rewrite(record SaveRecord, files []FileRecord) SaveRecord {303	return SaveRecord{304		id:            record.id305		parent:        record.parent306		message:       record.message307		created_at:    record.created_at308		machine:       record.machine309		is_checkpoint: record.is_checkpoint310		size_bytes:    record.size_bytes311		agent:         record.agent312		session:       record.session313		restores:      record.restores314		replayed_from: record.replayed_from315		files:         files316	}317}318319// apply_strip is what a machine does when it hears about somebody else's strip:320// the same removal, so both histories agree from then on. It returns how many321// saved files it removed.322pub fn apply_strip(store &storage.Store, event events.Event) !int {323	mut query := "DELETE FROM save_files WHERE (path = ? OR path LIKE ? ESCAPE '\\')"324	if event.save_id != '' {325		query += ' AND save_id = ?'326	}327	mut stmt := store.db.prepare(query)!328	stmt.bind_text(1, event.path)!329	stmt.bind_text(2, like_under(event.path))!330	if event.save_id != '' {331		stmt.bind_text(3, event.save_id)!332	}333	stmt.step()!334	removed := store.db.changes()335	stmt.finalize()336	if removed > 0 {337		store.drop_unreferenced_objects()!338	}339	return removed340}341342// like_under escapes LIKE's own wildcards so a path containing `%` or `_`343// matches itself and nothing else.344fn like_under(folder string) string {345	escaped := folder.replace('\\', '\\\\').replace('%', '\\%').replace('_', '\\_')346	return '${escaped}/%'347}