3.4.0: a whole-repo steal reads Kepr's /koh/bundle in one request, showing MiB received, and falls back to key by key

@@ -1 +1 @@1-3.3.11+3.4.0

@@ -431,8 +431,11 @@431431 	if exchange.signed_in_to(base) {432432 		return steal_repo(repo, args)433433 	}434-	backend := exchange.read_only('${base}/${repo}') or { return refuse(args.json(), err.msg()) }435-	return steal_whole(repo, args, backend, '${base}/${repo}')434+	mut web := remote.new_web(base) or { return refuse(args.json(), err.msg()) }435+	web.on_body = fn [repo] (received u64) {436+		working('stealing ${repo} · ${fmt.human_bytes(i64(received))} received')437+	}438+	return steal_whole(repo, args, web, '${base}/${repo}')436439 }437440 438441 // steal_repo fetches a whole repo into ./<name>, for a machine that does not

@@ -1630,3 +1630,16 @@16301630 		assert err.msg().contains('koh-remote 2')16311631 	}16321632 }1633+1634+fn test_a_bundle_lands_each_entry_and_refuses_a_cut_or_foreign_one() {1635+	mut fetched := Fetched{}1636+	take_bundle(mut fetched, 'objects/aa 3\nabcobjects/bb 0\n'.bytes())!1637+	assert fetched.repo_objects == {1638+		'aa': 'abc'.bytes()1639+		'bb': []u8{}1640+	}1641+	for cut in ['objects/aa 4\nabc', 'objects/aa 3', 'objects/aa x\nabc', 'releases/v1 1\nx'] {1642+		take_bundle(mut fetched, cut.bytes()) or { continue }1643+		assert false, cut1644+	}1645+}

@@ -41,6 +41,10 @@4141 	return '${repo}/head'4242 }4343 44+pub fn bundle_key(repo string) string {45+	return '${repo}/bundle'46+}47+4448 pub fn events_prefix(repo string) string {4549 	return '${repo}/events'4650 }

@@ -9,7 +9,6 @@99 module exchange1010 1111 import compress.zlib12-import os1312 import src.apply1413 import src.events1514 import src.fmt@@ -17,6 +16,7 @@1716 import src.remote1817 import src.snapshot1918 import src.storage19+import strconv2020 2121 pub enum Side {2222 	mine@@ -563,38 +563,72 @@563563 	if !remote_head.found {564564 		return error('the remote has no repo called "${name}"')565565 	}566-	save_ids := backend.list('${name}/saves')!567-	object_keys := backend.list('${name}/objects')!568-	event_keys := backend.list('${name}/events')!569566 	mut fetched := Fetched{570567 		head: remote_head.body.bytestr().trim_space()571568 	}572-	for key in object_keys {573-		blob := backend.get(key)!574-		if blob.found {575-			fetched.repo_objects[os.base(key)] = blob.body569+	bundle := backend.get(bundle_key(name))!570+	if bundle.found {571+		take_bundle(mut fetched, bundle.body)!572+	} else {573+		for kind in ['objects', 'saves', 'events'] {574+			for key in backend.list('${name}/${kind}')! {575+				body := backend.get(key)!576+				if body.found {577+					take(mut fetched, key.all_after('${name}/'), body.body)!578+				}579+			}576580 		}577581 	}578-	for key in save_ids {579-		body := backend.get(key)!580-		if !body.found {581-			continue582+	fetched.saves = parent_first(fetched.saves)!583+	fetched.events_out.sort(a.id < b.id)584+	return fetched585+}586+587+// take_bundle splits §22.7's bundle — `<key> <length>\n` and then that many bytes, over and over —588+// into what one GET of each key would have returned.589+fn take_bundle(mut fetched Fetched, body []u8) ! {590+	mut at := 0591+	for at < body.len {592+		// Scanned in place: a slice of `body` copies the rest of it, and that per entry is quadratic.593+		mut newline := at594+		for newline < body.len && body[newline] != `\n` {595+			newline++582596 		}583-		mut record := decode_save(body.body)!584-		fetched.unsafe_paths << drop_unsafe_paths(mut record)585-		fetched.saves << record597+		if newline == body.len {598+			return error('the bundle ends part way through an entry')599+		}600+		header := body[at..newline].bytestr()601+		length := strconv.atoi(header.all_after(' '))!602+		start := newline + 1603+		if length < 0 || length > body.len - start {604+			return error('the bundle ends part way through ${header.all_before(' ')}')605+		}606+		take(mut fetched, header.all_before(' '), body[start..start + length])!607+		at = start + length586608 	}587-	fetched.saves = parent_first(fetched.saves)!588-	for key in event_keys {589-		body := backend.get(key)!590-		if body.found {591-			event := decode_event(body.body)!609+}610+611+// take files one fetched key, relative to the repo: an object, a save or an event.612+fn take(mut fetched Fetched, key string, body []u8) ! {613+	match key.all_before('/') {614+		'objects' {615+			fetched.repo_objects[key.all_after('/')] = body616+		}617+		'saves' {618+			mut record := decode_save(body)!619+			fetched.unsafe_paths << drop_unsafe_paths(mut record)620+			fetched.saves << record621+		}622+		'events' {623+			event := decode_event(body)!592624 			if !ignored_strip(event) {593625 				fetched.events_out << event594626 			}595627 		}628+		else {629+			return error('the remote sent "${key}", which is not part of a repo')630+		}596631 	}597-	return fetched598632 }599633 600634 // parent_first orders saves so every parent lands before its children, as the

@@ -7,6 +7,8 @@77 pub struct Web {88 pub mut:99 	base string10+	// on_body hears how much of a response has arrived, each time it passes another MiB.11+	on_body fn (received u64) = unsafe { nil }1012 }1113 1214 pub fn new_web(base string) !Web {@@ -43,11 +45,17 @@4345 }4446 4547 pub fn (mut w Web) get(key string) !Object {48+	on_body := w.on_body4649 	response := http.fetch(47-		url:            w.url(key)48-		method:         .get49-		enable_http2:   false50-		allow_redirect: false50+		url:              w.url(key)51+		method:           .get52+		enable_http2:     false53+		allow_redirect:   false54+		on_progress_body: fn [on_body] (_ &http.Request, chunk []u8, received u64, _ u64, _ int) ! {55+			if on_body != unsafe { nil } && (received - u64(chunk.len)) >> 20 != received >> 20 {56+				on_body(received)57+			}58+		}5159 	)!5260 	if response.status_code == 404 {5361 		return Object{}

@@ -1,6 +1,6 @@11 Module {22 	name: 'koh3'3-	version: '3.3.1'3+	version: '3.4.0'44 	description: 'One SQLite file per repo, one line of history.'55 	license: 'MIT'66 }