From 7ecd7004ebc1c45c137d9db7d340738236009a01 Mon Sep 17 00:00:00 2001 From: imccyu <276526105+imccyu@users.noreply.github.com> Date: Thu, 27 Aug 2026 01:26:32 +0800 Subject: [PATCH] feat(inspector): project Host fetches through CDP Network --- .../src/client/inspection/network.ts | 4 + .../inspector/src/host/inspection/network.ts | 244 ++++++++++ .../src/shared/bridge/messages/network.ts | 12 + .../src/shared/network/observation.ts | 53 +++ .../src/worker/cdp/domains/network/session.ts | 211 +++++++++ .../src/worker/inspection/network-store.ts | 429 ++++++++++++++++++ .../tests/fetch-observer.host.spec.ts | 124 +++++ .../inspector/tests/network.host.spec.ts | 153 +++++++ 8 files changed, 1230 insertions(+) create mode 100644 packages/experimental/inspector/src/client/inspection/network.ts create mode 100644 packages/experimental/inspector/src/host/inspection/network.ts create mode 100644 packages/experimental/inspector/src/shared/bridge/messages/network.ts create mode 100644 packages/experimental/inspector/src/shared/network/observation.ts create mode 100644 packages/experimental/inspector/src/worker/cdp/domains/network/session.ts create mode 100644 packages/experimental/inspector/src/worker/inspection/network-store.ts create mode 100644 packages/experimental/inspector/tests/fetch-observer.host.spec.ts create mode 100644 packages/experimental/inspector/tests/network.host.spec.ts diff --git a/packages/experimental/inspector/src/client/inspection/network.ts b/packages/experimental/inspector/src/client/inspection/network.ts new file mode 100644 index 0000000000..23f4e6d818 --- /dev/null +++ b/packages/experimental/inspector/src/client/inspection/network.ts @@ -0,0 +1,4 @@ +/** Client network observation is not enabled in the current source producer. */ + +/** Observation topics published by the Client network adapter. */ +export const NETWORK_TOPICS: readonly string[] = [] diff --git a/packages/experimental/inspector/src/host/inspection/network.ts b/packages/experimental/inspector/src/host/inspection/network.ts new file mode 100644 index 0000000000..d73d979fa9 --- /dev/null +++ b/packages/experimental/inspector/src/host/inspection/network.ts @@ -0,0 +1,244 @@ +/** Full `globalThis.fetch` capture that publishes without delaying response delivery. */ + +import type { InspectorJsonValue } from '../../shared/json.ts' +import type { InspectorPublisher } from '../../shared/bridge/publisher.ts' +import { FETCH_TOPICS } from '../../shared/bridge/messages/network.ts' + +/** Observation topics published by the Host network adapter. */ +export const NETWORK_TOPICS: readonly string[] = FETCH_TOPICS + +/** Byte limits for request and response clone capture. */ +export interface FetchCaptureOptions { + readonly maxRequestBodyBytes: number + readonly maxResponseBodyBytes: number + readonly maxChunkBytes: number +} + +interface CaptureOutcome { + readonly capturedBytes: number + readonly truncated: boolean + readonly captureError?: string +} + +/** Active global fetch wrapper. */ +export interface FetchObserver { + /** Restore the prior fetch implementation, cancel clone readers, and await their settlement. */ + stop(): Promise +} + +/** + * Install full fetch capture for every later call through `globalThis.fetch`. + * @param publisher - Host source that receives fetch lifecycle records. + * @param options - Per-body capture limits. + * @returns The owner that stops capture and awaits pending body readers. + */ +export function installFetchObserver( + publisher: InspectorPublisher, + options: FetchCaptureOptions, +): FetchObserver { + const descriptor = Object.getOwnPropertyDescriptor(globalThis, 'fetch') + const original = globalThis.fetch + if (typeof original !== 'function') throw new Error('inspector: globalThis.fetch is unavailable') + if (descriptor !== undefined && !('value' in descriptor)) { + throw new Error('inspector: globalThis.fetch is an accessor and cannot be observed safely') + } + + const controller = new AbortController() + const pending = new Set>() + let nextRequestId = 0 + + const track = (promise: Promise): void => { + pending.add(promise) + void promise.then( + () => { pending.delete(promise) }, + () => { pending.delete(promise) }, + ) + } + + const observedFetch: typeof fetch = async (input, init) => { + const request = new Request(input, init) + const requestId = `fetch-${++nextRequestId}` + publisher.publish('fetch/start', { + requestId, + url: request.url, + method: request.method, + headers: headerEntries(request.headers), + hasBody: request.body !== null, + wallTimeMs: Date.now(), + }) + + let requestClone: Request | undefined + try { + requestClone = request.clone() + } catch (error) { + publisher.publish('fetch/request-body-end', { + requestId, + capturedBytes: 0, + truncated: false, + captureError: renderError(error), + }) + } + if (requestClone !== undefined) { + track(captureBody( + requestClone.body, + options.maxRequestBodyBytes, + options.maxChunkBytes, + controller.signal, + (data) => { publisher.publish('fetch/request-body-chunk', { requestId, data }) }, + ).then((outcome) => { + publisher.publish('fetch/request-body-end', compactOutcome(requestId, outcome)) + })) + } + + let response: Response + try { + response = await Reflect.apply(original, globalThis, [request]) + } catch (error) { + publisher.publish('fetch/error', { + requestId, + message: renderError(error), + canceled: request.signal.aborted || isAbortError(error), + }) + throw error + } + + publisher.publish('fetch/response', { + requestId, + url: response.url || request.url, + status: response.status, + statusText: response.statusText, + headers: headerEntries(response.headers), + mimeType: response.headers.get('content-type')?.split(';', 1)[0]?.trim() ?? '', + }) + + try { + const responseClone = response.clone() + track(captureBody( + responseClone.body, + options.maxResponseBodyBytes, + options.maxChunkBytes, + AbortSignal.any([controller.signal, request.signal]), + (data) => { publisher.publish('fetch/response-body-chunk', { requestId, data }) }, + ).then((outcome) => { + if (request.signal.aborted) { + publisher.publish('fetch/error', { + requestId, + message: request.signal.reason === undefined + ? 'AbortError: request aborted during response body capture' + : renderError(request.signal.reason), + canceled: true, + }) + return + } + publisher.publish('fetch/end', { + requestId, + capturedBytes: outcome.capturedBytes, + responseBodyTruncated: outcome.truncated, + ...(outcome.captureError === undefined ? {} : { responseCaptureError: outcome.captureError }), + }) + })) + } catch (error) { + publisher.publish('fetch/end', { + requestId, + capturedBytes: 0, + responseBodyTruncated: false, + responseCaptureError: renderError(error), + }) + } + return response + } + + Object.defineProperty(observedFetch, 'name', { value: original.name, configurable: true }) + Object.defineProperty(observedFetch, 'length', { value: original.length, configurable: true }) + Object.defineProperty(globalThis, 'fetch', descriptor === undefined + ? { value: observedFetch, writable: true, configurable: true } + : { ...descriptor, value: observedFetch }) + + let stopped: Promise | undefined + return { + stop(): Promise { + if (stopped !== undefined) return stopped + stopped = (async () => { + const current = Object.getOwnPropertyDescriptor(globalThis, 'fetch') + if (current !== undefined && 'value' in current && current.value === observedFetch) { + if (descriptor === undefined) Reflect.deleteProperty(globalThis, 'fetch') + else Object.defineProperty(globalThis, 'fetch', descriptor) + } + controller.abort() + await Promise.allSettled([...pending]) + })() + return stopped + }, + } +} + +async function captureBody( + body: ReadableStream | null, + limit: number, + chunkLimit: number, + signal: AbortSignal, + emit: (base64: string) => void, +): Promise { + if (body === null) return { capturedBytes: 0, truncated: false } + const reader = body.getReader() + const abort = (): void => { void reader.cancel(signal.reason).catch(() => undefined) } + signal.addEventListener('abort', abort, { once: true }) + let capturedBytes = 0 + let truncated = false + try { + while (!signal.aborted) { + const item = await reader.read() + if (item.done) break + let offset = 0 + while (offset < item.value.byteLength) { + const remaining = limit - capturedBytes + if (remaining <= 0) { + truncated = true + void reader.cancel('inspector body capture limit reached').catch(() => undefined) + return { capturedBytes, truncated } + } + const size = Math.min(chunkLimit, remaining, item.value.byteLength - offset) + const chunk = item.value.subarray(offset, offset + size) + emit(Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength).toString('base64')) + capturedBytes += size + offset += size + } + } + if (signal.aborted) { + void reader.cancel(signal.reason).catch(() => undefined) + return { capturedBytes, truncated, captureError: 'inspector stopped during body capture' } + } + return { capturedBytes, truncated } + } catch (error) { + return { capturedBytes, truncated, captureError: renderError(error) } + } finally { + signal.removeEventListener('abort', abort) + reader.releaseLock() + } +} + +function compactOutcome(requestId: string, outcome: CaptureOutcome): InspectorJsonValue { + return { + requestId, + capturedBytes: outcome.capturedBytes, + truncated: outcome.truncated, + ...(outcome.captureError === undefined ? {} : { captureError: outcome.captureError }), + } +} + +function headerEntries(headers: Headers): [string, string][] { + return [...headers.entries()] +} + +function isAbortError(error: unknown): boolean { + return error instanceof DOMException && error.name === 'AbortError' +} + +function renderError(error: unknown): string { + if (error instanceof Error) return `${error.name}: ${error.message}` + try { + return String(error) + } catch { + return 'unrenderable fetch error' + } +} diff --git a/packages/experimental/inspector/src/shared/bridge/messages/network.ts b/packages/experimental/inspector/src/shared/bridge/messages/network.ts new file mode 100644 index 0000000000..df83a777aa --- /dev/null +++ b/packages/experimental/inspector/src/shared/bridge/messages/network.ts @@ -0,0 +1,12 @@ +/** Observation topic names carried by the internal bridge for captured fetches. */ + +/** Complete set of fetch observation topics. */ +export const FETCH_TOPICS = [ + 'fetch/start', + 'fetch/request-body-chunk', + 'fetch/request-body-end', + 'fetch/response', + 'fetch/response-body-chunk', + 'fetch/end', + 'fetch/error', +] as const diff --git a/packages/experimental/inspector/src/shared/network/observation.ts b/packages/experimental/inspector/src/shared/network/observation.ts new file mode 100644 index 0000000000..b3266dc7c7 --- /dev/null +++ b/packages/experimental/inspector/src/shared/network/observation.ts @@ -0,0 +1,53 @@ +/** Full-capture fetch observations sent to the Inspector Worker. */ + +/** One header entry; arrays retain duplicate header names. */ +export type InspectorHeader = readonly [name: string, value: string] + +/** Common request identity. */ +export interface FetchIdentity { + readonly requestId: string +} + +/** A high-level global fetch call began. */ +export interface FetchStartPayload extends FetchIdentity { + readonly url: string + readonly method: string + readonly headers: InspectorHeader[] + readonly hasBody: boolean + readonly wallTimeMs: number +} + +/** One captured request-body chunk. */ +export interface FetchBodyChunkPayload extends FetchIdentity { + readonly data: string +} + +/** Terminal state of one captured request body. */ +export interface FetchRequestBodyEndPayload extends FetchIdentity { + readonly capturedBytes: number + readonly truncated: boolean + readonly captureError?: string +} + +/** Fetch resolved with response headers. */ +export interface FetchResponsePayload extends FetchIdentity { + readonly url: string + readonly status: number + readonly statusText: string + readonly headers: InspectorHeader[] + readonly mimeType: string +} + +/** One captured response-body chunk. */ +/** Fetch capture reached a terminal response-body state. */ +export interface FetchEndPayload extends FetchIdentity { + readonly capturedBytes: number + readonly responseBodyTruncated: boolean + readonly responseCaptureError?: string +} + +/** Fetch rejected before returning a Response. */ +export interface FetchErrorPayload extends FetchIdentity { + readonly message: string + readonly canceled: boolean +} diff --git a/packages/experimental/inspector/src/worker/cdp/domains/network/session.ts b/packages/experimental/inspector/src/worker/cdp/domains/network/session.ts new file mode 100644 index 0000000000..1988eca41a --- /dev/null +++ b/packages/experimental/inspector/src/worker/cdp/domains/network/session.ts @@ -0,0 +1,211 @@ +/** CDP Network projection over the Worker-owned normalized network store. */ + +import { Buffer } from 'node:buffer' +import type { InspectorHeader } from '../../../../shared/network/observation.ts' +import type { NetworkStore, NetworkStoreEvent } from '../../../inspection/network-store.ts' + +/** CDP session slice used by the Network domain. */ +export interface NetworkSink { + sendEvent(method: string, params: Readonly>): void +} + +/** Projects retained and live network observations into connection-local CDP state. */ +export class NetworkDomain { + private readonly enabled = new Set() + private readonly streamedRequests = new Map>() + private readonly unsubscribe: () => void + + constructor(private readonly store: NetworkStore) { + this.unsubscribe = store.subscribe((event) => { this.receive(event) }) + } + + /** + * Enable Network for one DevTools connection and replay retained lifecycle events. + * @param session - Connection receiving replay and subsequent events. + */ + enable(session: NetworkSink): void { + if (this.enabled.has(session)) return + for (const event of this.store.replay()) this.send(session, event) + this.enabled.add(session) + } + + /** + * Stop Network events for one DevTools connection. + * @param session - Connection leaving the enabled set. + */ + disable(session: NetworkSink): void { + this.enabled.delete(session) + this.streamedRequests.delete(session) + } + + /** + * Forget a closed DevTools connection. + * @param session - Closed DevTools connection. + */ + detach(session: NetworkSink): void { + this.disable(session) + } + + /** Release the repository subscription and all connection-local state. */ + close(): void { + this.unsubscribe() + this.enabled.clear() + this.streamedRequests.clear() + } + + /** + * Handle one Worker-local Network method. + * @param method - CDP method name. + * @param params - Parsed request parameters. + * @param session - Calling DevTools connection. + * @returns The CDP result fields. + */ + handle(method: string, params: Readonly>, session: NetworkSink): unknown { + switch (method) { + case 'Network.enable': + this.enable(session) + return {} + case 'Network.disable': + this.disable(session) + return {} + case 'Network.getResponseBody': { + const body = this.store.responseBody(params.requestId) + return { + body: Buffer.from(body.bytes).toString('base64'), + base64Encoded: true, + dshInspectorTruncated: body.truncated, + ...(body.captureError === undefined ? {} : { dshInspectorCaptureError: body.captureError }), + } + } + case 'Network.getRequestPostData': { + const body = this.store.requestBody(params.requestId) + return { + postData: Buffer.from(body.bytes).toString('utf8'), + dshInspectorTruncated: body.truncated, + ...(body.captureError === undefined ? {} : { dshInspectorCaptureError: body.captureError }), + } + } + case 'Network.streamResourceContent': { + const body = this.store.responseBody(params.requestId) + if (typeof params.requestId !== 'string') throw new Error('Network requestId must be a string') + if (!body.complete) { + let requests = this.streamedRequests.get(session) + if (requests === undefined) this.streamedRequests.set(session, requests = new Set()) + requests.add(params.requestId) + } + return { bufferedData: Buffer.from(body.bytes).toString('base64') } + } + case 'Network.setCacheDisabled': + case 'Network.setBypassServiceWorker': + case 'Network.setExtraHTTPHeaders': + case 'Network.clearBrowserCache': + case 'Network.clearBrowserCookies': + return {} + default: + throw new Error(`unsupported Network method ${method}`) + } + } + + private receive(event: NetworkStoreEvent): void { + if (event.type === 'request-evicted') { + for (const [session, requests] of this.streamedRequests) { + requests.delete(event.requestKey) + if (requests.size === 0) this.streamedRequests.delete(session) + } + return + } + for (const session of this.enabled) this.send(session, event) + } + + private send(session: NetworkSink, event: Exclude): void { + const timestamp = (event.timestampMs - performance.timeOrigin) / 1_000 + switch (event.type) { + case 'request-started': + session.sendEvent('Network.requestWillBeSent', { + requestId: event.requestId, + loaderId: 'dsh-inspector-loader', + documentURL: 'dsh://host', + request: { + url: event.url, + method: event.method, + headers: cdpHeaders(event.headers), + hasPostData: event.hasBody, + }, + timestamp, + wallTime: event.wallTimeMs / 1_000, + initiator: { type: 'other' }, + type: 'Fetch', + }) + return + case 'response-received': + session.sendEvent('Network.responseReceived', { + requestId: event.requestId, + loaderId: 'dsh-inspector-loader', + frameId: 'dsh-inspector-host-frame', + timestamp, + type: 'Fetch', + response: { + url: event.url, + status: event.status, + statusText: event.statusText, + headers: cdpHeaders(event.headers), + mimeType: event.mimeType, + connectionReused: false, + connectionId: 0, + encodedDataLength: 0, + securityState: 'neutral', + }, + }) + return + case 'response-data': + session.sendEvent('Network.dataReceived', { + requestId: event.requestId, + timestamp, + dataLength: event.byteLength, + encodedDataLength: event.byteLength, + ...(this.streamedRequests.get(session)?.has(event.requestKey) === true ? { data: event.data } : {}), + }) + return + case 'request-finished': + session.sendEvent('Network.loadingFinished', { + requestId: event.requestId, + timestamp, + encodedDataLength: event.encodedDataLength, + dshInspectorTruncated: event.truncated, + }) + this.stopStreaming(event.requestKey) + return + case 'request-failed': + session.sendEvent('Network.loadingFailed', { + requestId: event.requestId, + timestamp, + type: 'Fetch', + errorText: event.errorText, + canceled: event.canceled, + }) + this.stopStreaming(event.requestKey) + return + default: + return assertNever(event) + } + } + + private stopStreaming(requestKey: string): void { + for (const [session, requests] of this.streamedRequests) { + requests.delete(requestKey) + if (requests.size === 0) this.streamedRequests.delete(session) + } + } +} + +function cdpHeaders(entries: readonly InspectorHeader[]): Record { + const headers: Record = Object.create(null) as Record + for (const [name, value] of entries) { + headers[name] = headers[name] === undefined ? value : `${headers[name]}\n${value}` + } + return headers +} + +function assertNever(value: never): never { + throw new Error(`Unexpected network event: ${JSON.stringify(value)}`) +} diff --git a/packages/experimental/inspector/src/worker/inspection/network-store.ts b/packages/experimental/inspector/src/worker/inspection/network-store.ts new file mode 100644 index 0000000000..1599667503 --- /dev/null +++ b/packages/experimental/inspector/src/worker/inspection/network-store.ts @@ -0,0 +1,429 @@ +/** Worker-owned repository of normalized fetch observations and captured bodies. */ + +import { Buffer } from 'node:buffer' +import { FETCH_TOPICS } from '../../shared/bridge/messages/network.ts' +import type { InspectorHeader } from '../../shared/network/observation.ts' +import { isPlainObject } from '../../shared/json.ts' +import type { InspectorSourceDescriptor } from '../../shared/bridge/messages/observation.ts' +import type { IngestedInspectorRecord, InspectorRecordConsumer } from '../bridge/hub.ts' + +/** Bounded retention policy for observed network requests. */ +export interface NetworkStoreOptions { + readonly maxRetainedRequests: number + readonly maxJournalBytes: number +} + +/** Captured body data returned without a CDP representation. */ +export interface CapturedNetworkBody { + readonly bytes: Uint8Array + readonly truncated: boolean + readonly captureError?: string + readonly complete: boolean +} + +interface NetworkEventBase { + readonly requestKey: string + readonly requestId: string + readonly timestampMs: number +} + +/** Transport-independent changes emitted by the network repository. */ +export type NetworkStoreEvent = + | NetworkEventBase & { + readonly type: 'request-started' + readonly wallTimeMs: number + readonly url: string + readonly method: string + readonly headers: readonly InspectorHeader[] + readonly hasBody: boolean + } + | NetworkEventBase & { + readonly type: 'response-received' + readonly url: string + readonly status: number + readonly statusText: string + readonly headers: readonly InspectorHeader[] + readonly mimeType: string + } + | NetworkEventBase & { + readonly type: 'response-data' + readonly data: string + readonly byteLength: number + } + | NetworkEventBase & { + readonly type: 'request-finished' + readonly encodedDataLength: number + readonly truncated: boolean + } + | NetworkEventBase & { + readonly type: 'request-failed' + readonly errorText: string + readonly canceled: boolean + } + | { readonly type: 'request-evicted'; readonly requestKey: string } + +type ReplayableNetworkEvent = Exclude + +interface CapturedRequest { + readonly key: string + readonly requestId: string + readonly sourceId: string + readonly requestBody: Buffer[] + readonly responseBody: Buffer[] + requestBodyBytes: number + responseBodyBytes: number + requestBodyTruncated: boolean + responseBodyTruncated: boolean + requestCaptureError?: string + responseCaptureError?: string + responseSeen: boolean + completed: boolean +} + +/** Validated Network observation store independent of CDP connection state. */ +export class NetworkStore implements InspectorRecordConsumer { + readonly topics = new Set(FETCH_TOPICS) + private readonly requests = new Map() + private readonly journal: ReplayableNetworkEvent[] = [] + private readonly completed: string[] = [] + private readonly listeners = new Set<(event: NetworkStoreEvent) => void>() + private journalBytes = 0 + + constructor(private readonly options: NetworkStoreOptions) {} + + replace(source: InspectorSourceDescriptor, records: readonly IngestedInspectorRecord[]): void { + this.close(source, 'source state replaced') + this.append(source, records) + } + + append(source: InspectorSourceDescriptor, records: readonly IngestedInspectorRecord[]): void { + for (const record of records) { + if (!this.topics.has(record.topic)) continue + try { + this.ingest(source, record) + } catch { + // A malformed domain payload loses only that observation; later records remain independently useful. + } + } + } + + close(source: InspectorSourceDescriptor, reason: string): void { + for (const request of this.requests.values()) { + if (request.sourceId !== source.sourceId || request.completed) continue + request.completed = true + this.publish({ + type: 'request-failed', + requestKey: request.key, + requestId: request.requestId, + timestampMs: performance.timeOrigin + performance.now(), + errorText: reason, + canceled: true, + }) + this.completed.push(request.key) + } + this.enforceRetention() + } + + /** + * Read retained request lifecycle events. + * @returns Events in observation order. + */ + replay(): readonly ReplayableNetworkEvent[] { + return this.journal + } + + /** + * Subscribe to live request changes and eviction. + * @param listener - Consumer called synchronously after each accepted change. + * @returns A disposer removing the consumer. + */ + subscribe(listener: (event: NetworkStoreEvent) => void): () => void { + this.listeners.add(listener) + return () => { this.listeners.delete(listener) } + } + + /** + * Read one retained request body. + * @param requestId - Public request id assigned by this store. + * @returns Captured bytes and truncation metadata. + */ + requestBody(requestId: unknown): CapturedNetworkBody { + const request = this.requestById(requestId) + return body(request.requestBody, request.requestBodyTruncated, request.requestCaptureError, request.completed) + } + + /** + * Read one retained response body after response headers have arrived. + * @param requestId - Public request id assigned by this store. + * @returns Captured bytes and truncation metadata. + */ + responseBody(requestId: unknown): CapturedNetworkBody { + const request = this.requestById(requestId) + if (!request.responseSeen) throw new Error('response headers have not arrived') + return body(request.responseBody, request.responseBodyTruncated, request.responseCaptureError, request.completed) + } + + /** Release subscribers and all retained request data. */ + dispose(): void { + this.listeners.clear() + this.requests.clear() + this.journal.length = 0 + this.completed.length = 0 + this.journalBytes = 0 + } + + private ingest(source: InspectorSourceDescriptor, record: IngestedInspectorRecord): void { + const payload = requirePayload(record.payload) + const localId = stringField(payload, 'requestId') + const key = `${source.sourceId}:${source.generation}:${localId}` + const timestampMs = source.timeOriginMs + record.monotonicMs + if (record.topic === 'fetch/start') { + if (this.requests.has(key)) throw new Error('fetch observation reused an active request id') + const request: CapturedRequest = { + key, + requestId: key, + sourceId: source.sourceId, + requestBody: [], + responseBody: [], + requestBodyBytes: 0, + responseBodyBytes: 0, + requestBodyTruncated: false, + responseBodyTruncated: false, + responseSeen: false, + completed: false, + } + this.requests.set(key, request) + this.publish({ + type: 'request-started', + requestKey: key, + requestId: request.requestId, + timestampMs, + wallTimeMs: numberField(payload, 'wallTimeMs'), + url: stringField(payload, 'url'), + method: stringField(payload, 'method'), + headers: headerField(payload, 'headers'), + hasBody: booleanField(payload, 'hasBody'), + }) + this.enforceRetention() + return + } + const request = this.requests.get(key) + if (request === undefined) return + switch (record.topic) { + case 'fetch/request-body-chunk': + this.appendBody(request, 'request', stringField(payload, 'data')) + return + case 'fetch/request-body-end': { + request.requestBodyTruncated ||= booleanField(payload, 'truncated') + const captureError = optionalStringField(payload, 'captureError') + if (captureError !== undefined) request.requestCaptureError = captureError + return + } + case 'fetch/response': + request.responseSeen = true + this.publish({ + type: 'response-received', + requestKey: key, + requestId: request.requestId, + timestampMs, + url: stringField(payload, 'url'), + status: numberField(payload, 'status'), + statusText: stringField(payload, 'statusText'), + headers: headerField(payload, 'headers'), + mimeType: stringField(payload, 'mimeType'), + }) + return + case 'fetch/response-body-chunk': { + const data = stringField(payload, 'data') + const byteLength = this.appendBody(request, 'response', data) + this.emit({ type: 'response-data', requestKey: key, requestId: request.requestId, timestampMs, data, byteLength }) + return + } + case 'fetch/end': { + request.responseBodyTruncated ||= booleanField(payload, 'responseBodyTruncated') + const captureError = optionalStringField(payload, 'responseCaptureError') + if (captureError !== undefined) request.responseCaptureError = captureError + this.complete(request, { + type: 'request-finished', + requestKey: key, + requestId: request.requestId, + timestampMs, + encodedDataLength: request.responseBodyBytes, + truncated: request.responseBodyTruncated, + }) + return + } + case 'fetch/error': + this.complete(request, { + type: 'request-failed', + requestKey: key, + requestId: request.requestId, + timestampMs, + errorText: stringField(payload, 'message'), + canceled: booleanField(payload, 'canceled'), + }) + return + default: + return + } + } + + private appendBody(request: CapturedRequest, side: 'request' | 'response', encoded: string): number { + const bytes = decodeBase64(encoded) + this.evictCompletedFor(bytes.byteLength, request.key) + const retained = bytes.subarray(0, Math.max(0, this.options.maxJournalBytes - this.journalBytes)) + if (side === 'request') { + if (retained.byteLength > 0) request.requestBody.push(retained) + request.requestBodyBytes += retained.byteLength + request.requestBodyTruncated ||= retained.byteLength < bytes.byteLength + } else { + if (retained.byteLength > 0) request.responseBody.push(retained) + request.responseBodyBytes += retained.byteLength + request.responseBodyTruncated ||= retained.byteLength < bytes.byteLength + } + this.journalBytes += retained.byteLength + this.enforceRetention() + return bytes.byteLength + } + + private complete(request: CapturedRequest, event: ReplayableNetworkEvent): void { + if (request.completed) return + request.completed = true + this.publish(event) + this.completed.push(request.key) + this.enforceRetention() + } + + private publish(event: ReplayableNetworkEvent): void { + this.journal.push(event) + this.emit(event) + } + + private emit(event: NetworkStoreEvent): void { + for (const listener of [...this.listeners]) { + try { + listener(event) + } catch { + // One presentation adapter cannot interrupt repository ingestion or sibling consumers. + } + } + } + + private enforceRetention(): void { + while (this.requests.size > this.options.maxRetainedRequests || this.journalBytes > this.options.maxJournalBytes) { + const key = this.completed.shift() ?? this.oldestActiveRequestKey() + if (key === undefined) return + const request = this.requests.get(key) + if (request === undefined) continue + if (!request.completed) { + request.completed = true + this.publish({ + type: 'request-failed', + requestKey: request.key, + requestId: request.requestId, + timestampMs: performance.timeOrigin + performance.now(), + errorText: 'Inspector retained-request limit exceeded', + canceled: true, + }) + } + this.evict(request) + } + } + + private oldestActiveRequestKey(): string | undefined { + for (const request of this.requests.values()) { + if (!request.completed) return request.key + } + return undefined + } + + private evictCompletedFor(bytes: number, protectedKey: string): void { + while (this.journalBytes + bytes > this.options.maxJournalBytes) { + const index = this.completed.findIndex(key => key !== protectedKey) + if (index === -1) return + const [key] = this.completed.splice(index, 1) + if (key === undefined) return + const request = this.requests.get(key) + if (request !== undefined) this.evict(request) + } + } + + private evict(request: CapturedRequest): void { + this.journalBytes -= request.requestBodyBytes + request.responseBodyBytes + this.requests.delete(request.key) + for (let index = this.journal.length - 1; index >= 0; index--) { + if (this.journal[index]?.requestKey === request.key) this.journal.splice(index, 1) + } + this.emit({ type: 'request-evicted', requestKey: request.key }) + } + + private requestById(value: unknown): CapturedRequest { + if (typeof value !== 'string') throw new Error('Network requestId must be a string') + const request = [...this.requests.values()].find(candidate => candidate.requestId === value) + if (request === undefined) throw new Error(`No resource with given identifier: ${value}`) + return request + } +} + +function body( + chunks: readonly Buffer[], + truncated: boolean, + captureError: string | undefined, + complete: boolean, +): CapturedNetworkBody { + return { + bytes: Buffer.concat(chunks), + truncated, + complete, + ...(captureError === undefined ? {} : { captureError }), + } +} + +function decodeBase64(value: string): Buffer { + if (value.length === 0 || value.length % 4 !== 0 || !/^(?:[A-Za-z0-9+/]{4})*(?:[A-Za-z0-9+/]{2}==|[A-Za-z0-9+/]{3}=)?$/u.test(value)) { + throw new Error('fetch payload body chunk must be canonical base64') + } + const bytes = Buffer.from(value, 'base64') + if (bytes.toString('base64') !== value) throw new Error('fetch payload body chunk must be canonical base64') + return bytes +} + +function requirePayload(value: unknown): Readonly> { + if (!isPlainObject(value)) throw new Error('fetch payload must be an object') + return value +} + +function stringField(value: Readonly>, name: string): string { + const field = value[name] + if (typeof field !== 'string') throw new Error(`fetch payload ${name} must be a string`) + return field +} + +function optionalStringField(value: Readonly>, name: string): string | undefined { + const field = value[name] + if (field !== undefined && typeof field !== 'string') throw new Error(`fetch payload ${name} must be a string`) + return field +} + +function numberField(value: Readonly>, name: string): number { + const field = value[name] + if (typeof field !== 'number' || !Number.isFinite(field)) throw new Error(`fetch payload ${name} must be finite`) + return field +} + +function booleanField(value: Readonly>, name: string): boolean { + const field = value[name] + if (typeof field !== 'boolean') throw new Error(`fetch payload ${name} must be boolean`) + return field +} + +function headerField(value: Readonly>, name: string): InspectorHeader[] { + const field = value[name] + if (!Array.isArray(field)) throw new Error(`fetch payload ${name} must be a header list`) + return field.map((entry) => { + if (!Array.isArray(entry) || entry.length !== 2 || typeof entry[0] !== 'string' || typeof entry[1] !== 'string') { + throw new Error(`fetch payload ${name} contains an invalid header`) + } + return [entry[0], entry[1]] as const + }) +} diff --git a/packages/experimental/inspector/tests/fetch-observer.host.spec.ts b/packages/experimental/inspector/tests/fetch-observer.host.spec.ts new file mode 100644 index 0000000000..cabeb5a1a5 --- /dev/null +++ b/packages/experimental/inspector/tests/fetch-observer.host.spec.ts @@ -0,0 +1,124 @@ +/** Host fetch observation behavior. */ + +import { afterEach, describe, expect, it, vi } from 'vitest' +import { installFetchObserver, type FetchObserver } from '../src/host/inspection/network.ts' +import type { InspectorRecordInput } from '../src/shared/bridge/messages/observation.ts' +import type { InspectorJsonValue } from '../src/shared/json.ts' + +describe('full fetch observer', () => { + const originalDescriptor = Object.getOwnPropertyDescriptor(globalThis, 'fetch') + let observer: FetchObserver | undefined + + afterEach(async () => { + await observer?.stop() + observer = undefined + if (originalDescriptor === undefined) Reflect.deleteProperty(globalThis, 'fetch') + else Object.defineProperty(globalThis, 'fetch', originalDescriptor) + }) + + it('captures complete URL, headers, request body, response headers, and response body', async () => { + const records: InspectorRecordInput[] = [] + const native = vi.fn(async (request: Request) => { + expect(await request.clone().text()).toBe('secret request body') + return new Response('complete response body', { + status: 201, + statusText: 'Created', + headers: { authorization: 'response secret', 'content-type': 'text/plain' }, + }) + }) + Object.defineProperty(globalThis, 'fetch', { value: native, writable: true, configurable: true }) + observer = installFetchObserver({ + publish(topic: string, payload: InspectorJsonValue, monotonicMs = performance.now()) { + records.push({ topic, payload, monotonicMs }) + }, + }, { maxRequestBodyBytes: 1_024, maxResponseBodyBytes: 1_024, maxChunkBytes: 4 }) + + const response = await fetch('https://example.test/path?token=visible', { + method: 'POST', + headers: { authorization: 'Bearer visible' }, + body: 'secret request body', + }) + expect(await response.text()).toBe('complete response body') + await vi.waitFor(() => { expect(records.some(record => record.topic === 'fetch/end')).toBe(true) }) + + const start = payload(records, 'fetch/start') + expect(start).toMatchObject({ + url: 'https://example.test/path?token=visible', + method: 'POST', + }) + expect(start.headers).toEqual(expect.arrayContaining([['authorization', 'Bearer visible']])) + expect(decodeChunks(records, 'fetch/request-body-chunk')).toBe('secret request body') + const responseRecord = payload(records, 'fetch/response') + expect(responseRecord.status).toBe(201) + expect(responseRecord.headers).toEqual(expect.arrayContaining([['authorization', 'response secret']])) + expect(decodeChunks(records, 'fetch/response-body-chunk')).toBe('complete response body') + expect(payload(records, 'fetch/request-body-end')).toMatchObject({ truncated: false }) + expect(payload(records, 'fetch/end')).toMatchObject({ responseBodyTruncated: false }) + }) + + it('marks bodies truncated without changing the caller response', async () => { + const records: InspectorRecordInput[] = [] + Object.defineProperty(globalThis, 'fetch', { + value: vi.fn(() => Promise.resolve(new Response('response-long'))), + writable: true, + configurable: true, + }) + observer = installFetchObserver({ + publish(topic: string, payload: InspectorJsonValue, monotonicMs = performance.now()) { + records.push({ topic, payload, monotonicMs }) + }, + }, { maxRequestBodyBytes: 4, maxResponseBodyBytes: 4, maxChunkBytes: 2 }) + + const response = await fetch('https://example.test/', { method: 'POST', body: 'request-long' }) + expect(await response.text()).toBe('response-long') + await vi.waitFor(() => { expect(records.some(record => record.topic === 'fetch/end')).toBe(true) }) + + expect(decodeChunks(records, 'fetch/request-body-chunk')).toBe('requ') + expect(payload(records, 'fetch/request-body-end')).toMatchObject({ capturedBytes: 4, truncated: true }) + expect(decodeChunks(records, 'fetch/response-body-chunk')).toBe('resp') + expect(payload(records, 'fetch/end')).toMatchObject({ capturedBytes: 4, responseBodyTruncated: true }) + }) + + it('reports cancellation after response headers as a canceled request', async () => { + const records: InspectorRecordInput[] = [] + Object.defineProperty(globalThis, 'fetch', { + value: vi.fn(async (request: Request) => new Response(new ReadableStream({ + start(controller) { + controller.enqueue(Buffer.from('first')) + request.signal.addEventListener('abort', () => { + controller.error(new DOMException('aborted', 'AbortError')) + }, { once: true }) + }, + }))), + writable: true, + configurable: true, + }) + observer = installFetchObserver({ + publish(topic: string, payload: InspectorJsonValue, monotonicMs = performance.now()) { + records.push({ topic, payload, monotonicMs }) + }, + }, { maxRequestBodyBytes: 1_024, maxResponseBodyBytes: 1_024, maxChunkBytes: 4 }) + const abort = new AbortController() + + const response = await fetch('https://example.test/cancel-body', { signal: abort.signal }) + abort.abort() + await expect(response.text()).rejects.toThrow() + await vi.waitFor(() => { expect(records.some(record => record.topic === 'fetch/error')).toBe(true) }) + + expect(payload(records, 'fetch/error')).toMatchObject({ canceled: true }) + expect(records.some(record => record.topic === 'fetch/end')).toBe(false) + }) +}) + +function payload(records: readonly InspectorRecordInput[], topic: string): Record { + const record = records.find(candidate => candidate.topic === topic) + expect(record).toBeDefined() + return record!.payload as Record +} + +function decodeChunks(records: readonly InspectorRecordInput[], topic: string): string { + return Buffer.concat(records + .filter(record => record.topic === topic) + .map(record => Buffer.from(String((record.payload as Record).data), 'base64'))) + .toString('utf8') +} diff --git a/packages/experimental/inspector/tests/network.host.spec.ts b/packages/experimental/inspector/tests/network.host.spec.ts new file mode 100644 index 0000000000..05273e38e1 --- /dev/null +++ b/packages/experimental/inspector/tests/network.host.spec.ts @@ -0,0 +1,153 @@ +/** Worker-side Network projection behavior. */ + +import { describe, expect, it, vi } from 'vitest' +import { NetworkDomain, type NetworkSink } from '../src/worker/cdp/domains/network/session.ts' +import { NetworkStore } from '../src/worker/inspection/network-store.ts' +import { inspectorId } from '../src/shared/bridge/ids.ts' +import type { InspectorSourceDescriptor } from '../src/shared/bridge/messages/observation.ts' +import type { IngestedInspectorRecord } from '../src/worker/bridge/hub.ts' + +const source: InspectorSourceDescriptor = { + sourceId: inspectorId<'InspectorSourceId'>('host-network', 'sourceId'), + generation: inspectorId<'InspectorSourceGeneration'>('network-generation', 'generation'), + kind: 'host', + label: 'Host', + timeOriginMs: performance.timeOrigin, + capabilities: [], +} + +describe('Inspector Network domain', () => { + it('bounds incomplete bodies and marks the retained prefix truncated', () => { + const sendEvent = vi.fn() + const sink: NetworkSink = { sendEvent } + const store = new NetworkStore({ maxRetainedRequests: 10, maxJournalBytes: 4 }) + const network = new NetworkDomain(store) + network.enable(sink) + store.append(source, requestRecords('first', 'abcdef')) + + const response = network.handle('Network.getResponseBody', { requestId: requestId('first') }, sink) + expect(response).toEqual({ + body: Buffer.from('abcd').toString('base64'), + base64Encoded: true, + dshInspectorTruncated: true, + }) + const dataEvent = sendEvent.mock.calls.find(call => call[0] === 'Network.dataReceived') + expect(dataEvent?.[1]).toMatchObject({ dataLength: 6, encodedDataLength: 6 }) + expect(dataEvent?.[1]).not.toHaveProperty('data') + }) + + it('evicts completed requests before retaining a later body', () => { + const sink: NetworkSink = { sendEvent: vi.fn() } + const store = new NetworkStore({ maxRetainedRequests: 10, maxJournalBytes: 4 }) + const network = new NetworkDomain(store) + store.append(source, requestRecords('first', 'aaaa')) + store.append(source, requestRecords('second', 'bbbb')) + + expect(() => network.handle('Network.getResponseBody', { requestId: requestId('first') }, sink)).toThrow( + 'No resource with given identifier', + ) + expect(network.handle('Network.getResponseBody', { requestId: requestId('second') }, sink)).toEqual({ + body: Buffer.from('bbbb').toString('base64'), + base64Encoded: true, + dshInspectorTruncated: false, + }) + }) + + it('streams later response chunks only to CDP sessions that opted in', () => { + const firstSend = vi.fn() + const secondSend = vi.fn() + const first: NetworkSink = { sendEvent: firstSend } + const second: NetworkSink = { sendEvent: secondSend } + const store = new NetworkStore({ maxRetainedRequests: 10, maxJournalBytes: 1_024 }) + const network = new NetworkDomain(store) + network.enable(first) + network.enable(second) + const records = requestRecords('stream', 'data: first\n\n') + store.append(source, records.slice(0, 2)) + + expect(network.handle('Network.streamResourceContent', { requestId: requestId('stream') }, first)).toEqual({ + bufferedData: '', + }) + store.append(source, records.slice(2, 3)) + + const firstData = firstSend.mock.calls.findLast(call => call[0] === 'Network.dataReceived') + const secondData = secondSend.mock.calls.findLast(call => call[0] === 'Network.dataReceived') + expect(firstData?.[1]).toMatchObject({ data: Buffer.from('data: first\n\n').toString('base64') }) + expect(secondData?.[1]).not.toHaveProperty('data') + expect(network.handle('Network.streamResourceContent', { requestId: requestId('stream') }, second)).toEqual({ + bufferedData: Buffer.from('data: first\n\n').toString('base64'), + }) + + const later = Buffer.from('data: second\n\n').toString('base64') + store.append(source, [{ + sequence: 4, + monotonicMs: 4, + topic: 'fetch/response-body-chunk', + payload: { requestId: 'stream', data: later }, + }]) + expect(firstSend.mock.calls.findLast(call => call[0] === 'Network.dataReceived')?.[1]).toMatchObject({ data: later }) + expect(secondSend.mock.calls.findLast(call => call[0] === 'Network.dataReceived')?.[1]).toMatchObject({ data: later }) + }) + + it('bounds active request metadata and does not retain per-chunk events for replay', () => { + const firstSend = vi.fn() + const store = new NetworkStore({ maxRetainedRequests: 1, maxJournalBytes: 1_024 }) + const network = new NetworkDomain(store) + network.enable({ sendEvent: firstSend }) + store.append(source, requestRecords('active-first', 'first').slice(0, 1)) + store.append(source, requestRecords('active-second', 'second').slice(0, 1)) + + expect(firstSend).toHaveBeenCalledWith('Network.loadingFailed', expect.objectContaining({ + requestId: requestId('active-first'), + canceled: true, + })) + expect(() => network.handle( + 'Network.getRequestPostData', + { requestId: requestId('active-first') }, + { sendEvent: vi.fn() }, + )).toThrow('No resource with given identifier') + expect(() => { store.append(source, requestRecords('active-first', 'first').slice(1)) }).not.toThrow() + + store.append(source, requestRecords('active-second', 'second').slice(1)) + const replay = vi.fn() + network.enable({ sendEvent: replay }) + expect(replay.mock.calls.some(call => call[0] === 'Network.dataReceived')).toBe(false) + expect(replay).toHaveBeenCalledTimes(3) + expect(replay).toHaveBeenNthCalledWith(1, 'Network.requestWillBeSent', expect.any(Object)) + expect(replay).toHaveBeenNthCalledWith(2, 'Network.responseReceived', expect.any(Object)) + expect(replay).toHaveBeenNthCalledWith(3, 'Network.loadingFinished', expect.any(Object)) + }) +}) + +function requestRecords(localId: string, body: string): IngestedInspectorRecord[] { + return [ + { + sequence: 1, + monotonicMs: 1, + topic: 'fetch/start', + payload: { requestId: localId, url: 'https://example.test/', method: 'GET', headers: [], hasBody: false, wallTimeMs: 1 }, + }, + { + sequence: 2, + monotonicMs: 2, + topic: 'fetch/response', + payload: { requestId: localId, url: 'https://example.test/', status: 200, statusText: 'OK', headers: [], mimeType: 'text/plain' }, + }, + { + sequence: 3, + monotonicMs: 3, + topic: 'fetch/response-body-chunk', + payload: { requestId: localId, data: Buffer.from(body).toString('base64') }, + }, + { + sequence: 4, + monotonicMs: 4, + topic: 'fetch/end', + payload: { requestId: localId, capturedBytes: body.length, responseBodyTruncated: false }, + }, + ] +} + +function requestId(localId: string): string { + return `${source.sourceId}:${source.generation}:${localId}` +}