Skip to content

Commit 30528b0

Browse files
thibmeuthibmeu
committed
perf(chunked): pipeline the buffer-path methods through the streaming transforms
The four buffer chunked methods (ChunkedOHTTPClient.encapsulate / decapsulateResponse, ChunkedOHTTPServer.decapsulate / encapsulateResponse) sealed or opened one chunk per await, so the per-call WebCrypto latency never overlapped. The Request/Response API already streamed through the createRequest*/Response*Transform pipelines, which keep a window of AEAD calls in flight. Now the buffer methods use them too: wrap the input in a one-chunk stream, run it through the chunker and the encrypt/decrypt transform, collect the bytes (new streamOfBytes/collectStream helpers in streaming.ts). I went through the transforms rather than the per-chunk context closures because the closures bump their counter after the await and can't run concurrently; the transforms claim it up front. So the response contexts now carry the derived _aead/_aeadKey/_aeadNonce, next to the _senderContext / _recipientContext the request contexts already expose. createRequestEncryptTransform was the last serial transform, so it gets the same inflight window as the response side. Wire format is unchanged and the 125 tests pass.
1 parent af3398b commit 30528b0

3 files changed

Lines changed: 119 additions & 139 deletions

File tree

‎src/client.ts‎

Lines changed: 28 additions & 61 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import type { CipherSuite, SenderContext } from "hpke";
1+
import type { AEAD as AeadImpl, CipherSuite, SenderContext } from "hpke";
22
import { bhttp, MediaType } from "./constants.js";
33
import {
44
buildRequestHeader,
@@ -12,10 +12,8 @@ import {
1212
deriveChunkedResponseKeys,
1313
encapsulateRequest,
1414
FINAL_CHUNK_AAD,
15-
frameChunk,
1615
getResponseNonceLength,
1716
openResponseChunk,
18-
parseFramedChunk,
1917
type ResponseCrypto,
2018
} from "./encapsulation.js";
2119
import { OHTTPError, OHTTPErrorCode } from "./errors.js";
@@ -27,11 +25,13 @@ import {
2725
type KeyConfig,
2826
} from "./keyConfig.js";
2927
import {
28+
collectStream,
3029
createChunkerTransform,
3130
createRequestEncryptTransform,
3231
createResponseDecryptTransform,
3332
decodeBHttpResponseStream,
3433
encodeBHttpRequestStream,
34+
streamOfBytes,
3535
} from "./streaming.js";
3636
import { concat, toArrayBuffer } from "./utils.js";
3737

@@ -130,6 +130,12 @@ export interface ChunkedResponseContext {
130130
openChunk(ciphertext: Uint8Array): Promise<Uint8Array>;
131131
/** Open the final chunk */
132132
openFinalChunk(ciphertext: Uint8Array): Promise<Uint8Array>;
133+
/** @internal Derived response AEAD (for the pipelined buffer path) */
134+
readonly _aead: AeadImpl;
135+
/** @internal Derived response AEAD key */
136+
readonly _aeadKey: Uint8Array;
137+
/** @internal Derived response AEAD base nonce */
138+
readonly _aeadNonce: Uint8Array;
133139
}
134140

135141
/**
@@ -431,6 +437,10 @@ export class ChunkedOHTTPClient {
431437
const maxChunks = 2 ** 32;
432438

433439
return {
440+
_aead: aead,
441+
_aeadKey: aeadKey,
442+
_aeadNonce: aeadNonce,
443+
434444
async openChunk(ciphertext: Uint8Array): Promise<Uint8Array> {
435445
if (counter >= maxChunks) {
436446
throw new OHTTPError(OHTTPErrorCode.ChunkLimitExceeded);
@@ -471,34 +481,16 @@ export class ChunkedOHTTPClient {
471481
}> {
472482
const ctx = await this.createRequestContext();
473483

474-
const chunks: Uint8Array[] = [ctx.header];
475-
476-
// Split request into chunks
477-
let offset = 0;
478-
while (offset < request.length) {
479-
const remaining = request.length - offset;
480-
const isLast = remaining <= this.maxChunkSize;
481-
const chunkSize = Math.min(remaining, this.maxChunkSize);
482-
const chunk = request.subarray(offset, offset + chunkSize);
483-
offset += chunkSize;
484-
485-
if (isLast) {
486-
const sealed = await ctx.sealFinalChunk(chunk);
487-
chunks.push(frameChunk(sealed, true));
488-
} else {
489-
const sealed = await ctx.sealChunk(chunk);
490-
chunks.push(frameChunk(sealed, false));
491-
}
492-
}
493-
494-
// Handle empty request
495-
if (request.length === 0) {
496-
const sealed = await ctx.sealFinalChunk(new Uint8Array(0));
497-
chunks.push(frameChunk(sealed, true));
498-
}
484+
// Chunk + seal through the same pipelined transforms as the streaming API
485+
// (the seals run in a concurrent window), then prepend the header.
486+
const sealed = await collectStream(
487+
streamOfBytes(request)
488+
.pipeThrough(createChunkerTransform(this.maxChunkSize))
489+
.pipeThrough(createRequestEncryptTransform(ctx._senderContext)),
490+
);
499491

500492
return {
501-
encapsulatedRequest: concat(...chunks),
493+
encapsulatedRequest: concat(ctx.header, sealed),
502494
responseNonceLength: getResponseNonceLength(this.suite),
503495
createResponseContext: (nonce) => ctx.createResponseContext(nonce),
504496
};
@@ -522,38 +514,13 @@ export class ChunkedOHTTPClient {
522514
const responseNonce = encapsulatedResponse.subarray(0, nonceLength);
523515
const ctx = await createResponseContext(responseNonce);
524516

525-
// Parse and decrypt chunks
526-
const responseChunks: Uint8Array[] = [];
527-
let data = encapsulatedResponse.subarray(nonceLength);
528-
let sawFinal = false;
529-
530-
while (data.length > 0) {
531-
const parsed = parseFramedChunk(data);
532-
if (parsed === undefined) {
533-
throw new OHTTPError(OHTTPErrorCode.InvalidMessage);
534-
}
535-
536-
if (parsed.isFinal) {
537-
responseChunks.push(await ctx.openFinalChunk(parsed.ciphertext));
538-
sawFinal = true;
539-
break;
540-
}
541-
542-
const chunk = await ctx.openChunk(parsed.ciphertext);
543-
// A non-final chunk MUST NOT decrypt to zero-length plaintext (draft-08 Section 7.3).
544-
if (chunk.length === 0) {
545-
throw new OHTTPError(OHTTPErrorCode.DecryptionFailed);
546-
}
547-
responseChunks.push(chunk);
548-
data = data.subarray(parsed.bytesConsumed);
549-
}
550-
551-
// Without a final (0-length prefix) chunk the message is truncated.
552-
if (!sawFinal) {
553-
throw new OHTTPError(OHTTPErrorCode.InvalidMessage);
554-
}
555-
556-
return concat(...responseChunks);
517+
// Decrypt the framed body through the pipelined response transform
518+
// (opens run in a concurrent window; counters are claimed synchronously).
519+
return collectStream(
520+
streamOfBytes(encapsulatedResponse.subarray(nonceLength)).pipeThrough(
521+
createResponseDecryptTransform(ctx._aead, ctx._aeadKey, ctx._aeadNonce),
522+
),
523+
);
557524
}
558525

559526
/**

‎src/server.ts‎

Lines changed: 34 additions & 60 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,4 @@
1-
import type { RecipientContext } from "hpke";
1+
import type { AEAD as AeadImpl, RecipientContext } from "hpke";
22
import { bhttp, MediaType } from "./constants.js";
33
import {
44
buildRequestInfo,
@@ -11,22 +11,22 @@ import {
1111
deriveChunkedResponseKeys,
1212
encapsulateResponse,
1313
FINAL_CHUNK_AAD,
14-
frameChunk,
1514
getEncLength,
1615
getResponseNonceLength,
17-
parseFramedChunk,
1816
parseRequestHeader,
1917
type ResponseCrypto,
2018
sealResponseChunk,
2119
} from "./encapsulation.js";
2220
import { OHTTPError, OHTTPErrorCode } from "./errors.js";
2321
import type { KeyConfigWithPrivate } from "./keyConfig.js";
2422
import {
23+
collectStream,
2524
createChunkerTransform,
2625
createRequestDecryptTransform,
2726
createResponseEncryptTransform,
2827
decodeBHttpRequestStream,
2928
encodeBHttpResponseStream,
29+
streamOfBytes,
3030
} from "./streaming.js";
3131
import { concat, toArrayBuffer } from "./utils.js";
3232

@@ -120,6 +120,12 @@ export interface ChunkedServerResponseContext {
120120
sealChunk(chunk: Uint8Array): Promise<Uint8Array>;
121121
/** Seal the final chunk */
122122
sealFinalChunk(chunk: Uint8Array): Promise<Uint8Array>;
123+
/** @internal Derived response AEAD (for the pipelined buffer path) */
124+
readonly _aead: AeadImpl;
125+
/** @internal Derived response AEAD key */
126+
readonly _aeadKey: Uint8Array;
127+
/** @internal Derived response AEAD base nonce */
128+
readonly _aeadNonce: Uint8Array;
123129
}
124130

125131
/**
@@ -373,6 +379,9 @@ export class ChunkedOHTTPServer {
373379

374380
return {
375381
responseNonce,
382+
_aead: aead,
383+
_aeadKey: aeadKey,
384+
_aeadNonce: aeadNonce,
376385

377386
async sealChunk(chunk: Uint8Array): Promise<Uint8Array> {
378387
if (counter >= maxChunks) {
@@ -410,39 +419,16 @@ export class ChunkedOHTTPServer {
410419

411420
const ctx = await this.createRequestContext(header);
412421

413-
// Parse and decrypt all chunks
414-
const requestChunks: Uint8Array[] = [];
415-
let data = encapsulatedRequest.subarray(headerOffset);
416-
let sawFinal = false;
417-
418-
while (data.length > 0) {
419-
const parsed = parseFramedChunk(data);
420-
if (parsed === undefined) {
421-
throw new OHTTPError(OHTTPErrorCode.InvalidMessage);
422-
}
423-
424-
if (parsed.isFinal) {
425-
requestChunks.push(await ctx.openFinalChunk(parsed.ciphertext));
426-
sawFinal = true;
427-
break;
428-
}
429-
430-
const chunk = await ctx.openChunk(parsed.ciphertext);
431-
// A non-final chunk MUST NOT decrypt to zero-length plaintext (draft-08 Section 7.3).
432-
if (chunk.length === 0) {
433-
throw new OHTTPError(OHTTPErrorCode.DecryptionFailed);
434-
}
435-
requestChunks.push(chunk);
436-
data = data.subarray(parsed.bytesConsumed);
437-
}
438-
439-
// Without a final (0-length prefix) chunk the message is truncated.
440-
if (!sawFinal) {
441-
throw new OHTTPError(OHTTPErrorCode.InvalidMessage);
442-
}
422+
// Decrypt the framed body through the request decrypt transform (opens run
423+
// in a concurrent window).
424+
const request = await collectStream(
425+
streamOfBytes(encapsulatedRequest.subarray(headerOffset)).pipeThrough(
426+
createRequestDecryptTransform(ctx._recipientContext),
427+
),
428+
);
443429

444430
return {
445-
request: concat(...requestChunks),
431+
request,
446432
keyConfig: ctx.keyConfig,
447433
createResponseContext: () => ctx.createResponseContext(),
448434
};
@@ -457,33 +443,21 @@ export class ChunkedOHTTPServer {
457443
responseContext: ChunkedServerResponseContext,
458444
response: Uint8Array,
459445
): Promise<Uint8Array> {
460-
const chunks: Uint8Array[] = [responseContext.responseNonce];
461-
462-
// Split response into chunks
463-
let offset = 0;
464-
while (offset < response.length) {
465-
const remaining = response.length - offset;
466-
const isLast = remaining <= this.maxChunkSize;
467-
const chunkSize = Math.min(remaining, this.maxChunkSize);
468-
const chunk = response.subarray(offset, offset + chunkSize);
469-
offset += chunkSize;
470-
471-
if (isLast) {
472-
const sealed = await responseContext.sealFinalChunk(chunk);
473-
chunks.push(frameChunk(sealed, true));
474-
} else {
475-
const sealed = await responseContext.sealChunk(chunk);
476-
chunks.push(frameChunk(sealed, false));
477-
}
478-
}
479-
480-
// Handle empty response
481-
if (response.length === 0) {
482-
const sealed = await responseContext.sealFinalChunk(new Uint8Array(0));
483-
chunks.push(frameChunk(sealed, true));
484-
}
446+
// Chunk + seal through the pipelined response transform (seals run in a
447+
// concurrent window), prefixed with the response nonce.
448+
const sealed = await collectStream(
449+
streamOfBytes(response)
450+
.pipeThrough(createChunkerTransform(this.maxChunkSize))
451+
.pipeThrough(
452+
createResponseEncryptTransform(
453+
responseContext._aead,
454+
responseContext._aeadKey,
455+
responseContext._aeadNonce,
456+
),
457+
),
458+
);
485459

486-
return concat(...chunks);
460+
return concat(responseContext.responseNonce, sealed);
487461
}
488462

489463
/**

‎src/streaming.ts‎

Lines changed: 57 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,29 @@ import {
1616
} from "./constants.js";
1717
import { FINAL_CHUNK_AAD, openResponseChunk, sealResponseChunk } from "./encapsulation.js";
1818
import { OHTTPError, OHTTPErrorCode } from "./errors.js";
19+
import { concat } from "./utils.js";
20+
21+
/** A ReadableStream that emits `bytes` as a single chunk (if non-empty) then closes. */
22+
export function streamOfBytes(bytes: Uint8Array): ReadableStream<Uint8Array> {
23+
return new ReadableStream<Uint8Array>({
24+
start(controller) {
25+
if (bytes.length > 0) controller.enqueue(bytes);
26+
controller.close();
27+
},
28+
});
29+
}
30+
31+
/** Collect a stream of byte chunks into one contiguous buffer. */
32+
export async function collectStream(stream: ReadableStream<Uint8Array>): Promise<Uint8Array> {
33+
const parts: Uint8Array[] = [];
34+
const reader = stream.getReader();
35+
for (;;) {
36+
const { done, value } = await reader.read();
37+
if (done) break;
38+
parts.push(value);
39+
}
40+
return concat(...parts);
41+
}
1942

2043
/**
2144
* Maximum chunks allowed per draft-ietf-ohai-chunked-ohttp-08 Section 7.3
@@ -145,36 +168,52 @@ export function createRequestEncryptTransform(
145168
senderContext: SenderContext,
146169
): TransformStream<Uint8Array, Uint8Array> {
147170
let pendingChunk: Uint8Array | undefined;
171+
// Seals issued but not yet emitted, oldest first (all non-final chunks).
172+
// hpke's SenderContext.Seal claims its sequence number synchronously, so a
173+
// window of seals can run concurrently while output stays in order.
174+
const inflight: Array<Promise<SettledChunk>> = [];
175+
176+
// Await the oldest in-flight seal and enqueue its frame (order-preserving).
177+
const emitOldest = async (
178+
controller: TransformStreamDefaultController<Uint8Array>,
179+
): Promise<boolean> => {
180+
const result = await (inflight.shift() as Promise<SettledChunk>);
181+
if (!result.ok) {
182+
controller.error(new OHTTPError(OHTTPErrorCode.EncryptionFailed));
183+
return false;
184+
}
185+
// length prefix + ciphertext as two enqueues (avoids a copy)
186+
controller.enqueue(encodeVarint(result.value.length));
187+
controller.enqueue(result.value);
188+
return true;
189+
};
148190

149191
return new TransformStream<Uint8Array, Uint8Array>({
150192
async transform(chunk, controller) {
151-
// If we have a pending chunk, seal it as non-final
193+
// Seal the previous chunk as non-final; emit once the window is full so
194+
// independent seals overlap.
152195
if (pendingChunk !== undefined) {
153-
try {
154-
const sealed = await senderContext.Seal(pendingChunk);
155-
// length prefix + ciphertext as two enqueues (avoids a copy)
156-
controller.enqueue(encodeVarint(sealed.length));
157-
controller.enqueue(sealed);
158-
} catch {
159-
controller.error(new OHTTPError(OHTTPErrorCode.EncryptionFailed));
160-
return;
161-
}
196+
inflight.push(settle(senderContext.Seal(pendingChunk)));
197+
if (inflight.length >= AEAD_PIPELINE_DEPTH && !(await emitOldest(controller))) return;
162198
}
163199
// Store current chunk as pending (might be final)
164200
pendingChunk = chunk;
165201
},
166202

167203
async flush(controller) {
168-
// Seal the last chunk as final
169-
const finalChunk = pendingChunk ?? EMPTY;
170-
try {
171-
const sealed = await senderContext.Seal(finalChunk, FINAL_CHUNK_AAD);
172-
// Final chunk has length prefix 0
173-
controller.enqueue(encodeVarint(0));
174-
controller.enqueue(sealed);
175-
} catch {
204+
// Seal the last chunk as final, draining the window in order first.
205+
const final = settle(senderContext.Seal(pendingChunk ?? EMPTY, FINAL_CHUNK_AAD));
206+
while (inflight.length > 0) {
207+
if (!(await emitOldest(controller))) return;
208+
}
209+
const result = await final;
210+
if (!result.ok) {
176211
controller.error(new OHTTPError(OHTTPErrorCode.EncryptionFailed));
212+
return;
177213
}
214+
// Final chunk has length prefix 0
215+
controller.enqueue(encodeVarint(0));
216+
controller.enqueue(result.value);
178217
},
179218
});
180219
}

0 commit comments

Comments
 (0)