3.4.2: a steal over HTTPS reads the bundle as it arrives instead of holding it: peach-browser 2.5 GB -> 1.19 GB

@@ -1 +1 @@1-3.4.11+3.4.2

@@ -441,23 +441,24 @@441441 // steal_repo fetches a whole repo into ./<name>, for a machine that does not442442 // have it yet.443443 fn steal_repo(name string, args Args) int {444-	backend := this_machine_backend() or { return refuse(args.json(), err.msg()) }444+	// Counted against the listings, so the line can say how many of each kind are in.445+	backend := Watched{446+		label: 'stealing ${name}'447+		whole: true448+		inner: this_machine_backend() or { return refuse(args.json(), err.msg()) }449+	}445450 	return steal_whole(name, args, backend, '')446451 }447452 448453 // steal_whole is that fetch, from whichever remote was chosen: the whole history in, then the head's449454 // files onto the disk. A non-empty `source` means the repo came from a read-only address and stays450455 // local until this machine signs in there.451-fn steal_whole(name string, args Args, inner remote.Backend, source string) int {456+fn steal_whole(name string, args Args, chosen remote.Backend, source string) int {452457 	target := os.join_path(os.getwd(), name)453458 	if os.exists(target) {454459 		return refuse(args.json(), '${name} already exists here — run steal inside it instead')455460 	}456-	mut backend := remote.Backend(Watched{457-		label: 'stealing ${name}'458-		whole: true459-		inner: inner460-	})461+	mut backend := chosen461462 	fetched := exchange.fetch_repo(mut backend, name) or { return refuse(args.json(), err.msg()) }462463 	store := land_repo(target, name, fetched, source) or {463464 		os.rmdir_all(target) or {}

@@ -1631,16 +1631,24 @@16311631 	}16321632 }16331633 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\nend 0\n'.bytes())!1637-	assert fetched.repo_objects == {1638-		'aa': 'abc'.bytes()1639-		'bb': []u8{}1634+fn test_a_bundle_lands_each_entry_however_it_is_split_and_refuses_a_cut_or_foreign_one() {1635+	whole := 'objects/aa 3\nabcobjects/bb 0\nend 0\n'1636+	for size in [1, 2, 5, whole.len] {1637+		mut bundle := Bundle{}1638+		for at := 0; at < whole.len; at += size {1639+			bundle.feed(whole[at..int_min(at + size, whole.len)].bytes())!1640+		}1641+		bundle.finish()!1642+		assert bundle.fetched.repo_objects == {1643+			'aa': 'abc'.bytes()1644+			'bb': []u8{}1645+		}16401646 	}16411647 	for cut in ['objects/aa 3\nabc', 'objects/aa 4\nabc', 'objects/aa 3', 'objects/aa x\nabc',16421648 		'releases/v1 1\nx', 'end 0\nobjects/aa 3\nabc'] {1643-		take_bundle(mut fetched, cut.bytes()) or { continue }1649+		mut bundle := Bundle{}1650+		bundle.feed(cut.bytes()) or { continue }1651+		bundle.finish() or { continue }16441652 		assert false, cut16451653 	}16461654 }

@@ -566,9 +566,11 @@566566 	mut fetched := Fetched{567567 		head: remote_head.body.bytestr().trim_space()568568 	}569-	bundle := backend.get(bundle_key(name))!570-	if bundle.found {571-		take_bundle(mut fetched, bundle.body)!569+	mut bundle := &Bundle{570+		fetched: fetched571+	}572+	if fetch_bundle(mut backend, bundle_key(name), mut bundle)! {573+		fetched = bundle.fetched572574 	} else {573575 		for kind in ['objects', 'saves', 'events'] {574576 			for key in backend.list('${name}/${kind}')! {@@ -584,33 +586,78 @@584586 	return fetched585587 }586588 587-// take_bundle splits §22.7's bundle — `<key> <length>\n` and then that many bytes, over and over,588-// until `end 0` — into what one GET of each key would have returned. A connection that drops589-// between two entries still looks like a whole bundle, so only the end line proves it arrived.590-fn take_bundle(mut fetched Fetched, body []u8) ! {589+// fetch_bundle reads §22.7's bundle, when the remote has one. Over HTTPS it is read as it arrives:590+// held whole, a 282 MB bundle cost 2.5 GB, mostly copies the HTTP client makes.591+fn fetch_bundle(mut backend remote.Backend, key string, mut bundle Bundle) !bool {592+	found := if mut backend is remote.Web {593+		backend.stream(key, fn [mut bundle] (piece []u8) ! {594+			bundle.feed(piece)!595+		})!596+	} else {597+		whole := backend.get(key)!598+		if whole.found {599+			bundle.feed(whole.body)!600+		}601+		whole.found602+	}603+	if found {604+		bundle.finish()!605+	}606+	return found607+}608+609+// Bundle takes §22.7's bundle — `<key> <length>\n` and then that many bytes, over and over, until610+// `end 0` — a piece at a time, into what one GET of each key would have returned. A connection611+// that drops between two entries still looks like a whole bundle, so only the end line proves it612+// arrived.613+struct Bundle {614+mut:615+	fetched Fetched616+	pending []u8 // the start of an entry still arriving617+	ended   bool618+}619+620+fn (mut b Bundle) feed(piece []u8) ! {621+	b.pending << piece591622 	mut at := 0592-	for at < body.len {593-		// Scanned in place: a slice of `body` copies the rest of it, and that per entry is quadratic.623+	for at < b.pending.len {624+		// Scanned in place: a slice copies on this V, and copying the rest per entry is quadratic.594625 		mut newline := at595-		for newline < body.len && body[newline] != `\n` {626+		for newline < b.pending.len && b.pending[newline] != `\n` {596627 			newline++597628 		}598-		if newline == body.len {599-			return error('the bundle ends part way through an entry')629+		if newline == b.pending.len {630+			break600631 		}601-		header := body[at..newline].bytestr()602-		if header == 'end 0' && newline + 1 == body.len {603-			return632+		header := b.pending[at..newline].bytestr()633+		if b.ended {634+			return error('the bundle goes on after its end')604635 		}636+		if header == 'end 0' {637+			b.ended = true638+			at = newline + 1639+			continue640+		}605641 		length := strconv.atoi(header.all_after(' '))!606642 		start := newline + 1607-		if length < 0 || length > body.len - start {608-			return error('the bundle ends part way through ${header.all_before(' ')}')643+		if length < 0 {644+			return error('the bundle has a negative length for ${header.all_before(' ')}')609645 		}610-		take(mut fetched, header.all_before(' '), body[start..start + length])!646+		if length > b.pending.len - start {647+			break648+		}649+		take(mut b.fetched, header.all_before(' '), b.pending[start..start + length])!611650 		at = start + length612651 	}613-	return error('the bundle was cut short — try the steal again')652+	if at > 0 {653+		b.pending = b.pending[at..]654+	}655+}656+657+fn (b &Bundle) finish() ! {658+	if !b.ended || b.pending.len > 0 {659+		return error('the bundle was cut short — try the steal again')660+	}614661 }615662 616663 // take files one fetched key, relative to the repo: an object, a save or an event.

@@ -45,17 +45,11 @@4545 }4646 4747 pub fn (mut w Web) get(key string) !Object {48-	on_body := w.on_body4948 	response := http.fetch(50-		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-		}49+		url:            w.url(key)50+		method:         .get51+		enable_http2:   false52+		allow_redirect: false5953 	)!6054 	if response.status_code == 404 {6155 		return Object{}@@ -68,6 +62,38 @@6862 		body:  response.body.bytes()6963 		tag:   response.header.get_custom('etag', http.HeaderQueryConfig{}) or { '' }7064 	}65+}66+67+// stream is get for a body too big to hold: each piece goes to `sink` as it arrives, and none of it68+// is kept. False when the key does not exist.69+pub fn (mut w Web) stream(key string, sink fn (piece []u8) !) !bool {70+	on_body := w.on_body71+	response := http.fetch(72+		url:                w.url(key)73+		method:             .get74+		enable_http2:       false75+		allow_redirect:     false76+		// Enough for the status line and headers, which are all that is read from the response.77+		stop_copying_limit: 64 * 102478+		on_progress_body:   fn [sink, on_body] (_ &http.Request, piece []u8, received u64, _ u64, status int) ! {79+			// V hands over a read that holds only part of the headers as if it were body, with80+			// nothing received yet; a real piece of body always counts itself.81+			if status < 200 || status >= 300 || received == 0 {82+				return83+			}84+			sink(piece)!85+			if on_body != unsafe { nil } && (received - u64(piece.len)) >> 20 != received >> 20 {86+				on_body(received)87+			}88+		}89+	)!90+	if response.status_code == 404 {91+		return false92+	}93+	if response.status_code < 200 || response.status_code >= 300 {94+		return error('web GET ${key} failed: ${response.status_code}')95+	}96+	return true7197 }7298 7399 pub fn (mut w Web) list(prefix string) ![]string {

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