Implement tryReadSync/tryWriteSync support - #7036
Conversation
|
Model not found: cloudflare-ai-gateway/anthropic/claude-opus-4-6. Did you mean: anthropic/claude-opus-4.5, anthropic/claude-opus-4.6, anthropic/claude-opus-4.7? |
|
@jasnell Bonk workflow failed. Check the logs for details. View workflow run · To retry, trigger Bonk again. |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #7036 +/- ##
==========================================
- Coverage 67.80% 67.76% -0.04%
==========================================
Files 468 471 +3
Lines 132283 132805 +522
Branches 21474 21578 +104
==========================================
+ Hits 89691 89994 +303
- Misses 29520 29676 +156
- Partials 13072 13135 +63 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
ae6f334 to
30cf740
Compare
This comment was marked as low quality.
This comment was marked as low quality.
guybedford
left a comment
guybedford
left a comment
There was a problem hiding this comment.
Findings from a full read-through (inline comments below):
- [HIGH] Queued sync write path leaks write-buffer accounting (
adjustWriteBufferSize/onChunkDequeuednever balanced) — permanent backpressure with a highWaterMark, observer metric skew. - [HIGH]
Pipe::writefast path doesn't handle a sync throw fromtryWriteSync(), ending in a different terminal state than an async write rejection. - [MEDIUM]
write()fast path forwards zero-length writes to the sink; the queued path deliberately never does. - [BLOCKING] capnp pin has an empty
sha256and points at the unmerged capnproto PR head (presumably the internal-build failure). - [QUESTION]
DEFAULT_AUTO_ALLOCATE_CHUNK_SIZE_216K→32K bump bundled in, mid-autogate-rollout. - [QUESTION] The JS-observable timing changes ship ungated:
reader.read()/writer.write()promises now settle without the event-loop round trip for compression streams, memory-backed bodies, and drainingRead. The identity-transform note in this PR documents exactly this ordering inversion breaking its own backpressure test — that stream was exempted, but the same class of user-observable reordering now applies everywhere else. Should the controller-level fast paths (the JS-visible ones, as opposed to the pure C++ pump paths) sit behind an autogate for rollout, matching how UPDATED_AUTO_ALLOCATE_CHUNK_SIZE is being rolled out? - [LOW] Redundant try/catch in
pumpToImpl.
The wrapper forwarding (tee, neuterable, abortable, encoded, container-client) all respect the decline/no-side-effect contract, and the ResponseStreamWrapper lock-recursion note is important. Test coverage note: none of the new tests exercise the queued sync write path in writeLoopAfterFrontOutputLock, which is where the accounting bug lives.
This review was written with AI assistance and may contain mistakes; treat each finding on its merits.
| if (syncSuccess) { | ||
| maybeResolvePromise(js, request.promise); | ||
| queue.pop_front(); | ||
|
|
||
| if (maybeAbort(js, *this)) return js.resolvedPromise(); | ||
|
|
||
| return writeLoop(js, ioContext, syncDepth + 1); | ||
| } |
There was a problem hiding this comment.
The enqueue side charged adjustWriteBufferSize(js, len) + onChunkEnqueued(len) in write(), and both async completion continuations balance with adjustWriteBufferSize(js, -amountToWrite) + onChunkDequeued(amountToWrite) before popping. This sync-success path pops without either, and drain()/doError() never reset currentWriteBufferSize, so every queued write completed synchronously leaks amountToWrite into the accounting. With a highWaterMark set that becomes permanent backpressure (writer.ready is replaced and never resolves; desiredSize skews negative), and the StreamObserver enqueue/dequeue metrics diverge.
| if (syncSuccess) { | |
| maybeResolvePromise(js, request.promise); | |
| queue.pop_front(); | |
| if (maybeAbort(js, *this)) return js.resolvedPromise(); | |
| return writeLoop(js, ioContext, syncDepth + 1); | |
| } | |
| if (syncSuccess) { | |
| maybeResolvePromise(js, request.promise); | |
| adjustWriteBufferSize(js, -amountToWrite); | |
| KJ_IF_SOME(o, observer) { | |
| o->onChunkDequeued(amountToWrite); | |
| } | |
| queue.pop_front(); | |
| if (maybeAbort(js, *this)) return js.resolvedPromise(); | |
| return writeLoop(js, ioContext, syncDepth + 1); | |
| } |
The sync KJ_CATCH path above has the same omission relative to the async error continuation, which also does -amountToWrite + onChunkDequeued before rejecting.
| // Fast path: complete the write synchronously when the sink can accept data immediately. | ||
| auto syncResult = KJ_ASSERT_NONNULL(parent.state.whenActive( | ||
| [&](IoOwn<Writable>& writable) { return writable->sink->tryWriteSync(data); })); | ||
| if (syncResult) { | ||
| return js.resolvedPromise(); | ||
| } |
There was a problem hiding this comment.
This is the one tryWriteSync() call site that doesn't map a synchronous throw onto the async-rejection handling. A throw here propagates out of the read continuation in pipeLoop(), skipping the write-rejection functor (tryErrorParent → destination doError, sink abort on the next loop iteration) and instead landing in handlePromise's error functor — which errors the source and rejects the pipe promise but leaves the destination controller un-errored and the sink un-aborted. Different terminal state than an async write failure.
Surfacing the throw as a rejected promise routes it through the exact same continuation as an async rejection:
| // Fast path: complete the write synchronously when the sink can accept data immediately. | |
| auto syncResult = KJ_ASSERT_NONNULL(parent.state.whenActive( | |
| [&](IoOwn<Writable>& writable) { return writable->sink->tryWriteSync(data); })); | |
| if (syncResult) { | |
| return js.resolvedPromise(); | |
| } | |
| // Fast path: complete the write synchronously when the sink can accept data immediately. | |
| bool syncSuccess = false; | |
| KJ_TRY { | |
| syncSuccess = KJ_ASSERT_NONNULL(parent.state.whenActive( | |
| [&](IoOwn<Writable>& writable) { return writable->sink->tryWriteSync(data); })); | |
| } | |
| KJ_CATCH(exception) { | |
| // A sync throw is equivalent to a rejected write() promise; surface it as one so the | |
| // caller's rejection continuation handles it identically to an async write failure. | |
| return js.rejectedPromise<void>(js.exceptionToJs(kj::mv(exception))); | |
| } | |
| if (syncSuccess) { | |
| return js.resolvedPromise(); | |
| } |
| // | ||
| // TODO(perf): Consider adding a synchronous "output gate is open" check so that actors | ||
| // can also take this fast path when no storage writes are pending. | ||
| if (queue.empty() && maybePendingAbort == kj::none && |
There was a problem hiding this comment.
Zero-length writes reach the sink through this fast path: processChunk returns a non-none empty array for empty strings, so tryWriteSync(empty) gets invoked. The queued path deliberately never forwards zero-length writes to the sink (see the note in writeLoopAfterFrontOutputLock about distinguishing disconnections from zero-length reads on the other end of a TransformStream). Excluding them here preserves the queued no-op semantics:
| if (queue.empty() && maybePendingAbort == kj::none && | |
| if (len > 0 && queue.empty() && maybePendingAbort == kj::none && |
| name = "capnp-cpp", | ||
| sha256 = "6753378bd099029cb2830fecd32dd158218019e459ffd3c8e379cbf025906eb8", | ||
| strip_prefix = "capnproto-capnproto-a1cd1c4/c++", | ||
| sha256 = "", |
There was a problem hiding this comment.
Blocking: empty sha256, and the tarball points at the head of the unmerged capnproto PR (capnproto/capnproto#2740). Needs repinning to the merged capnp commit with the real hash before this can land — presumably also the cause of the internal-build failure.
| // so carefully to avoid introducing memory regressions and causing workers to | ||
| // hit OOM errors. We'll use an autogate to roll out the new default. | ||
| static constexpr int DEFAULT_AUTO_ALLOCATE_CHUNK_SIZE_2 = 16 * 1024; | ||
| static constexpr int DEFAULT_AUTO_ALLOCATE_CHUNK_SIZE_2 = 32 * 1024; |
There was a problem hiding this comment.
This 16 KiB → 32 KiB bump looks unrelated to tryReadSync/tryWriteSync and isn't mentioned in the PR description. It's also the value behind the in-flight UPDATED_AUTO_ALLOCATE_CHUNK_SIZE autogate — if that gate is partially rolled out, this silently doubles the allocation mid-rollout. Intentional? If so it seems worth its own PR with the memory-regression reasoning the comment above alludes to.
| bool syncSuccess = false; | ||
| KJ_TRY { | ||
| syncSuccess = sink->tryWriteSync(pieces); | ||
| } | ||
| KJ_CATCH(exception) { | ||
| // tryWriteSync() may throw when a synchronous write is possible but fails. Per | ||
| // the tryWriteSync() contract this is equivalent to a rejected write() promise. | ||
| writeFailed = true; | ||
| kj::throwFatalException(kj::mv(exception)); | ||
| } | ||
| if (!syncSuccess) { | ||
| co_await sink->write(pieces); | ||
| } |
There was a problem hiding this comment.
The KJ_TRY/KJ_CATCH is redundant: KJ_ON_SCOPE_FAILURE(writeFailed = true) two lines up already covers a synchronous throw from tryWriteSync(), so a bare call behaves identically.
| bool syncSuccess = false; | |
| KJ_TRY { | |
| syncSuccess = sink->tryWriteSync(pieces); | |
| } | |
| KJ_CATCH(exception) { | |
| // tryWriteSync() may throw when a synchronous write is possible but fails. Per | |
| // the tryWriteSync() contract this is equivalent to a rejected write() promise. | |
| writeFailed = true; | |
| kj::throwFatalException(kj::mv(exception)); | |
| } | |
| if (!syncSuccess) { | |
| co_await sink->write(pieces); | |
| } | |
| if (!sink->tryWriteSync(pieces)) { | |
| co_await sink->write(pieces); | |
| } |
3 participants
capnp PR: capnproto/capnproto#2740