Daan MeijerandClaude Opus 5.5
168605148f
decodeFrames joined each network piece (~64 KB) onto the frame so far: a chunk of MBs copied over and over, 5.4% of a worker's time in an end-to-end profile (the bench reads files, so it never showed). Pieces are now joined once a frame is complete. A batch post's gzip (1.6%) runs on a thread-pool thread. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
40 lines
2.0 KiB
TypeScript
40 lines
2.0 KiB
TypeScript
import { test } from '@japa/runner'
|
|
import { KEEPALIVE, decodeFrames, encodeFrame, isKeepalive } from '#lib/edge/index'
|
|
|
|
test.group('edge / stream frames', () => {
|
|
test('frames survive being split anywhere; keep-alive frames are recognised', async ({ assert }) => {
|
|
const whole = Buffer.concat([
|
|
encodeFrame({ from: 160_000_000_001, to: 160_000_050_000, data: Buffer.from('first') }),
|
|
encodeFrame(KEEPALIVE),
|
|
encodeFrame({ from: 160_000_050_001, to: 160_000_100_000, data: Buffer.from('second chunk') }),
|
|
])
|
|
const pieces = [whole.subarray(0, 7), whole.subarray(7, 30), whole.subarray(30)]
|
|
const frames = []
|
|
for await (const frame of decodeFrames(pieces)) frames.push(frame)
|
|
assert.deepEqual(frames.map((f) => [f.from, f.to, f.data.toString(), isKeepalive(f)]), [
|
|
[160_000_000_001, 160_000_050_000, 'first', false],
|
|
[0, 0, '', true],
|
|
[160_000_050_001, 160_000_100_000, 'second chunk', false],
|
|
])
|
|
})
|
|
|
|
test('a stream that ends inside a frame is an error', async ({ assert }) => {
|
|
const cut = encodeFrame({ from: 1, to: 2, data: Buffer.from('abc') }).subarray(0, 21)
|
|
await assert.rejects(async () => {
|
|
for await (const _ of decodeFrames([cut])) {
|
|
}
|
|
}, /inside a frame/)
|
|
})
|
|
test('frames come out whole whatever the piece sizes, also pieces holding several frames', async ({ assert }) => {
|
|
const sent = Array.from({ length: 12 }, (_, i) => ({ from: i * 10 + 1, to: i * 10 + 10, data: Buffer.alloc((i * 997) % 5000, i) }))
|
|
const whole = Buffer.concat(sent.map(encodeFrame))
|
|
for (const size of [1, 3, 20, 21, 64, 1000, 4096, whole.length]) {
|
|
const pieces = []
|
|
for (let at = 0; at < whole.length; at += size) pieces.push(new Uint8Array(whole.subarray(at, at + size)))
|
|
const frames = []
|
|
for await (const frame of decodeFrames(pieces)) frames.push(frame)
|
|
assert.deepEqual(frames.map((f) => [f.from, f.to, f.data.toString('hex')]), sent.map((f) => [f.from, f.to, f.data.toString('hex')]), `pieces of ${size}`)
|
|
}
|
|
})
|
|
})
|