From 447cf4d0ddc81165fb20e9c2eb42c1d49d86d3ea Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 14 Oct 2026 05:34:30 +0000 Subject: [PATCH 1/3] Stripe CLI transfers across relay and direct paths; fix relay deadline Profiling showed the direct WebRTC path is CPU-bound in pion's SCTP stack (about 30 MiB/s on loopback) while the relay WebSocket alone moves about 120 MiB/s, yet once the direct path opened the relay sat idle. Senders now keep both busy: each path's writer goroutine takes the next data record when it has room, and a perturb-and-observe controller tunes the relay's share of the in-flight budget toward the highest acknowledged rate. Relay records stay held until acknowledged and go out again on the direct path if the relay fails. Receivers already merge both paths by sequence number and drop second copies, so no protocol change is needed. CD_STRIPE=0 restores the previous behavior. Also clear the read deadline waitRelayEvent leaves on the WebSocket (10s on a receiver): the link's relay reader inherited it, so a receive that stayed on the relay for more than ~10s died with "transfer stalled". Local bench, 256 MiB CLI -> CLI: 28.8 -> 54.4 MiB/s (median of 3). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012g5D2t4GGkGV1pn3Z2EcC4 --- cmd/cdx/direct.go | 4 +- cmd/cdx/link.go | 34 ++++- cmd/cdx/sender.go | 1 + cmd/cdx/stripe.go | 309 +++++++++++++++++++++++++++++++++++++++++ cmd/cdx/stripe_test.go | 195 ++++++++++++++++++++++++++ 5 files changed, 540 insertions(+), 3 deletions(-) create mode 100644 cmd/cdx/stripe.go create mode 100644 cmd/cdx/stripe_test.go diff --git a/cmd/cdx/direct.go b/cmd/cdx/direct.go index 034ce60..6c4f593 100644 --- a/cmd/cdx/direct.go +++ b/cmd/cdx/direct.go @@ -44,6 +44,8 @@ const ( maxSDPBytes = 64 * 1024 ) +var errDirectStalled = errors.New("direct connection stalled") + // directICEServers is a variable so tests can run without STUN. var directICEServers = []webrtc.ICEServer{{URLs: []string{"stun:stun.l.google.com:19302", "stun:global.stun.twilio.com:3478"}}} @@ -230,7 +232,7 @@ func (path *directPath) send(record []byte) error { case <-path.lost: return errors.New("direct connection lost") case <-time.After(transferIdle): - return errors.New("direct connection stalled") + return errDirectStalled } } for start := 0; start < len(record); start += directPieceSize { diff --git a/cmd/cdx/link.go b/cmd/cdx/link.go index acd14cd..b0fc217 100644 --- a/cmd/cdx/link.go +++ b/cmd/cdx/link.go @@ -44,6 +44,13 @@ type recordLink struct { // drops whichever copy arrives second. retained []retainedRecord switched bool + // striping, set by senders, keeps the relay carrying data records + // alongside the direct path once it opens (see stripe.go); stripe is + // its state from then on. + striping bool + stripe *striper + // ackedBytes is the peer's last acknowledged byte count. + ackedBytes uint64 } type retainedRecord struct { @@ -89,6 +96,10 @@ func newRecordLink(connection *websocket.Conn, sealer *recordSealer, opener *rec // start begins reading the relay WebSocket; call it once the caller no // longer reads the socket itself. func (link *recordLink) start() { + // Clear the deadline waitRelayEvent left behind (10s on a receiver): + // next and poll time out on their own, and a deadline here would end + // the relay reader mid-transfer. + _ = link.connection.SetReadDeadline(time.Time{}) go link.readRelay() go link.pingRelay() } @@ -174,18 +185,30 @@ func (link *recordLink) sendReleasable(kind messageKind, payload []byte, release if err != nil { return err } + if link.stripe != nil { + // The writer goroutines own both paths now. + return link.sendReleasableStriped(record, release) + } path := link.direct.Load() if path != nil && path.isOpen() { link.directUsed.Store(true) if !link.switched { link.switched = true + if link.striping { + link.startStriping(path) + } for _, earlier := range link.retained { - if err := path.send(earlier.record); err != nil { + if link.stripe != nil { + link.enqueueControl(earlier.record) + } else if err := path.send(earlier.record); err != nil { return err } } link.retained = nil } + if link.stripe != nil { + return link.sendReleasableStriped(record, release) + } return path.send(record) } if path != nil && !link.switched && !path.isLost() { @@ -210,6 +233,10 @@ func (link *recordLink) acknowledged(bytes uint64) { } } link.retained = kept + if link.stripe != nil { + link.stripe.acknowledged(bytes, link.ackedBytes, time.Now()) + } + link.ackedBytes = bytes } // next returns the next record in sequence order, waiting up to timeout. @@ -274,7 +301,10 @@ func (link *recordLink) accept(item linkItem) error { return nil // an unused direct path failing changes nothing } if !item.direct && link.directUsed.Load() { - return nil // the relay is optional once the direct path carries the transfer + // The relay is optional once the direct path carries the + // transfer; records striped onto it go out again there. + link.relayFailed() + return nil } return item.err } diff --git a/cmd/cdx/sender.go b/cmd/cdx/sender.go index 2602ddc..7dad8af 100644 --- a/cmd/cdx/sender.go +++ b/cmd/cdx/sender.go @@ -350,6 +350,7 @@ func sendFile(ctx context.Context, paths []string, linkMode bool, hooks sendHook return err } link := newRecordLink(connection, sealer, opener, errors.New("receiver disconnected (they closed the tab or lost network)")) + link.striping = stripeEnabled() defer link.close() // Gather direct-path candidates while waiting for the receiver. var offerSDP chan string diff --git a/cmd/cdx/stripe.go b/cmd/cdx/stripe.go new file mode 100644 index 0000000..5192cdd --- /dev/null +++ b/cmd/cdx/stripe.go @@ -0,0 +1,309 @@ +package main + +import ( + "math" + "os" + "sync" + "time" + + "github.com/gorilla/websocket" +) + +// Striping: once the direct path opens, a sender keeps the relay carrying +// data too, so the two paths' throughput adds up. Each path has a writer +// goroutine that takes the next data record whenever it has room, so the +// faster path naturally takes more. Receivers already merge both paths in +// sequence order and drop second copies, so this needs nothing new from +// them. +// +// Room is a per-path cap on bytes in flight, set from one number: the +// relay's share of the sender's in-flight budget. The share is tuned by +// perturb and observe: every epoch it moves a step, and when the delivery +// rate the receiver acknowledges fell, the next step goes the other way. It +// settles where the two paths deliver the most together, whatever limits +// them (a slow relay, a slow direct path, or CPU when both share it). +// +// Relay records stay held until acknowledged: if the relay fails they go +// out again on the direct path. Losing the direct path ends the transfer, as +// before. + +// stripeEnabled lets tests and benchmarks keep data on the direct path +// alone (CD_STRIPE=0). +func stripeEnabled() bool { + return os.Getenv("CD_STRIPE") != "0" +} + +const ( + // stripeEpoch is how long each share is held before it is judged. + stripeEpoch = 200 * time.Millisecond + // stripeShareStep is how far the relay's share moves per epoch. + stripeShareStep = 0.125 + // stripeStartShare splits the budget evenly at first. + stripeStartShare = 0.5 + // stripeBudget is the in-flight budget the share divides. Relay + // records sent before the switch may still sit in the relay's 32 MiB + // per-peer buffer, so the relay's part stays at half of it; the direct + // path's stays within its usual high-water mark. Each keeps at least + // one record so its delivery rate stays known. + stripeBudget = 16 * 1024 * 1024 + stripeMinCap = chunkSize + 64 + stripePoll = 2 * time.Millisecond + // stripeQueueRecords bounds the queues; the ack window keeps them far + // below it. + stripeQueueRecords = 512 +) + +type stripeItem struct { + record []byte + release uint64 +} + +type striper struct { + // data holds data records for whichever path is ready first; control + // holds control records and re-sends, which take the direct path. + data chan stripeItem + control chan []byte + // relayRoom wakes the relay writer when an ack frees room. + relayRoom chan struct{} + + mu sync.Mutex + relayCap uint64 + directCap uint64 + // share is the relay's part of stripeBudget; step is the next move. + share float64 + step float64 + // epochStart, epochBytes, and lastRate measure each epoch's delivery. + epochStart time.Time + epochBytes uint64 + lastRate float64 + // relayUnacked keeps relay records (to re-send) and relayBytes their + // size; directReleases is the release offset of each direct record. + relayUnacked []stripeItem + relayBytes uint64 + directReleases []uint64 + relayDead bool +} + +func newStriper() *striper { + s := &striper{ + data: make(chan stripeItem, stripeQueueRecords), + control: make(chan []byte, stripeQueueRecords), + relayRoom: make(chan struct{}, 1), + step: stripeShareStep, + } + s.setShare(stripeStartShare) + return s +} + +// setShare sets the relay's share and both caps from it. +func (s *striper) setShare(share float64) { + s.share = min(max(share, 0), 1) + s.relayCap = max(uint64(s.share*stripeBudget), stripeMinCap) + s.directCap = min(max(uint64((1-s.share)*stripeBudget), stripeMinCap), directBufferHigh) +} + +// acknowledged drops records the peer confirmed and, once per epoch, +// moves the relay's share. +func (s *striper) acknowledged(bytes uint64, previous uint64, at time.Time) { + s.mu.Lock() + covered := 0 + for covered < len(s.relayUnacked) && s.relayUnacked[covered].release <= bytes { + s.relayBytes -= uint64(len(s.relayUnacked[covered].record)) + covered++ + } + clear(s.relayUnacked[:covered]) + s.relayUnacked = s.relayUnacked[covered:] + covered = 0 + for covered < len(s.directReleases) && s.directReleases[covered] <= bytes { + covered++ + } + s.directReleases = s.directReleases[covered:] + if !s.relayDead { + s.adjust(bytes-min(previous, bytes), at) + } + s.mu.Unlock() + select { + case s.relayRoom <- struct{}{}: + default: + } +} + +// adjust counts delivered bytes and, at the end of each epoch, keeps the +// share moving the same way if delivery improved and turns it otherwise. +func (s *striper) adjust(delivered uint64, at time.Time) { + if s.epochStart.IsZero() { + s.epochStart = at + return + } + s.epochBytes += delivered + span := at.Sub(s.epochStart) + if span < stripeEpoch { + return + } + rate := float64(s.epochBytes) / span.Seconds() + if rate < s.lastRate { + s.step = -s.step + } + s.lastRate = rate + s.epochStart, s.epochBytes = at, 0 + s.setShare(s.share + s.step) +} + +// failRelay stops using the relay and returns its unacknowledged records +// for the direct path; nil if the relay had already failed. +func (s *striper) failRelay() [][]byte { + s.mu.Lock() + defer s.mu.Unlock() + if s.relayDead { + return nil + } + s.relayDead = true + s.directCap = directBufferHigh + records := make([][]byte, 0, len(s.relayUnacked)) + for _, item := range s.relayUnacked { + records = append(records, item.record) + } + s.relayUnacked, s.relayBytes = nil, 0 + return records +} + +// startStriping hands both paths to writer goroutines. Call with writeMu +// held, when the direct path first opens. +func (link *recordLink) startStriping(path *directPath) { + link.stripe = newStriper() + go link.writeDirect(path) + go link.writeRelay() +} + +// sendReleasableStriped queues a sealed record: data records for either +// path, control records (math.MaxUint64 release) for the direct path. +func (link *recordLink) sendReleasableStriped(record []byte, release uint64) error { + if release == math.MaxUint64 { + link.enqueueControl(record) + } else { + link.enqueueData(record, release) + } + return nil +} + +func (link *recordLink) enqueueControl(record []byte) { + select { + case link.stripe.control <- record: + case <-link.stop: + } +} + +func (link *recordLink) enqueueData(record []byte, release uint64) { + select { + case link.stripe.data <- stripeItem{record: record, release: release}: + case <-link.stop: + } +} + +// directHasRoom reports whether the data channel's buffer is under the +// direct path's cap. +func (link *recordLink) directHasRoom(path *directPath) bool { + link.stripe.mu.Lock() + limit := link.stripe.directCap + link.stripe.mu.Unlock() + return path.channel.BufferedAmount() < limit +} + +func (link *recordLink) writeDirect(path *directPath) { + s := link.stripe + idleSince := time.Now() + for { + var record []byte + select { + case <-link.stop: + return + case <-path.lost: + return // directLost already reported it + case record = <-s.control: + default: + if !link.directHasRoom(path) { + if time.Since(idleSince) > transferIdle { + link.deliver(linkItem{err: errDirectStalled, direct: true}) + return + } + select { + case <-link.stop: + return + case <-path.drained: + case <-time.After(stripePoll): + } + continue + } + select { + case <-link.stop: + return + case <-path.lost: + return + case record = <-s.control: + case item := <-s.data: + record = item.record + s.mu.Lock() + s.directReleases = append(s.directReleases, item.release) + s.mu.Unlock() + } + } + idleSince = time.Now() + if err := path.send(record); err != nil { + link.deliver(linkItem{err: err, direct: true}) + return + } + } +} + +func (link *recordLink) writeRelay() { + s := link.stripe + for { + s.mu.Lock() + dead, room := s.relayDead, s.relayBytes < s.relayCap + s.mu.Unlock() + if dead { + return + } + if !room { + select { + case <-link.stop: + return + case <-s.relayRoom: + } + continue + } + var item stripeItem + select { + case <-link.stop: + return + case item = <-s.data: + } + s.mu.Lock() + if s.relayDead { + s.mu.Unlock() + link.enqueueControl(item.record) + return + } + s.relayUnacked = append(s.relayUnacked, item) + s.relayBytes += uint64(len(item.record)) + s.mu.Unlock() + err := link.connection.SetWriteDeadline(time.Now().Add(transferIdle)) + if err == nil { + err = link.connection.WriteMessage(websocket.BinaryMessage, item.record) + } + if err != nil { + link.relayFailed() + return + } + } +} + +// relayFailed moves the relay's unacknowledged records to the direct path. +func (link *recordLink) relayFailed() { + if link.stripe == nil { + return + } + for _, record := range link.stripe.failRelay() { + link.enqueueControl(record) + } +} diff --git a/cmd/cdx/stripe_test.go b/cmd/cdx/stripe_test.go new file mode 100644 index 0000000..f863cb5 --- /dev/null +++ b/cmd/cdx/stripe_test.go @@ -0,0 +1,195 @@ +package main + +import ( + "reflect" + "testing" + "time" +) + +func TestStriperStartsAtHalfShare(t *testing.T) { + s := newStriper() + if s.share != stripeStartShare { + t.Fatalf("share = %v, want %v", s.share, stripeStartShare) + } + if want := uint64(stripeBudget / 2); s.relayCap != want { + t.Fatalf("relayCap = %d, want %d", s.relayCap, want) + } + if want := uint64(stripeBudget / 2); s.directCap != want { + t.Fatalf("directCap = %d, want %d", s.directCap, want) + } +} + +func TestStriperSetShareClampsToUnitInterval(t *testing.T) { + s := newStriper() + s.setShare(-3) + if s.share != 0 { + t.Fatalf("share after setShare(-3) = %v, want 0", s.share) + } + s.setShare(7) + if s.share != 1 { + t.Fatalf("share after setShare(7) = %v, want 1", s.share) + } + s.setShare(0.25) + if s.share != 0.25 { + t.Fatalf("share after setShare(0.25) = %v, want 0.25", s.share) + } +} + +func TestStriperCapsStayWithinBounds(t *testing.T) { + for _, share := range []float64{-1, 0, 0.1, 0.5, 0.9, 1, 2} { + s := newStriper() + s.setShare(share) + if s.relayCap < stripeMinCap { + t.Errorf("share %v: relayCap = %d, below stripeMinCap %d", share, s.relayCap, stripeMinCap) + } + if s.directCap > directBufferHigh { + t.Errorf("share %v: directCap = %d, above directBufferHigh %d", share, s.directCap, directBufferHigh) + } + if s.directCap < stripeMinCap { + t.Errorf("share %v: directCap = %d, below stripeMinCap %d", share, s.directCap, stripeMinCap) + } + } +} + +func TestStriperAcknowledgedPrunesRelayRecords(t *testing.T) { + s := newStriper() + s.relayUnacked = []stripeItem{ + {record: make([]byte, 100), release: 100}, + {record: make([]byte, 100), release: 200}, + {record: make([]byte, 100), release: 300}, + } + s.relayBytes = 300 + + s.acknowledged(200, 0, time.Unix(0, 0)) + + if len(s.relayUnacked) != 1 || s.relayUnacked[0].release != 300 { + t.Fatalf("relayUnacked releases = %v, want [300]", releasesOf(s.relayUnacked)) + } + if s.relayBytes != 100 { + t.Fatalf("relayBytes = %d, want 100", s.relayBytes) + } +} + +func TestStriperAcknowledgedPrunesDirectReleases(t *testing.T) { + s := newStriper() + s.directReleases = []uint64{50, 100, 150} + + s.acknowledged(100, 0, time.Unix(0, 0)) + + if want := []uint64{150}; !reflect.DeepEqual(s.directReleases, want) { + t.Fatalf("directReleases = %v, want %v", s.directReleases, want) + } +} + +func TestStriperAdjustStartsEpochThenMovesShare(t *testing.T) { + const mib = 1 << 20 + s := newStriper() + t0 := time.Unix(0, 0) + + // The first call only starts the epoch; its bytes are not counted. + s.acknowledged(1*mib, 0, t0) + if s.share != 0.5 { + t.Fatalf("share after first call = %v, want 0.5", s.share) + } + + // Each step: bytes acknowledged so far, previous acknowledged, time, + // and the share expected afterwards. + steps := []struct { + name string + bytes uint64 + previous uint64 + at time.Time + wantShare float64 + wantStep float64 + }{ + // Epoch 1 closes at 200 ms with 1 MiB: 5 MiB/s, first rate, so up. + {"epoch 1 closes", 2 * mib, 1 * mib, t0.Add(200 * time.Millisecond), 0.625, +stripeShareStep}, + // Inside epoch 2, not yet judged. + {"mid epoch 2", 2*mib + mib/2, 2 * mib, t0.Add(300 * time.Millisecond), 0.625, +stripeShareStep}, + // Epoch 2 closes at 400 ms with 2 MiB total: 10 MiB/s, faster, keep going up. + {"epoch 2 closes faster", 4 * mib, 2*mib + mib/2, t0.Add(400 * time.Millisecond), 0.75, +stripeShareStep}, + // Epoch 3 closes at 600 ms with 0.5 MiB: 2.5 MiB/s, slower, so turn back. + {"epoch 3 closes slower", 4*mib + mib/2, 4 * mib, t0.Add(600 * time.Millisecond), 0.625, -stripeShareStep}, + // Epoch 4 closes at 800 ms with 1 MiB: 5 MiB/s, faster, keep going down. + {"epoch 4 closes faster", 5*mib + mib/2, 4*mib + mib/2, t0.Add(800 * time.Millisecond), 0.5, -stripeShareStep}, + } + for _, step := range steps { + s.acknowledged(step.bytes, step.previous, step.at) + if s.share != step.wantShare { + t.Fatalf("%s: share = %v, want %v", step.name, s.share, step.wantShare) + } + if s.step != step.wantStep { + t.Fatalf("%s: step = %v, want %v", step.name, s.step, step.wantStep) + } + } +} + +func TestStriperFailRelayReturnsUnacknowledgedInOrder(t *testing.T) { + s := newStriper() + s.setShare(1) + s.relayUnacked = []stripeItem{ + {record: []byte("a"), release: 10}, + {record: []byte("bb"), release: 20}, + {record: []byte("ccc"), release: 30}, + } + s.relayBytes = 6 + + got := s.failRelay() + + want := [][]byte{[]byte("a"), []byte("bb"), []byte("ccc")} + if !reflect.DeepEqual(got, want) { + t.Fatalf("failRelay() = %q, want %q", got, want) + } + if !s.relayDead { + t.Fatal("relayDead = false after failRelay") + } + if s.directCap != directBufferHigh { + t.Fatalf("directCap = %d, want directBufferHigh %d", s.directCap, directBufferHigh) + } + if again := s.failRelay(); again != nil { + t.Fatalf("second failRelay() = %q, want nil", again) + } +} + +func TestStriperAcknowledgedAfterRelayFailureKeepsShare(t *testing.T) { + const mib = 1 << 20 + s := newStriper() + s.failRelay() + t0 := time.Unix(0, 0) + + s.acknowledged(1*mib, 0, t0) + s.acknowledged(2*mib, 1*mib, t0.Add(200*time.Millisecond)) + s.acknowledged(4*mib, 2*mib, t0.Add(400*time.Millisecond)) + + if s.share != stripeStartShare { + t.Fatalf("share after relay failure = %v, want unchanged %v", s.share, stripeStartShare) + } +} + +func TestStriperAcknowledgedSignalsRoomWithoutBlocking(t *testing.T) { + s := newStriper() + done := make(chan struct{}) + go func() { + defer close(done) + t0 := time.Unix(0, 0) + for i := range 5 { + s.acknowledged(uint64(i+1)<<20, uint64(i)<<20, t0.Add(time.Duration(i)*stripeEpoch)) + } + }() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("acknowledged blocked on relayRoom") + } + if n := len(s.relayRoom); n != 1 { + t.Fatalf("relayRoom holds %d signals, want 1 pending", n) + } +} + +func releasesOf(items []stripeItem) []uint64 { + var out []uint64 + for _, item := range items { + out = append(out, item.release) + } + return out +} From 2a24ffb5198523a8ad0aa02a7a25985762a42cd6 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 14 Oct 2026 05:34:31 +0000 Subject: [PATCH 2/3] Overlap browser disk writes and skip a copy on the send path The OPFS sink now starts each flush without waiting for it, keeping one write in flight so decrypting the next records overlaps disk I/O; close drains it and a failed write surfaces on the next write or close. rechunk yields pieces that are already chunk-sized as-is instead of copying them. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012g5D2t4GGkGV1pn3Z2EcC4 --- src/sink.js | 22 ++++++- src/sink.test.js | 129 +++++++++++++++++++++++++++++++++++++++++ src/zip-bundle.js | 9 ++- src/zip-bundle.test.js | 41 +++++++++++++ 4 files changed, 198 insertions(+), 3 deletions(-) diff --git a/src/sink.js b/src/sink.js index d9af1e0..65ef6d0 100644 --- a/src/sink.js +++ b/src/sink.js @@ -94,6 +94,12 @@ export async function createOPFSSink({ name }) { const FLUSH_BYTES = 512 * 1024; let pending = []; let pendingBytes = 0; + // At most one writable.write() is outstanding. A flush starts its write + // without waiting for it, so the next chunk's decrypt overlaps disk I/O; + // the next flush (or close) waits for the previous one first, which keeps + // writes in byte order and bounds memory to two flushes. + let inflight = null; + let failure = null; const flush = async () => { if (pendingBytes === 0) return; let merged; @@ -109,7 +115,18 @@ export async function createOPFSSink({ name }) { } pending = []; pendingBytes = 0; - await writable.write(merged); + if (inflight) await inflight; + const next = writable.write(merged); + // Record a failure now so the next write() or close() reports it even + // when no flush is due; the .catch keeps it from being unhandled. + next.catch((error) => { failure ??= error; }); + inflight = next; + }; + // Waits for every queued write, then surfaces any failure. + const drain = async () => { + await flush(); + if (inflight) await inflight; + if (failure) throw failure; }; const discard = async () => { pending = []; @@ -121,6 +138,7 @@ export async function createOPFSSink({ name }) { kind: 'download', internalName: internal, async write(bytes) { + if (failure) throw failure; const view = bytes instanceof Uint8Array ? bytes : ArrayBuffer.isView(bytes) @@ -131,7 +149,7 @@ export async function createOPFSSink({ name }) { if (pendingBytes >= FLUSH_BYTES) await flush(); }, async file() { - await flush(); + await drain(); await writable.close(); return fileHandle.getFile(); }, diff --git a/src/sink.test.js b/src/sink.test.js index 82e7078..0263eeb 100644 --- a/src/sink.test.js +++ b/src/sink.test.js @@ -155,6 +155,135 @@ describe('download sink lifecycle', () => { }); }); + function deferred() { + let resolve; + let reject; + const promise = new Promise((res, rej) => { resolve = res; reject = rej; }); + return { promise, resolve, reject }; + } + + function opfsWith(writable) { + return { async getDirectory() { return { + async getFileHandle() { return { + async createWritable() { return writable; }, + async getFile() { return new Blob([]); } + }; }, + async removeEntry() {} + }; } }; + } + + const MiB = 1024 * 1024; + + it('keeps bytes in order with one write in flight at a time', async (t) => { + const written = []; + const events = []; + let outstanding = 0; + let maxOutstanding = 0; + const writable = { + write(bytes) { + const copy = Uint8Array.from(bytes); + const index = written.length; + written.push(copy); + outstanding++; + maxOutstanding = Math.max(maxOutstanding, outstanding); + // Vary completion timing so out-of-order completion would show up. + return new Promise((resolve) => setTimeout(resolve, index % 2 ? 5 : 0)) + .then(() => { outstanding--; events.push(`done:${index}`); }); + }, + async close() { events.push('close'); }, + async abort() {} + }; + await withBrowser(t, opfsWith(writable), async () => { + const sink = await createOPFSSink({ name: 'ordered.bin' }); + const expected = []; + for (let index = 0; index < 8; index++) { + const chunk = new Uint8Array(256 * 1024).fill(index + 1); + expected.push(chunk); + await sink.write(chunk); + } + const tail = new Uint8Array(1000).fill(0xee); + expected.push(tail); + await sink.write(tail); + await sink.file(); + const actual = Buffer.concat(written.map((part) => Buffer.from(part))); + assert.equal(actual.equals(Buffer.concat(expected.map((part) => Buffer.from(part)))), true); + assert.equal(maxOutstanding, 1); + assert.deepEqual(events.slice(0, -1), written.map((_, index) => `done:${index}`)); + assert.equal(events.at(-1), 'close'); + }); + }); + + it('surfaces an in-flight write error to the next flush that waits on it', async (t) => { + const first = deferred(); + let calls = 0; + const writable = { + write() { calls++; return calls === 1 ? first.promise : Promise.resolve(); }, + async close() { throw new Error('close must not run after a failed write'); }, + async abort() {} + }; + await withBrowser(t, opfsWith(writable), async () => { + const sink = await createOPFSSink({ name: 'fail.bin' }); + await sink.write(new Uint8Array(512 * 1024)); + // Second flush is due while the first is still in flight, so it waits on it. + const second = sink.write(new Uint8Array(512 * 1024)); + first.reject(new Error('disk full')); + await assert.rejects(second, { message: 'disk full' }); + assert.equal(calls, 1); + }); + }); + + it('surfaces an in-flight write error on the next write even without a flush', async (t) => { + const first = deferred(); + const writable = { + write() { return first.promise; }, + async close() { throw new Error('close must not run after a failed write'); }, + async abort() {} + }; + await withBrowser(t, opfsWith(writable), async () => { + const sink = await createOPFSSink({ name: 'fail-small.bin' }); + await sink.write(new Uint8Array(512 * 1024)); + first.reject(new Error('disk full')); + await new Promise((resolve) => setImmediate(resolve)); + await assert.rejects(sink.write(new Uint8Array(10)), { message: 'disk full' }); + }); + }); + + it('close waits for a pending write and reports its failure', async (t) => { + const pending = deferred(); + const events = []; + const writable = { + write() { return pending.promise.then(() => events.push('write-done'), (error) => { events.push('write-failed'); throw error; }); }, + async close() { events.push('close'); }, + async abort() {} + }; + await withBrowser(t, opfsWith(writable), async () => { + const sink = await createOPFSSink({ name: 'close-wait.bin' }); + await sink.write(new Uint8Array(512 * 1024)); + const closing = sink.file(); + await new Promise((resolve) => setImmediate(resolve)); + assert.deepEqual(events, []); + pending.resolve(); + await closing; + assert.deepEqual(events, ['write-done', 'close']); + }); + + const failing = deferred(); + const failingEvents = []; + const failingWritable = { + write() { return failing.promise; }, + async close() { failingEvents.push('close'); }, + async abort() {} + }; + await withBrowser(t, opfsWith(failingWritable), async () => { + const sink = await createOPFSSink({ name: 'close-fail.bin' }); + await sink.write(new Uint8Array(512 * 1024)); + const closing = sink.file(); + failing.reject(new Error('disk full')); + await assert.rejects(closing, { message: 'disk full' }); + assert.deepEqual(failingEvents, []); + }); + }); + it('preserves bytes in the fallback and revokes cleanup only once', async (t) => { await withBrowser(t, {}, async () => { const sink = await createSink({ name: 'hello.txt', size: 5, mediaType: 'text/plain' }); diff --git a/src/zip-bundle.js b/src/zip-bundle.js index 5dd3a19..2d2d288 100644 --- a/src/zip-bundle.js +++ b/src/zip-bundle.js @@ -158,11 +158,18 @@ export function fileSource(file, { chunkBytes = 256 * 1024 } = {}) { } // rechunk regroups a byte stream into pieces of exactly `size` bytes (the -// last may be shorter), the unit the relay protocol sends. +// last may be shorter), the unit the relay protocol sends. A piece that is +// already exactly `size` bytes with nothing buffered is yielded as-is, so the +// caller receives the source's own buffer: sources must not reuse a yielded +// buffer after handing it over (fileSource and the zip stream allocate fresh). export async function* rechunk(source, size) { let buffer = new Uint8Array(size); let filled = 0; for await (const piece of source) { + if (filled === 0 && piece.length === size) { + yield piece; + continue; + } let offset = 0; while (offset < piece.length) { const take = Math.min(size - filled, piece.length - offset); diff --git a/src/zip-bundle.test.js b/src/zip-bundle.test.js index b446fd2..216526f 100644 --- a/src/zip-bundle.test.js +++ b/src/zip-bundle.test.js @@ -43,3 +43,44 @@ test('rechunk emits fixed-size pieces', async () => { for await (const piece of rechunk(source.stream(), 4096)) sizes.push(piece.length); assert.deepEqual(sizes, [4096, 4096, 1808]); }); + +test('rechunk yields exact-size pieces from the source without copying', async () => { + const first = new Uint8Array(4096).fill(1); + const second = new Uint8Array(4096).fill(2); + async function* source() { + yield first; + yield second; + } + const out = []; + for await (const piece of rechunk(source(), 4096)) out.push(piece); + assert.equal(out.length, 2); + assert.equal(out[0], first); + assert.equal(out[1], second); +}); + +test('rechunk mixes fast-path and buffered pieces in order', async () => { + const bytes = Uint8Array.from({ length: 10 }, (_, index) => index); + async function* source() { + yield bytes.slice(0, 3); // partial: starts buffering + yield bytes.slice(3, 7); // 4 bytes: completes the first chunk + yield bytes.slice(7, 11); // 3 bytes: not exact, goes through the buffer + } + const out = []; + for await (const piece of rechunk(source(), 4)) out.push(Array.from(piece)); + assert.deepEqual(out, [[0, 1, 2, 3], [4, 5, 6, 7], [8, 9]]); + + // A buffered partial piece is copied; once the buffer flushes to empty, an + // exact-size piece takes the fast path and is yielded as the same object. + const data = Uint8Array.from({ length: 14 }, (_, index) => index); + const exact = data.slice(8, 12); + async function* mixed() { + yield data.slice(0, 2); // buffered + yield data.slice(2, 8); // completes [0..3], leaves [4..7] to copy + yield exact; // exact size with nothing buffered + yield data.slice(12, 14); // tail, flushed at end + } + const result = []; + for await (const piece of rechunk(mixed(), 4)) result.push(piece); + assert.deepEqual(result.map((piece) => Array.from(piece)), [[0, 1, 2, 3], [4, 5, 6, 7], [8, 9, 10, 11], [12, 13]]); + assert.equal(result[2], exact); +}); From aa55e74acb1061dc437a883de165d2d9ee2210a8 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 14 Oct 2026 05:34:31 +0000 Subject: [PATCH 3/3] Fix browser bench waiting 10 minutes on a hidden Connect button A complete short code connects as it is typed, which hides the button, so the bench's click retried until its 600s timeout and never measured the transfer. Start timing before the fill and click only if visible. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_012g5D2t4GGkGV1pn3Z2EcC4 --- scripts/bench.mjs | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/scripts/bench.mjs b/scripts/bench.mjs index a9cf479..2c623b5 100644 --- a/scripts/bench.mjs +++ b/scripts/bench.mjs @@ -150,10 +150,12 @@ async function browserTransfer(browser, baseUrl, payloadPath, payload, mode) { await receiver.goto(baseUrl); await receiver.locator('.workbench:not([inert])').waitFor(); await receiver.locator('#receive-mode-btn').click(); - await receiver.locator('#code-input').fill(code); const downloadEvent = receiver.waitForEvent('download', { timeout: 600_000 }); const started = performance.now(); - await receiver.locator('#connect-btn').click(); + // A complete short code connects as it is typed; longer input waits + // for the button. + await receiver.locator('#code-input').fill(code); + if (await receiver.locator('#connect-btn').isVisible()) await receiver.locator('#connect-btn').click(); const download = await downloadEvent; const seconds = (performance.now() - started) / 1000; const savedPath = join(work, `${mode}.bin`);