feat(streams): R2 chunk log — segment write-ahead log behind new Streams({ r2 }) - #2215
Closed
mattzcarey wants to merge 2 commits into
Closed
feat(streams): R2 chunk log — segment write-ahead log behind new Streams({ r2 })#2215mattzcarey wants to merge 2 commits into
new Streams({ r2 })#2215mattzcarey wants to merge 2 commits into
Conversation
…ams({ r2 })
Stream rows (state, tag index, metadata, cursor) stay in DO SQLite; chunks
go to R2 as immutable segment objects checkpointed every 25 chunks or 1 s
(r2Checkpoint), so a Durable Object that dies mid-stream leaves everything
up to its last checkpoint in R2 and a resumed producer continues from the
durable cursor in a new epoch. Settlement compacts the segments into one
exact-size object; Streams.flush() awaits it. The synchronous storage
aperture used by chat stays SQLite-only.
🦋 Changeset detectedLatest commit: f2ac0ff The changes in this PR will be included in the next version bump. This PR includes changesets to release 2 packages
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Note
This report is out of date. Scroll down for Devin Review's latest report on this PR.
Devin Review found 5 potential issues.
2 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
Comment on lines
+334
to
+336
| const rows = this.#r2 | ||
| ? await this.#r2.readPage(streamId, next, batchSize) | ||
| : this.#sql<StreamChunkRow>` |
Contributor
There was a problem hiding this comment.
🔴 Backend changes erase existing histories
When R2 is enabled after SQLite streams exist, readPage ignores their stored chunks. Settled histories become empty, and live writers resume at zero.
Prompt for agents
Add an explicit per-stream backend marker and migration strategy in packages/agents/src/streams/streams.ts. Existing rows may own chunks in cf_agents_stream_chunks when an installation first enables r2, while rows created under R2 may outlive a later configuration change. Reads, resume, settlement, and deletion must select the backend that owns each stream. Either migrate existing chunk logs safely before switching the marker or preserve backend-specific reads until migration completes.
Was this helpful? React with 👍 or 👎 to provide feedback.
Comment on lines
635
to
+638
| if (settled > 0) { | ||
| // The row is terminal now; the R2 file and WAL cleanup finish in the | ||
| // background (`flush()` awaits them) while memory keeps serving reads. | ||
| void this.#r2?.settle(streamId); |
Contributor
There was a problem hiding this comment.
🔴 Closing can discard final chunks
settle starts R2 persistence after marking the stream terminal. An isolate death before the put completes permanently drops the final chunks.
Prompt for agents
Redesign R2 settlement so a stream cannot become irreversibly terminal before its complete chunk log is durable. The current synchronous Streams writer API updates cf_agents_streams first, then starts R2ChunkLog.settle in an unawaited promise. A restart can therefore see a terminal row with neither the final segment nor body available and has no recovery path. Introduce a durable settling phase or equivalent recovery metadata, ensure background work is attached to the invocation lifetime where possible, and make restart finish or resume settlement before exposing a terminal result.
Was this helpful? React with 👍 or 👎 to provide feedback.
Comment on lines
+334
to
+336
| const rows = this.#r2 | ||
| ? await this.#r2.readPage(streamId, next, batchSize) | ||
| : this.#sql<StreamChunkRow>` |
Contributor
There was a problem hiding this comment.
🟡 Live readers can miss wakeups
The awaited readPage breaks atomic poll-and-wait registration. An append during that gap leaves a live reader asleep despite available data.
Prompt for agents
Restore a lost-wakeup-safe tailing protocol for asynchronous R2 reads. The SQLite implementation relied on polling, checking state, and registering the waiter in one synchronous block. R2ChunkLog.readPage introduces a suspension before waiter registration. Register a generation-aware waiter before the asynchronous poll, or recheck a monotonic in-memory generation after registration, while preserving abort cleanup and settlement handling.
Was this helpful? React with 👍 or 👎 to provide feedback.
Contributor
🔴 agents import sizesMeasured 267 runtime imports as minified bundles. The primary size is gzip; raw minified size is included for diagnosis. An existing import growing by more than 10% is marked red. This report is informational.
Compared Changed imports (102)
All 267 current runtime imports
Reported by agent-think[bot]. |
…d stamp ops - Evict landed lines from memory beyond a 256 KB hot window; readers behind it read segments from R2. resume() lists once and points at the chain without loading it. Settle streams segments back through one exact-size put, so a stream's length never grows isolate memory. - Segment puts retry with backoff; a failed range folds into the next checkpoint and the chain picker prefers the wider key, so nothing is skipped. Cursor stamps are throttled to one row write per 5 s and re-stamped exactly on resume. - No list at settle (landed keys are tracked). Large settled bodies replay through a line-offset index with ranged gets; small ones are cached whole in an 8 MiB LRU.
Contributor
There was a problem hiding this comment.
Devin Review found 1 new potential issue.
4 flags not posted on this PR by your GitHub settings — view them in Devin Review. (Configure)
Comment on lines
+245
to
+257
| async resume(streamId: string): Promise<number> { | ||
| const existing = this.#sinks.get(streamId); | ||
| if (existing) return existing.cursor; | ||
| this.#uncache(streamId); | ||
| const segments = await this.#listSegments(streamId); | ||
| const chain = contiguousChain(segments); | ||
| const covered = new Set(chain.map((s) => s.key)); | ||
| await this.#deleteKeys( | ||
| segments.filter((s) => !covered.has(s.key)).map((s) => s.key) | ||
| ); | ||
| const epoch = segments.reduce((max, s) => Math.max(max, s.epoch), -1) + 1; | ||
| const sink = new Sink(epoch, chain); | ||
| this.#sinks.set(streamId, sink); |
Contributor
There was a problem hiding this comment.
🔴 Concurrent recovery discards new chunks
Concurrent open() calls can both enter resume after restart. The later recovery replaces the live sink and discards chunks appended after the first.
Prompt for agents
R2ChunkLog.resume checks #sinks only before awaiting R2 list/delete operations. Concurrent open() calls can therefore build separate sinks and the later completion overwrites the first. Serialize recovery per stream or share an in-flight resume promise, then re-check the live sink before installing recovered state. Ensure deletion of uncovered segments also occurs only once and cannot race with newly appended checkpoints.
Was this helpful? React with 👍 or 👎 to provide feedback.
Contributor
Author
|
Superseded by #2216: the R2 segment log was measured against the SQLite alternatives and lost on cost (a Class A put is 4.5 SQLite row writes; the row-write cost was the cleanup, not the append). #2216 keeps stream blocks in the same DO storage as the session message and makes the handoff one transaction. The R2 measurements and the spike stay on this branch for reference. |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What
agents/streamsgets a second chunk log. HandStreamsan R2 binding and chunks leave DO SQLite:Nothing else changes:
open,append,read,status,list,deletebehave as before andappend()stays synchronous. Stream rows (state, tag index, metadata, cursor) stay in SQLite, where a point read or tag lookup costs nothing. Chunks go to the bucket.Why
Cost. SQLite bills one row per stored chunk and again to delete it. R2 bills one Class A op per checkpoint, deletes are free, and storage is about 13× cheaper with no 10 GB ceiling. One R2 put costs the same as 4.5 SQLite row writes, so the R2 log is cheaper whenever a checkpoint covers more than ~4.5 stored rows: unpacked producers, or a loss window of a few seconds. Against chat's 10-to-1 packed SQLite at a 1 s cadence it is a wash on ops and wins only on storage. The docs say this in the same words.
Per 400-chunk chat turn: $0.0008 SQLite unpacked, $0.00008 packed, $0.00009 R2 at the default cadence, $0.00002 at 5 s.
How
R2 has no append, rejects bodies of unknown length, and stores nothing from a put that has not completed (all measured on real R2, see below). So the log is a write-ahead log of segment objects:
everyChunksappends oreveryMsafter the first unflushed one, the new lines are put as one immutable segment<prefix><id>/seg/<epoch>/<from>-<to>. Each landed segment is the durability: when the isolate dies, everything up to the last landed segment is in R2. Puts are chained so they land in order, retried with backoff, and a range that still fails folds into the next checkpoint so nothing is skipped.status()never reports more than R2 holds and the stamp costs a fraction of the puts.open()after a death re-stamps it exactly.open()on a stream whose isolate died lists the segments once, keeps the contiguous chain, deletes keys it does not cover, and continues in a new epoch from the chain's end without loading it into memory. A resumed producer starts at the durable cursor, so the Tasks resume contract holds and a discarded generation can never be spliced back in.close()/error()settle the row synchronously, then in the background stream the segments back through one exact-size put at<id>/bodyand drop them (keys are tracked, no list; deletes batched at 1000). Segments stay readable until the body lands.await streams.flush(id)waits for it.SQLite accounting on the R2 backend: 100 appends read and write 0 rows; the cursor stamp is 1 row write per 5 s of streaming;
status()reads 1, tag lookup 2, settle reads 2 and writes 2.Chat's
ResumableStreamuses the synchronous SQLite aperture and is unaffected; it throws if given an R2-backedStreams.Measured on real R2 (2026-09-03)
put(key, unknownLengthStream)FixedLengthStreamput, slow writeslist/headuntil completecompleterejects (10011, below 5 MiB minimum)Those are why there is no streaming file put (it never contributes durability) and why the rows stay in SQLite.
Tests
src/tests/streams/r2-backend.test.ts(miniflare R2 bindingSTREAMS_R2): cursor + live tail + body at close; no chunk rows in SQLite;error()path; resume from the segment chain after a simulated restart; replay before recovery never pins a truncated log; delete; SQLite accounting; aperture refusal.r2-backend.test.tsalso covers memory: 2000 × 500 B chunks keep under half the lines in memory while a reader far behind is served from segments, and the 1 MB settled body replays from any cursor through the offset index.sseResponsewhile streaming, body landed, segments cleaned;ctx.abort()at chunk 120 left five segments and a durable cursor of 100, resume continued from 100, final body 200 lines. A 4000 × 500 B stream (2.2 MB) settled through the streamed compaction with no segments left, and the same stream killed at chunk 925 resumed from 925 and settled to 4000 lines.Not in this PR
r2Checkpointis capability-wide.