mirror_worker: stream and incrementally commit add-entries uploads - #267
mirror_worker: stream and incrementally commit add-entries uploads#267lukevalenta wants to merge 8 commits into
Conversation
|
/bonk review |
|
UnknownError: Expected 'id' to be a string. |
|
@lukevalenta Bonk workflow failed. Check the logs for details. View workflow run · To retry, trigger Bonk again. |
a65ac22 to
26671bf
Compare
26671bf to
998be29
Compare
ccb281d to
8353d9e
Compare
|
/bigbonk review |
![ask-bonk[bot]](https://avatars.githubusercontent.com/in/2454580?s=60&v=4){kind=small}
There was a problem hiding this comment.
Review Summary
This PR replaces the whole-body buffering in add-entries with a streaming decode and incremental commit pipeline. The design is sound: StreamBuffer bridges the async chunk stream to synchronous Read-based parsers via a retry-on-UnexpectedEof loop, GzipInflater wraps flate2::write::GzDecoder for incremental decompression, and flush_chunk advances the persisted-entry frontier every commit_packages packages. The invariants that make this safe (immutable content-addressed tiles, monotone CAS frontier advance) are correctly identified and documented.
The code is well-structured, thoroughly documented, and well-tested. Clippy is clean, all 72 existing+new tests pass, and formatting is correct. I have a few observations but nothing blocking.
Observations
1. Gzip decode errors surface as 500, not 400 (body.rs:101-105, add_entries.rs:244)
GzipInflater::push and finish map decode failures to worker::Error (via Error::from(format!("gzip decode failed: ..."))). When parse_header or parse_next_package calls buf.pull_one().await?, that ? propagates the worker::Error through ApiResult and hits From<worker::Error> for AppError, which maps it to InternalServerError (500). A malformed gzip body is a client fault (400), not a server fault. The old code returned AppError::BadRequest("gzip decode failed: ...") explicitly. This is a behavioral regression for malformed-gzip requests.
One fix: have decoded_stream/gunzip map decode errors to a distinguishable error type that the add-entries handler can catch and map to 400. Alternatively, gunzip could wrap decode errors in a sentinel that parse_header/parse_next_package recognize. The trailing-data check at line 244 has the same issue: buf.pull_one().await? would 500 on a gzip error that surfaces while draining the stream, when it should 400.
2. chunk_pkgs counts packages that contributed no entries (add_entries.rs:378)
chunk_pkgs is incremented unconditionally (line 378) even when the package is entirely below initial_next (all entries already persisted) and nothing was added to chunk. This means a long run of already-persisted packages can trigger flush_chunk calls with an empty chunk (the if !chunk.is_empty() guard at line 381 prevents actual I/O, but the counter resets and the pattern repeats). This is functionally harmless but slightly misleading -- chunk_pkgs doesn't reflect the number of packages that actually contributed buffered entries. Not a bug, just a readability nit.
3. parse_next_package returns CleanEof on empty-buffer-after-pull-failure (add_entries.rs:699-702)
When pull_one returns false (stream ended) and buf.len() == 0, the function returns CleanEof. But this path is reached when the buffer was non-empty at loop entry (the early-return at line 687 didn't fire), a partial parse consumed nothing (cursor position 0 on UnexpectedEof), the pull failed, and the buffer is now empty. The only way buf.len() == 0 here is if the buffer was already empty before the pull -- but if it were empty, the early return at line 687 would have fired (assuming is_eof was already true). So this path is only reachable if the buffer was non-empty, the parse failed with UnexpectedEof, and then pull_one returned false and the buffer somehow became empty, which can't happen since pull_one doesn't drain the buffer. This means the buf.len() == 0 branch at line 700 is dead code in practice, and the function would always return MidPackageEof when the stream ends mid-parse. Not a bug, but the dead branch is slightly misleading.
Actually, on closer inspection: the buffer could be empty if the previous iteration's successful parse consumed all bytes (via consume), then a new iteration enters the loop with an empty buffer, fails UnexpectedEof on the empty slice, pulls and gets EOF. In that case buf.len() == 0 is true and CleanEof is correct. So this path is reachable and correct -- it handles the edge case where the previous package consumed exactly all buffered bytes and the stream has no more data.
4. StreamBuffer::pull_one error doesn't set eof (stream_buffer.rs:83)
When the underlying stream yields Some(Err(e)), pull_one propagates the error but does not set self.eof = true. If the caller catches the error and retries (which current callers don't do -- they all ?-propagate), the next pull_one would re-poll the stream. In practice this is fine because all call sites propagate errors immediately, but it's a subtle API footprint: is_eof() returns false after an error, which could be surprising if someone later adds error recovery.
5. Minor: doc comment on parse_header mentions Ok(Err(resp)) (add_entries.rs:637)
The doc says "Returns Ok(Ok(header)) on success or Ok(Err(resp)) where resp is a fully-formed 400 response" but the actual signature returns ApiResult<AddEntriesRequestHeader> (no nested Result). The doc describes the old design; the actual code correctly returns Ok(header) or Err(AppError::BadRequest(...)).
Verdict
The architecture is solid and the implementation is careful. Finding #1 (gzip errors surfacing as 500 instead of 400) is the only behavioral concern; the rest are minor. Good test coverage, especially the chunked_commit_matches_single_commit test that validates the core streaming invariant.
|
Review posted. Here's a summary: The PR is well-designed and well-implemented. All 72 tests pass, clippy is clean, and formatting is correct. The core architecture -- The one behavioral concern is that gzip decode errors now surface as 500 instead of 400: the old code explicitly returned |
4a8688b to
4a55b2b
Compare
4a55b2b to
875c691
Compare
|
/bigbonk review |
|
@lukevalenta Bonk workflow failed. Check the logs for details. View workflow run · To retry, trigger Bonk again. |
9e14a9a to
3b512ce
Compare
3b512ce to
e7a7b23
Compare
Replace the whole-body buffer with a streaming decode: gunzip and parse the body as it arrives through a small stream-buffer adapter (retry the parse as chunks land, consume on success), so a large upload no longer holds the entire body in memory. Persist incrementally every `commit_packages` packages (default 32, config capped at 1024) so a long or interrupted upload advances the frontier as it streams rather than only at the end. Each flush commits from the last persisted frontier and is resumable; the running frontier is threaded locally across flushes so the DO is not re-queried between chunks. Tiles are immutable and content-addressed and the DO advance is a monotone compare-and-swap, so a repeated or concurrent flush of the same range is harmless.
Gzip decode/trailer failures were surfaced as worker::Error, which the handler's blanket From<worker::Error> mapped to 500. A malformed or truncated gzip request body is a client fault. Introduce a typed BodyError (Decode vs Transport) on the body stream so decode faults map to 400 while genuine transport failures stay 500; parse_next_package now returns ApiResult so the distinction survives to the handler.
- Count only entry-contributing packages toward a chunk flush, so a run of already-persisted packages no longer triggers empty flushes; drop the now-redundant empty-chunk guard (commit_packages >= 1). - Mark StreamBuffer eof on a stream error so is_eof() stays consistent and a recovering caller won't re-poll the failed stream. - Explain the clean-vs-mid-package truncation branch in parse_next_package, and fix the stale parse_header doc that described a nested-Result return the code no longer uses.
A valid header split partway through log_origin returned 400 instead of pulling the next chunk. read_from special-cased that short read as a distinct LogOriginTruncated variant, so parse_header's single Io(UnexpectedEof) arm missed it. Drop the variant so every short read in the fixed-layout header is a plain Io(UnexpectedEof) that parse_header treats as "need more bytes", 400ing only once the stream ends. Adds streaming parse_header tests for byte-by-byte delivery, a split inside log_origin, and genuine truncation.
GzipInflater::push wrote whole compressed chunks into the decoder and took the entire accumulated plaintext at once, so a small but highly compressible chunk could inflate into a Vec far larger than the isolate's memory ceiling. Feed the decoder in INFLATE_STEP slices, drain the sink per slice, and emit plaintext in <=INFLATE_STEP runs that gunzip queues and yields one per poll. Adds a zip-bomb test.
commit_packages bounds only the number of buffered packages, but a single package can hold 256 entries of up to 65535 bytes (~16 MiB), so a count-only cap could buffer far past the isolate's memory ceiling. Add max_chunk_bytes (default 16 MiB) and flush early once buffered entry bytes reach it, whichever cap trips first.
parse_next_package rebuilt a Cursor at byte zero and re-ran EntryPackage::read_from on every pull, re-parsing and re-allocating all already-read entries; a large package arriving in small chunks was O(n^2) in copies. Parse entry-by-entry instead, consuming each entry and the proof from the StreamBuffer as soon as it is fully buffered, so a short read resumes from the next unread unit. Adds tests for byte-by-byte reassembly, back-to-back packages, both truncation classes, and the num_hashes limit.
INFLATE_STEP capped the compressed slice handed to write_all, but write_all inflates its whole slice into the decoder sink before returning, so a 64 KiB slice could produce ~64 MiB of plaintext at DEFLATE's ~1032x. Shrink the input step to 4 KiB, capping transient plaintext at ~4 MiB, and stop feeding once a step produces output, parking the rest of the chunk for the next poll. Downstream runs stay capped by a separate EMIT_STEP.
e7a7b23 to
53a322f
Compare
2 participants
Replace the whole-body buffer with a streaming decode: gunzip and parse the body as it arrives through a small stream-buffer adapter (retry the parse as chunks land, consume on success), so a large upload no longer holds the entire body in memory.
Persist incrementally every
commit_packagespackages (default 32, config capped at 1024) so a long or interrupted upload advances the frontier as it streams rather than only at the end. Each flush commits from the last persisted frontier and is resumable; the running frontier is threaded locally across flushes so the DO is not re-queried between chunks. Tiles are immutable and content-addressed and the DO advance is a monotone compare-and-swap, so a repeated or concurrent flush of the same range is harmless.