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 +} 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`); 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); +});