diff --git a/.github/dictionary.txt b/.github/dictionary.txt index fe399a24f..6c51df219 100644 --- a/.github/dictionary.txt +++ b/.github/dictionary.txt @@ -1,3 +1,5 @@ +backpressures dont +dups fleek republisher diff --git a/packages/block-brokers/.aegir.js b/packages/block-brokers/.aegir.js index 59033d126..be783040e 100644 --- a/packages/block-brokers/.aegir.js +++ b/packages/block-brokers/.aegir.js @@ -1,6 +1,47 @@ +import { CarWriter } from '@ipld/car' import cors from 'cors' +import { CID } from 'multiformats/cid' +import * as raw from 'multiformats/codecs/raw' +import { sha256 } from 'multiformats/hashes/sha2' import polka from 'polka' +/** + * Fixed raw blocks used by the CAR-stream session tests. The test recomputes + * these from the same bytes, so the CIDs match without passing them around. + */ +const CAR_BLOCK_BYTES = [ + Uint8Array.from([1, 2, 3, 4]), + Uint8Array.from([5, 6, 7, 8]), + Uint8Array.from([9, 10, 11, 12]) +] + +async function makeRawBlock (bytes) { + return { cid: CID.createV1(raw.code, await sha256.digest(bytes)), bytes } +} + +async function buildCar (blocks, roots) { + const { writer, out } = CarWriter.create(roots) + const chunks = [] + const collecting = (async () => { + for await (const chunk of out) { + chunks.push(chunk) + } + })() + for (const block of blocks) { + await writer.put(block) + } + await writer.close() + await collecting + const total = chunks.reduce((n, c) => n + c.length, 0) + const car = new Uint8Array(total) + let offset = 0 + for (const chunk of chunks) { + car.set(chunk, offset) + offset += chunk.length + } + return car +} + /** * Middleware to log requests @@ -99,18 +140,59 @@ const options = { await badGateway.listen() const { port: badGatewayPort } = badGateway.server.address() + // Gateway that serves a CAR of a small DAG, plus the raw blocks. The CAR + // contains the root and the first leaf but NOT the second leaf, so the + // session must gap-fill that one over ?format=raw. + const blocks = await Promise.all(CAR_BLOCK_BYTES.map(makeRawBlock)) + const carBytes = await buildCar([blocks[0], blocks[1]], [blocks[0].cid]) + const blockByCid = new Map(blocks.map(b => [b.cid.toString(), b.bytes])) + let carRequests = 0 + + const carGateway = polka({ port: 0, host: '127.0.0.1' }) + carGateway.use(cors()) + carGateway.all('/car-requests', (req, res) => { + res.writeHead(200, { 'content-type': 'application/json' }) + res.end(JSON.stringify({ carRequests })) + }) + carGateway.all('/ipfs/:cid', (req, res) => { + // no-store so the browser HTTP cache does not satisfy getCar's + // `cache: 'force-cache'` across tests, which would hide CAR requests + // from the counter below. + const format = new URL(req.url, 'http://localhost').searchParams.get('format') + if (format === 'car') { + carRequests++ + res.writeHead(200, { 'content-type': 'application/vnd.ipld.car', 'cache-control': 'no-store' }) + res.end(carBytes) + return + } + const bytes = blockByCid.get(req.params.cid) + if (bytes == null) { + res.writeHead(404) + res.end() + return + } + res.writeHead(200, { 'content-type': 'application/vnd.ipld.raw', 'content-length': bytes.length, 'cache-control': 'no-store' }) + res.end(bytes) + }) + + await carGateway.listen() + const { port: carGatewayPort } = carGateway.server.address() + return { goodGateway, badGateway, + carGateway, env: { TRUSTLESS_GATEWAY: `http://127.0.0.1:${goodGatewayPort}`, - BAD_TRUSTLESS_GATEWAY: `http://127.0.0.1:${badGatewayPort}` + BAD_TRUSTLESS_GATEWAY: `http://127.0.0.1:${badGatewayPort}`, + CAR_GATEWAY: `http://127.0.0.1:${carGatewayPort}` } } }, async after (options, before) { await before.goodGateway.server.close() await before.badGateway.server.close() + await before.carGateway.server.close() } } } diff --git a/packages/block-brokers/package.json b/packages/block-brokers/package.json index dbf9d13bc..9febcc5d0 100644 --- a/packages/block-brokers/package.json +++ b/packages/block-brokers/package.json @@ -51,6 +51,7 @@ "@helia/bitswap": "^3.2.3", "@helia/interface": "^6.2.1", "@helia/utils": "^2.5.2", + "@ipld/car": "^5.4.3", "@libp2p/interface": "^3.2.0", "@libp2p/peer-id": "^6.0.6", "@libp2p/utils": "^7.0.15", diff --git a/packages/block-brokers/src/trustless-gateway/broker.ts b/packages/block-brokers/src/trustless-gateway/broker.ts index ae2db6a04..07e312b6e 100644 --- a/packages/block-brokers/src/trustless-gateway/broker.ts +++ b/packages/block-brokers/src/trustless-gateway/broker.ts @@ -1,3 +1,4 @@ +import { createTrustlessGatewayCarSession } from './car-session.ts' import { DEFAULT_ALLOW_INSECURE, DEFAULT_ALLOW_LOCAL } from './index.ts' import { createTrustlessGatewaySession } from './session.ts' import { findHttpGatewayProviders } from './utils.ts' @@ -45,6 +46,7 @@ export class TrustlessGatewayBlockBroker implements BlockBroker = {}): Promise { @@ -99,7 +102,8 @@ export class TrustlessGatewayBlockBroker implements BlockBroker { - return createTrustlessGatewaySession({ + const create = this.carStreamSessions ? createTrustlessGatewayCarSession : createTrustlessGatewaySession + return create({ logger: this.logger, routing: this.routing }, { diff --git a/packages/block-brokers/src/trustless-gateway/car-session.ts b/packages/block-brokers/src/trustless-gateway/car-session.ts new file mode 100644 index 000000000..40d112d50 --- /dev/null +++ b/packages/block-brokers/src/trustless-gateway/car-session.ts @@ -0,0 +1,434 @@ +import { CarBlockIterator } from '@ipld/car' +import { isPeerId } from '@libp2p/interface' +import { multiaddrToUri } from '@multiformats/multiaddr-to-uri' +import { base64 } from 'multiformats/bases/base64' +import { DEFAULT_ALLOW_INSECURE, DEFAULT_ALLOW_LOCAL, DEFAULT_MAX_SIZE } from './index.ts' +import { TrustlessGateway } from './trustless-gateway.ts' +import { filterNonHTTPMultiaddrs, findHttpGatewayProviders } from './utils.ts' +import type { CreateTrustlessGatewaySessionOptions } from './broker.ts' +import type { TrustlessGatewayGetBlockProgressEvents } from './index.ts' +import type { TrustlessGatewaySessionComponents } from './session.ts' +import type { TransformRequestInit } from './trustless-gateway.ts' +import type { BlockRetrievalOptions, Routing, SessionBlockBroker } from '@helia/interface' +import type { AbortOptions, ComponentLogger, Logger, PeerId } from '@libp2p/interface' +import type { Multiaddr } from '@multiformats/multiaddr' +import type { CID } from 'multiformats/cid' + +/** + * Default soft-to-hard cap on blocks parsed from the CAR stream but not yet + * requested by the consumer. The gateway emits the DAG in the same depth-first + * order a walk asks for it, so in practice the buffer stays near-empty; the cap + * bounds the pathological case where the stream order diverges from the walk. + */ +export const DEFAULT_MAX_BUFFERED_BLOCKS = 128 + +/** + * How long the pump may sit paused on a full buffer with no waiter before the + * CAR fetch is torn down. This bounds the resource an abandoned session holds + * (`SessionBlockBroker` has no close hook — see + * https://github.com/ipfs/helia/issues/1051), after which later retrieves + * recover blocks over per-block raw fetches. + */ +export const DEFAULT_STREAM_IDLE_TIMEOUT_MS = 60_000 + +interface Waiter { + resolve(bytes: Uint8Array | null): void + reject(err: unknown): void +} + +/** + * Index key for a block: its multihash bytes, so CIDv0/v1 forms of the same + * block collapse to one entry. + */ +function blockKey (cid: CID): string { + return base64.encode(cid.multihash.bytes) +} + +/** + * A session that fetches the whole DAG over a single `?format=car` request to + * one trustless gateway, then serves each block the consumer asks for from the + * streamed CAR. This collapses the one-request-per-block cost of the default + * session to a single request for many-small-block DAGs (a sharded HAMT + * directory is hundreds of tiny blocks). + * + * The CAR is untrusted: every block is verified through the per-retrieve + * `validateFn` before it is returned, exactly as the default broker does. A + * block the CAR omits, or one whose bytes fail that check, falls through to a + * single `?format=raw` fetch for that block; if the raw fetch also fails or + * mismatches, the retrieve rejects. + * + * The session is scoped to the first CID it is asked for, which it treats as + * the CAR root. This mirrors how `AbstractSession` scopes provider discovery to + * the first retrieved CID, and is necessary because `createSession` is not + * passed the root (see https://github.com/ipfs/helia/issues/1051). It does not + * extend `AbstractSession`: that class models per-block, per-provider racing, + * which is orthogonal to a single root-scoped stream. + */ +class TrustlessGatewayCarSession implements SessionBlockBroker { + public readonly name = 'trustless-gateway-car-session' + private readonly routing: Routing + private readonly logger: ComponentLogger + private readonly log: Logger + private readonly allowInsecure: boolean + private readonly allowLocal: boolean + private readonly transformRequestInit?: TransformRequestInit + private readonly maxBuffered: number + private readonly idleTimeoutMs: number + + /** Session providers passed by the caller, tried before routing discovery. */ + readonly #initialProviders: Array + #root: CID | null = null + /** The gateway serving the CAR stream, used for raw gap-fill too. */ + #gateway: TrustlessGateway | null = null + /** Extra gateways added via `addPeer`, tried for gap-fill if the primary fails. */ + readonly #extraGateways: TrustlessGateway[] = [] + /** + * Signal passed to the CAR fetch and gateway discovery. It is deliberately + * NOT tied to per-retrieve signals (`raceBlockRetrievers` aborts those after + * every block). With no session-close hook on `SessionBlockBroker` there is + * no trigger to abort it today; an abandoned session's fetch is instead + * bounded by the buffer cap, which backpressures the stream once full. A + * close hook (see https://github.com/ipfs/helia/issues/1051) would let this + * cancel the stream eagerly. + */ + readonly #controller = new AbortController() + /** Parsed-but-not-yet-requested blocks, keyed by multihash. Hard-bounded by `maxBuffered`. */ + readonly #arrived = new Map() + /** One-shot waiters per multihash; resolved with bytes on arrival, null on stream end. */ + readonly #waiters = new Map() + #streamEnded = false + #streamError: unknown = null + #wakePump: (() => void) | null = null + #gapFillCount = 0 + + constructor (components: TrustlessGatewaySessionComponents, init: CreateTrustlessGatewaySessionOptions) { + this.log = components.logger.forComponent('helia:trustless-gateway-car-session') + this.logger = components.logger + this.routing = components.routing + this.allowInsecure = init.allowInsecure ?? DEFAULT_ALLOW_INSECURE + this.allowLocal = init.allowLocal ?? DEFAULT_ALLOW_LOCAL + this.transformRequestInit = init.transformRequestInit + this.maxBuffered = DEFAULT_MAX_BUFFERED_BLOCKS + this.idleTimeoutMs = DEFAULT_STREAM_IDLE_TIMEOUT_MS + this.#initialProviders = [...(init.providers ?? [])] + } + + /** + * The number of blocks served by the raw fallback rather than the CAR stream. + * A non-zero count means the gateway's CAR was incomplete for this root. + */ + get gapFillCount (): number { + return this.#gapFillCount + } + + async retrieve (cid: CID, options: BlockRetrievalOptions = {}): Promise { + options.signal?.throwIfAborted() + + // The CAR stream must outlive any single retrieve. `raceBlockRetrievers` + // aborts the per-retrieve signal in `finally` after every successful block + // (helia/utils storage.ts), so the stream is deliberately NOT tied to it — + // it runs on its own `#controller`. A parked retrieve still honours its own + // signal (see `#awaitBlock`). With no session-close hook on + // `SessionBlockBroker` (https://github.com/ipfs/helia/issues/1051), an + // abandoned stream is bounded by the buffer cap and torn down after an idle + // timeout rather than cancelled eagerly. + const key = blockKey(cid) + + if (this.#root == null) { + this.#root = cid + void this.#startStream(cid, options) + } + + const buffered = this.#arrived.get(key) + if (buffered != null) { + this.#arrived.delete(key) + this.#resumePump() + return this.#serve(cid, buffered, options) + } + + // streamEnded/streamError → the block did not arrive; #controller.aborted → + // the stream was torn down (idle timeout). All three mean: recover via raw. + if (this.#streamEnded || this.#streamError != null || this.#controller.signal.aborted) { + return this.#gapFill(cid, options) + } + + const delivered = await this.#awaitBlock(key, options.signal) + if (delivered != null) { + return this.#serve(cid, delivered, options) + } + + return this.#gapFill(cid, options) + } + + async addPeer (peer: PeerId | Multiaddr | Multiaddr[], options?: AbortOptions): Promise { + const gateway = await this.#toGateway(peer) + if (gateway == null) { + return + } + if (this.#gateway?.url.toString() === gateway.url.toString()) { + return + } + if (this.#extraGateways.some(g => g.url.toString() === gateway.url.toString())) { + return + } + this.#extraGateways.push(gateway) + } + + /** + * Verify a CAR-supplied block through the consumer's `validateFn` and honour + * the per-retrieve `maxSize`. On a size or validation miss the block is not + * trusted/usable, so fall back to a verified, size-limited raw fetch. + */ + async #serve (cid: CID, bytes: Uint8Array, options: BlockRetrievalOptions): Promise { + if (bytes.byteLength > (options.maxSize ?? DEFAULT_MAX_SIZE)) { + // the caller asked for a smaller cap than this CAR block; let getRawBlock + // enforce the caller's maxSize + return this.#gapFill(cid, options) + } + if (options.validateFn == null) { + return bytes + } + try { + await options.validateFn(bytes) + return bytes + } catch (err) { + // The CAR served bytes that do not hash to the CID: this gateway is + // serving bad data, so drop its reliability and recover over raw. + this.#gateway?.incrementInvalidBlocks() + this.log.error('block %c from the CAR stream failed validation, falling back to a raw fetch - %e', cid, err) + return this.#gapFill(cid, options) + } + } + + async #gapFill (cid: CID, options: BlockRetrievalOptions): Promise { + // Dedupe by URL: `addPeer` can land a gateway in `#extraGateways` before + // `#startStream` resolves the same one as `#gateway`. + const seen = new Set() + const gateways = [this.#gateway, ...this.#extraGateways].filter((g): g is TrustlessGateway => { + if (g == null || seen.has(g.url.toString())) { + return false + } + seen.add(g.url.toString()) + return true + }) + if (gateways.length === 0) { + throw new Error(`no gateway available to fetch block ${cid}`) + } + + const errors: Error[] = [] + for (const gateway of gateways) { + try { + const block = await gateway.getRawBlock(cid, options) + await options.validateFn?.(block) + // Count only blocks actually served by the raw fallback: a non-zero + // count means the gateway's CAR was incomplete for this root. + this.#gapFillCount++ + return block + } catch (err) { + this.log.error('raw fallback for %c from %s failed - %e', cid, gateway.url, err) + errors.push(err instanceof Error ? err : new Error(String(err))) + if (options.signal?.aborted === true) { + break + } + } + } + + throw new AggregateError(errors, `Unable to fetch block ${cid} from the CAR stream or any gateway`) + } + + #awaitBlock (key: string, signal?: AbortSignal): Promise { + return new Promise((resolve, reject) => { + if (this.#controller.signal.aborted) { + reject(this.#controller.signal.reason ?? new Error('session aborted')) + return + } + if (signal?.aborted === true) { + reject(signal.reason ?? new Error('aborted')) + return + } + + const waiter: Waiter = { resolve, reject } + + if (signal != null) { + const onAbort = (): void => { + const list = this.#waiters.get(key) + const i = list?.indexOf(waiter) ?? -1 + if (list != null && i !== -1) { + list.splice(i, 1) + if (list.length === 0) { + this.#waiters.delete(key) + } + } + reject(signal.reason ?? new Error('aborted')) + } + signal.addEventListener('abort', onAbort, { once: true }) + waiter.resolve = (bytes) => { + signal.removeEventListener('abort', onAbort) + resolve(bytes) + } + waiter.reject = (err) => { + signal.removeEventListener('abort', onAbort) + reject(err) + } + } + + const list = this.#waiters.get(key) + if (list == null) { + this.#waiters.set(key, [waiter]) + } else { + list.push(waiter) + } + // A new waiter may need a block past a full buffer — keep the pump moving. + this.#resumePump() + }) + } + + /** + * Resolve every waiter for `key` with `value` and remove them. Returns false + * if there were none. `null` tells the parked retrieve the stream will not + * serve this block, so it gap-fills. + */ + #deliver (key: string, value: Uint8Array | null): boolean { + const list = this.#waiters.get(key) + if (list == null || list.length === 0) { + return false + } + this.#waiters.delete(key) + for (const { resolve } of list) { + resolve(value) + } + return true + } + + /** Release every parked waiter so its retrieve gap-fills. Called when the stream ends. */ + #flushWaiters (): void { + for (const [, list] of this.#waiters) { + for (const { resolve } of list) { + resolve(null) + } + } + this.#waiters.clear() + } + + #resumePump (): void { + if (this.#wakePump != null) { + const wake = this.#wakePump + this.#wakePump = null + wake() + } + } + + async #startStream (root: CID, options: BlockRetrievalOptions): Promise { + let body: ReadableStream | null = null + + try { + for await (const gateway of this.#candidateGateways(root, options)) { + this.#gateway ??= gateway + try { + body = await gateway.getCar(root, { signal: this.#controller.signal, onProgress: options.onProgress }) + this.#gateway = gateway + break + } catch (err) { + this.log.error('failed to open a CAR stream from %s - %e', gateway.url, err) + } + } + + if (body == null) { + this.#streamError = new Error(`no gateway served a CAR for ${root}`) + return + } + + for await (const { cid, bytes } of await CarBlockIterator.fromIterable(body)) { + const key = blockKey(cid) + + // Drop blocks over the raw-path ceiling (helia#790) so they are never + // indexed or served. Note `@ipld/car` has already read the full + // gateway-declared block into memory by this point, so this bounds + // retained/served bytes, not the transient per-block read — a streaming + // cap would need support from the CAR reader. Any waiter for a dropped + // block is released to null so its retrieve gap-fills immediately + // (through the size-limited `getRawBlock`) rather than parking until + // end-of-stream; leaving it parked would also disable the backpressure + // pause below and let a huge CAR drain after the needed block was gone. + if (bytes.byteLength > DEFAULT_MAX_SIZE) { + this.log.error('dropping CAR block %c: %d bytes exceeds the %d limit', cid, bytes.byteLength, DEFAULT_MAX_SIZE) + this.#deliver(key, null) + continue + } + + if (!this.#deliver(key, bytes) && !this.#arrived.has(key) && this.#arrived.size < this.maxBuffered) { + this.#arrived.set(key, bytes) + } + // else: a non-waited block past the cap is dropped, recovered via + // gap-fill if the consumer asks for it later. This makes the cap a hard + // bound under stream/walk order divergence. + + // Pause only while the buffer is full AND nothing is waiting; an + // outstanding waiter means the consumer needs a block still ahead in the + // stream, so keep reading (dropping overflow) to reach it. If the pause + // outlasts the idle timeout (the consumer has stopped), tear the fetch + // down so it does not hold a socket open forever; later retrieves + // gap-fill over raw. + while (this.#arrived.size >= this.maxBuffered && this.#waiters.size === 0 && !this.#controller.signal.aborted) { + const resumed = await new Promise(resolve => { + const timer = setTimeout(() => { + this.#wakePump = null + resolve(false) + }, this.idleTimeoutMs) + this.#wakePump = () => { + clearTimeout(timer) + resolve(true) + } + }) + if (!resumed) { + this.#controller.abort(new Error('CAR stream idle timeout')) + break + } + } + } + } catch (err) { + this.#streamError = err + } finally { + this.#streamEnded = true + this.#flushWaiters() + } + } + + /** + * Gateways to try for the CAR stream: any explicitly-passed session providers + * first (so a caller with known gateways skips routing), then routing + * discovery. Mirrors how `AbstractSession` seeds from `init.providers`. + */ + async * #candidateGateways (root: CID, options: BlockRetrievalOptions): AsyncGenerator { + for (const provider of this.#initialProviders) { + const gateway = await this.#toGateway(provider) + if (gateway != null) { + yield gateway + } + } + yield * findHttpGatewayProviders(root, this.routing, this.logger, this.allowInsecure, this.allowLocal, { + signal: this.#controller.signal, + transformRequestInit: this.transformRequestInit, + onProgress: options.onProgress + }) + } + + async #toGateway (peer: PeerId | Multiaddr | Multiaddr[]): Promise { + if (isPeerId(peer)) { + return + } + const httpAddresses = filterNonHTTPMultiaddrs(Array.isArray(peer) ? peer : [peer], this.allowInsecure, this.allowLocal) + if (httpAddresses.length === 0) { + return + } + return new TrustlessGateway(multiaddrToUri(httpAddresses[0]), { + logger: this.logger, + transformRequestInit: this.transformRequestInit, + routing: 'manual' + }) + } +} + +export function createTrustlessGatewayCarSession (components: TrustlessGatewaySessionComponents, init: CreateTrustlessGatewaySessionOptions): TrustlessGatewayCarSession { + return new TrustlessGatewayCarSession(components, init) +} diff --git a/packages/block-brokers/src/trustless-gateway/index.ts b/packages/block-brokers/src/trustless-gateway/index.ts index 8ce36d6fc..c0f58c953 100644 --- a/packages/block-brokers/src/trustless-gateway/index.ts +++ b/packages/block-brokers/src/trustless-gateway/index.ts @@ -62,6 +62,18 @@ export interface TrustlessGatewayBlockBrokerInit { * Provide a function that will be called before querying trustless-gateways. This lets you modify the fetch options to pass custom headers or other necessary things. */ transformRequestInit?: TransformRequestInit + + /** + * When true, sessions created by this broker fetch the whole DAG over a single + * `?format=car` request to one gateway and serve each block from the stream, + * instead of one `?format=raw` request per block. This trades the default + * session's multi-gateway per-block racing for a single request, which is much + * faster for many-small-block DAGs. Blocks the CAR omits or that fail + * validation fall back to a single `?format=raw` fetch. + * + * @default false + */ + carStreamSessions?: boolean } export interface TrustlessGatewayBlockBrokerComponents { diff --git a/packages/block-brokers/src/trustless-gateway/trustless-gateway.ts b/packages/block-brokers/src/trustless-gateway/trustless-gateway.ts index 851cbed41..29f8ec0f2 100644 --- a/packages/block-brokers/src/trustless-gateway/trustless-gateway.ts +++ b/packages/block-brokers/src/trustless-gateway/trustless-gateway.ts @@ -224,6 +224,83 @@ HTTP/1.1 %d %s } } + /** + * Open a streaming CAR of the DAG rooted at `cid` from `this.url`, following + * the trustless gateway spec (`?format=car&dag-scope=all`). The query is + * pinned to a deterministic, deduplicated, depth-first CARv1 so the blocks + * arrive in the order a DAG walk asks for them. The response body is returned + * unconsumed for the caller to parse and verify block-by-block — unlike + * `getRawBlock` it is not size-limited here, since a CAR spans the whole DAG; + * the caller is responsible for verifying every block against its CID. + */ + async getCar (cid: CID, options: GetRawBlockOptions = {}): Promise> { + const gwUrl = new URL(this.url.toString()) + gwUrl.pathname = `/ipfs/${cid.toString()}` + gwUrl.search = '?format=car&dag-scope=all&car-version=1&car-order=dfs&car-dups=n' + + if (options.signal?.aborted === true) { + throw new Error(`Signal to fetch CAR for CID ${cid} from gateway ${this.url} was aborted prior to fetch`) + } + + try { + this.#attempts++ + const defaultReqInit: RequestInit = { + signal: options.signal, + headers: { + Accept: 'application/vnd.ipld.car' + }, + cache: 'force-cache' + } + + const reqInit: RequestInit = this.transformRequestInit != null ? await this.transformRequestInit(defaultReqInit) : defaultReqInit + + options.onProgress?.(new CustomProgressEvent('helia:block-broker:connect', { + broker: 'trustless-gateway', + type: 'connect', + provider: this.peer, + cid + })) + + const res = await fetch(gwUrl.toString(), reqInit) + + if (!res.ok) { + this.#errors++ + throw new Error(`Unable to fetch CAR for CID ${cid} from gateway ${this.url}, received ${res.status} ${res.statusText}`) + } + + if (res.body == null) { + this.#errors++ + throw new Error(`Unable to fetch CAR for CID ${cid} from gateway ${this.url}, response had no body`) + } + + options.onProgress?.(new CustomProgressEvent('helia:block-broker:connected', { + broker: 'trustless-gateway', + type: 'connected', + provider: this.peer, + address: uriToMultiaddr(gwUrl.toString()), + cid + })) + + options.onProgress?.(new CustomProgressEvent('helia:block-broker:request-block', { + broker: 'trustless-gateway', + type: 'request-block', + provider: this.peer, + cid + })) + + this.#successes++ + return res.body + } catch (cause: any) { + // @ts-expect-error - TS thinks signal?.aborted can only be false now + // because it was checked for true above. + if (options.signal?.aborted === true) { + throw new Error(`Fetching CAR for CID ${cid} from gateway ${this.url} was aborted`) + } + this.#errors++ + throw new Error(`Unable to fetch CAR for CID ${cid} - ${cause.message}`) + } + } + /** * Encapsulate the logic for determining whether a gateway is considered * reliable, for prioritization. This is based on the number of successful attempts made diff --git a/packages/block-brokers/test/trustless-gateway-car-session.spec.ts b/packages/block-brokers/test/trustless-gateway-car-session.spec.ts new file mode 100644 index 000000000..b8e56f3a8 --- /dev/null +++ b/packages/block-brokers/test/trustless-gateway-car-session.spec.ts @@ -0,0 +1,161 @@ +import { generateKeyPair } from '@libp2p/crypto/keys' +import { defaultLogger } from '@libp2p/logger' +import { peerIdFromPrivateKey } from '@libp2p/peer-id' +import { uriToMultiaddr } from '@multiformats/uri-to-multiaddr' +import { expect } from 'aegir/chai' +import { CID } from 'multiformats/cid' +import * as raw from 'multiformats/codecs/raw' +import { sha256 } from 'multiformats/hashes/sha2' +import { stubInterface } from 'sinon-ts' +import { trustlessGateway } from '../src/index.ts' +import { createTrustlessGatewayCarSession } from '../src/trustless-gateway/car-session.ts' +import type { Routing } from '@helia/interface' +import type { ComponentLogger } from '@libp2p/interface' +import type { StubbedInstance } from 'sinon-ts' + +// The same bytes the CAR test gateway in .aegir.js serves, so the CIDs match. +const BLOCK_BYTES = [ + Uint8Array.from([1, 2, 3, 4]), + Uint8Array.from([5, 6, 7, 8]), + Uint8Array.from([9, 10, 11, 12]) +] + +async function rawBlock (bytes: Uint8Array): Promise<{ cid: CID, bytes: Uint8Array }> { + return { cid: CID.createV1(raw.code, await sha256.digest(bytes)), bytes } +} + +/** A validateFn that checks the bytes hash to the requested CID, like helia's own. */ +function validateFn (cid: CID): (bytes: Uint8Array) => Promise { + return async (bytes) => { + const actual = CID.createV1(raw.code, await sha256.digest(bytes)) + if (!actual.equals(cid)) { + throw new Error(`hash mismatch for ${cid}`) + } + } +} + +async function carRequestCount (): Promise { + const res = await fetch(`${process.env.CAR_GATEWAY ?? ''}/car-requests`) + const body = await res.json() + return body.carRequests +} + +describe('trustless-gateway CAR session', () => { + let components: { logger: ComponentLogger, routing: StubbedInstance> } + let root: { cid: CID, bytes: Uint8Array } + let inCar: { cid: CID, bytes: Uint8Array } + let missing: { cid: CID, bytes: Uint8Array } + + beforeEach(async () => { + components = { + logger: defaultLogger(), + routing: stubInterface() + } + + ;[root, inCar, missing] = await Promise.all(BLOCK_BYTES.map(rawBlock)) + + // every session discovers the CAR gateway + components.routing.findProviders.callsFake(async function * () { + yield { + id: peerIdFromPrivateKey(await generateKeyPair('Ed25519')), + multiaddrs: [uriToMultiaddr(process.env.CAR_GATEWAY ?? '')], + routing: 'test-routing' + } + }) + }) + + it('serves the root and an in-CAR block from a single CAR request', async () => { + const session = createTrustlessGatewayCarSession(components, { allowInsecure: true, allowLocal: true }) + const before = await carRequestCount() + + await expect(session.retrieve(root.cid, { validateFn: validateFn(root.cid) })).to.eventually.deep.equal(root.bytes) + await expect(session.retrieve(inCar.cid, { validateFn: validateFn(inCar.cid) })).to.eventually.deep.equal(inCar.bytes) + + expect(await carRequestCount() - before).to.equal(1) + expect(session.gapFillCount).to.equal(0) + }) + + it('gap-fills a block missing from the CAR over a raw request', async () => { + const session = createTrustlessGatewayCarSession(components, { allowInsecure: true, allowLocal: true }) + + // open the stream on the root, then ask for the block the CAR omits + await session.retrieve(root.cid, { validateFn: validateFn(root.cid) }) + await expect(session.retrieve(missing.cid, { validateFn: validateFn(missing.cid) })).to.eventually.deep.equal(missing.bytes) + + expect(session.gapFillCount).to.equal(1) + }) + + it('gap-fills when a CAR block fails validation', async () => { + const session = createTrustlessGatewayCarSession(components, { allowInsecure: true, allowLocal: true }) + await session.retrieve(root.cid, { validateFn: validateFn(root.cid) }) + + // force the in-CAR block to fail validation once, so it falls back to raw + let calls = 0 + const flakyValidate = async (bytes: Uint8Array): Promise => { + calls++ + if (calls === 1) { + throw new Error('synthetic validation failure') + } + await validateFn(inCar.cid)(bytes) + } + + await expect(session.retrieve(inCar.cid, { validateFn: flakyValidate })).to.eventually.deep.equal(inCar.bytes) + expect(session.gapFillCount).to.equal(1) + }) + + it('rejects when the block is in neither the CAR nor a raw response', async () => { + const session = createTrustlessGatewayCarSession(components, { allowInsecure: true, allowLocal: true }) + await session.retrieve(root.cid, { validateFn: validateFn(root.cid) }) + + const absent = CID.createV1(raw.code, await sha256.digest(Uint8Array.from([99, 98, 97, 96]))) + await expect(session.retrieve(absent, { validateFn: validateFn(absent) })).to.eventually.be.rejected() + }) + + // raceBlockRetrievers aborts each retrieve's signal in `finally` after a + // successful block. The CAR stream must survive that so later blocks still + // come from the one stream rather than re-fetching per block. + it('keeps the stream alive when a completed retrieve signal is aborted', async () => { + const session = createTrustlessGatewayCarSession(components, { allowInsecure: true, allowLocal: true }) + const before = await carRequestCount() + + const controller = new AbortController() + await session.retrieve(root.cid, { validateFn: validateFn(root.cid), signal: controller.signal }) + controller.abort() // mimic raceBlockRetrievers cleanup after the root resolves + + await expect(session.retrieve(inCar.cid, { validateFn: validateFn(inCar.cid) })).to.eventually.deep.equal(inCar.bytes) + + expect(await carRequestCount() - before).to.equal(1) + expect(session.gapFillCount).to.equal(0) + }) + + it('uses explicitly-passed session providers before routing discovery', async () => { + components.routing.findProviders.callsFake(async function * () { + // no routing providers — the session must use init.providers + }) + + const session = createTrustlessGatewayCarSession(components, { + allowInsecure: true, + allowLocal: true, + providers: [uriToMultiaddr(process.env.CAR_GATEWAY ?? '')] + }) + + await expect(session.retrieve(root.cid, { validateFn: validateFn(root.cid) })).to.eventually.deep.equal(root.bytes) + }) + + it('is selected only when carStreamSessions is set on the broker', () => { + const carBroker = trustlessGateway({ allowInsecure: true, allowLocal: true, carStreamSessions: true })(components) + expect(carBroker.createSession?.()?.name).to.equal('trustless-gateway-car-session') + + const defaultBroker = trustlessGateway({ allowInsecure: true, allowLocal: true })(components) + expect(defaultBroker.createSession?.()?.name).to.equal('trustless-gateway-session') + }) + + it('honours a smaller per-retrieve maxSize by deferring to a size-limited raw fetch', async () => { + const session = createTrustlessGatewayCarSession(components, { allowInsecure: true, allowLocal: true }) + await session.retrieve(root.cid, { validateFn: validateFn(root.cid) }) + + // inCar is 4 bytes; maxSize 2 must not return the CAR block, and the raw + // fallback (content-length 4) also exceeds it, so the retrieve rejects. + await expect(session.retrieve(inCar.cid, { validateFn: validateFn(inCar.cid), maxSize: 2 })).to.eventually.be.rejected() + }) +})