diff --git a/.env.example b/.env.example index 583a44814..9b3f421ea 100644 --- a/.env.example +++ b/.env.example @@ -355,3 +355,43 @@ # OTEL Collector image tag # OTEL_COLLECTOR_IMAGE_TAG=0.119.0 + +# ------------------------------------------------------------------------------ +# IPFS (named-content serving + fleet durability) +# ------------------------------------------------------------------------------ + +# Enable the IPFS subsystem (serve /ipfs/{cid} via a paired Kubo node). Requires +# the `ipfs` docker-compose profile (the kubo sidecar). Default: false +# IPFS_ENABLED=true + +# Kubo read-only gateway (:8080) and RPC API (:5001) endpoints. The RPC API is +# required for local-only (offline) reads and for peer-fetch imports. +# IPFS_KUBO_URL=http://kubo:8080 +# IPFS_KUBO_API_URL=http://kubo:5001 + +# Pin ArNS-resolved (named) CIDs so read-only content survives Kubo GC. When +# true, peer-fetched roots are pinned on import. Default: false +# IPFS_PIN_ARNS_CONTENT=true + +# --- Peer-fetch durability layer ------------------------------------------- +# When a gateway lacks a named CID, fetch it from a peer AR.IO gateway that holds +# it as a verifiable CAR; Kubo verifies every block against the CID on import, so +# a lying/tampered peer is rejected. Named content then survives as long as ANY +# fleet gateway holds it. Ships dark — false is a pure passthrough (no behavior +# change). Enable only after validation. Default: false +# IPFS_PEER_FETCH_ENABLED=true + +# Peers to try per CID before falling through to public IPFS. Default: 3 +# IPFS_PEER_FETCH_COUNT=3 + +# Overall deadline for a peer-fetch attempt, ms (kept short — public IPFS is the +# patient fallback). Default: 5000 +# IPFS_PEER_FETCH_TIMEOUT_MS=5000 + +# Max CAR bytes accepted from a peer; above this, skip peers → public IPFS. +# Default: 104857600 (100 MB) +# IPFS_PEER_FETCH_MAX_CAR_BYTES=104857600 + +# Optional comma-separated peer gateway base URLs (private fleets / testing). +# When unset, peers come from the on-chain gateway registry (GAR). Default: unset +# IPFS_PEER_FETCH_STATIC_PEERS=https://peer-a.example,https://peer-b.example diff --git a/docker-compose.yaml b/docker-compose.yaml index 565fa7f11..2233825cd 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -176,6 +176,11 @@ services: - IPFS_RATE_LIMITER_RESOURCE_TOKENS_PER_BUCKET=${IPFS_RATE_LIMITER_RESOURCE_TOKENS_PER_BUCKET:-} - IPFS_RATE_LIMITER_RESOURCE_REFILL_PER_SEC=${IPFS_RATE_LIMITER_RESOURCE_REFILL_PER_SEC:-} - IPFS_MAX_RESPONSE_SIZE_BYTES=${IPFS_MAX_RESPONSE_SIZE_BYTES:-} + - IPFS_PEER_FETCH_ENABLED=${IPFS_PEER_FETCH_ENABLED:-} + - IPFS_PEER_FETCH_COUNT=${IPFS_PEER_FETCH_COUNT:-} + - IPFS_PEER_FETCH_TIMEOUT_MS=${IPFS_PEER_FETCH_TIMEOUT_MS:-} + - IPFS_PEER_FETCH_MAX_CAR_BYTES=${IPFS_PEER_FETCH_MAX_CAR_BYTES:-} + - IPFS_PEER_FETCH_STATIC_PEERS=${IPFS_PEER_FETCH_STATIC_PEERS:-} - NODE_MAX_OLD_SPACE_SIZE=${NODE_MAX_OLD_SPACE_SIZE:-} - ENABLE_FS_HEADER_CACHE_CLEANUP=${ENABLE_FS_HEADER_CACHE_CLEANUP:-} - ON_DEMAND_RETRIEVAL_ORDER=${ON_DEMAND_RETRIEVAL_ORDER:-} diff --git a/package.json b/package.json index e3cb1b7db..799614fb4 100644 --- a/package.json +++ b/package.json @@ -48,6 +48,7 @@ "express-openapi-validator": "^5.4.2", "express-prom-bundle": "^7.0.0", "fastq": "^1.19.1", + "form-data": "^4.0.5", "fs-extra": "^11.3.2", "graphql": "^16.11.0", "ioredis": "^5.8.0", diff --git a/src/config.ts b/src/config.ts index 944ecd4d0..e50d0ff2a 100644 --- a/src/config.ts +++ b/src/config.ts @@ -3320,6 +3320,44 @@ export const IPFS_MAX_RESPONSE_SIZE_BYTES = env.positiveIntOrDefault( 1 * 1024 * 1024 * 1024, // 1 GB ); +// +// IPFS peer-fetch (fleet durability layer) +// + +// Master switch for peer-fetch. Ships dark: when false the IPFS composite is a +// pure passthrough to Kubo (zero behavior change). Enable only after the +// multi-node integration test passes. +export const IPFS_PEER_FETCH_ENABLED = + env.varOrDefault('IPFS_PEER_FETCH_ENABLED', 'false') === 'true'; + +// Peers to try per CID before falling through to public IPFS. +export const IPFS_PEER_FETCH_COUNT = env.positiveIntOrDefault( + 'IPFS_PEER_FETCH_COUNT', + 3, +); + +// Overall deadline for the whole peer-fetch attempt (kept short — public IPFS is +// the patient fallback). +export const IPFS_PEER_FETCH_TIMEOUT_MS = env.positiveIntOrDefault( + 'IPFS_PEER_FETCH_TIMEOUT_MS', + 5000, +); + +// Max CAR bytes accepted from a peer; above this, skip peers → public IPFS. +export const IPFS_PEER_FETCH_MAX_CAR_BYTES = env.positiveIntOrDefault( + 'IPFS_PEER_FETCH_MAX_CAR_BYTES', + 100 * 1024 * 1024, // 100 MB +); + +// Optional deterministic peer override (comma-separated gateway base URLs) for +// private fleets and integration tests; when unset, peers come from the GAR via +// ArIOPeerManager. +export const IPFS_PEER_FETCH_STATIC_PEERS = env + .varOrDefault('IPFS_PEER_FETCH_STATIC_PEERS', '') + .split(',') + .map((s) => s.trim()) + .filter((s) => s.length > 0); + // // StandaloneSqlite worker pools // diff --git a/src/constants.ts b/src/constants.ts index 21e4dc2bf..fdfff03ee 100644 --- a/src/constants.ts +++ b/src/constants.ts @@ -29,6 +29,11 @@ export const headerNames = { // Which retrieval source served the body (e.g. 'ipfs'). Declared centrally so // it's referenced consistently and is a candidate for HTTPSIG trigger headers. arIoSource: 'X-Ar-Io-Source', + // IPFS local-only serve mode: on a request, "serve only from the local + // blockstore, never touch public IPFS/DHT" (peer-fetch recursion guard + + // trustless holding probe); echoed on a local-only hit so a caller/observer + // can assert the server honored the mode. + ipfsLocalOnly: 'X-Ar-Io-Local-Only', origin: 'X-AR-IO-Origin', originNodeRelease: 'X-AR-IO-Origin-Node-Release', digest: 'X-AR-IO-Digest', diff --git a/src/ipfs/ipfs-content-source.ts b/src/ipfs/ipfs-content-source.ts new file mode 100644 index 000000000..58f5bf254 --- /dev/null +++ b/src/ipfs/ipfs-content-source.ts @@ -0,0 +1,31 @@ +/** + * AR.IO Gateway + * Copyright (C) 2022-2025 Permanent Data Solutions, Inc. All Rights Reserved. + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +import { Span } from '@opentelemetry/api'; + +import { IpfsContentResult } from './kubo-data-source.js'; + +// Shared shape for any IPFS content tier (local Kubo, fleet peers, public Kubo). +// Mirrors KuboDataSource.getContent so the composite (SequentialIpfsSource) can +// compose sources uniformly, and IpfsService can depend on the interface rather +// than the concrete KuboDataSource. +export interface IpfsContentSourceOptions { + cidString: string; + path?: string; + signal?: AbortSignal; + parentSpan?: Span; + range?: string; + // Trustless response format: a single verifiable block (`raw`) or a verifiable + // DAG archive (`car`). Absent = UnixFS proxy. + format?: 'raw' | 'car'; + // Serve ONLY from the local blockstore (offline) — never peers, never public + // IPFS. The recursion guard + holding-measurement primitive. + localOnly?: boolean; +} + +export interface IpfsContentSource { + getContent(opts: IpfsContentSourceOptions): Promise; +} diff --git a/src/ipfs/ipfs-peer-data-source.test.ts b/src/ipfs/ipfs-peer-data-source.test.ts new file mode 100644 index 000000000..f84c19c8c --- /dev/null +++ b/src/ipfs/ipfs-peer-data-source.test.ts @@ -0,0 +1,264 @@ +/** + * AR.IO Gateway + * Copyright (C) 2022-2025 Permanent Data Solutions, Inc. All Rights Reserved. + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +import { describe, it, beforeEach, afterEach, mock } from 'node:test'; +import { strict as assert } from 'node:assert'; +import { Readable } from 'node:stream'; +import axios from 'axios'; + +import { createTestLogger } from '../../test/test-logger.js'; +import { + IpfsPeerDataSource, + IPFS_PEER_CATEGORY, +} from './ipfs-peer-data-source.js'; +import { KuboDataSource } from './kubo-data-source.js'; +import { ArIOPeerManager } from '../peers/ar-io-peer-manager.js'; +import * as metrics from '../metrics.js'; + +const CID = 'bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi'; + +// Sum a labeled Prometheus counter's value for a specific label set. +const labeledCounter = async ( + counter: { get: () => Promise<{ values: { labels: any; value: number }[] }> }, + labels: Record, +): Promise => { + const m = await counter.get(); + return m.values + .filter((v) => + Object.entries(labels).every(([k, val]) => v.labels[k] === val), + ) + .reduce((sum, v) => sum + v.value, 0); +}; + +describe('IpfsPeerDataSource', () => { + const log = createTestLogger({ suite: 'IpfsPeerDataSource' }); + + let peerManager: ArIOPeerManager; + let kuboDataSource: KuboDataSource; + let interceptorId: number; + + const reServed = { + stream: Readable.from([Buffer.from('served')]), + size: 6, + contentType: 'text/plain', + statusCode: 200, + }; + + beforeEach(() => { + peerManager = { + selectPeersForKey: mock.fn(() => []), + selectPeers: mock.fn(() => []), + reportSuccess: mock.fn(() => {}), + reportFailure: mock.fn(() => {}), + } as unknown as ArIOPeerManager; + + kuboDataSource = { + // The re-serve after a verified import. + getContent: mock.fn(async () => reServed), + // Post-import presence confirmation (offline block/stat). Defaults to held. + isHeldLocally: mock.fn(async () => true), + } as unknown as KuboDataSource; + }); + + afterEach(() => { + if (interceptorId !== undefined) + axios.interceptors.request.eject(interceptorId); + }); + + const build = (staticPeers: string[], maxCarBytes = 100 * 1024 * 1024) => + new IpfsPeerDataSource({ + log, + peerManager, + kuboApiUrl: 'http://kubo:5001', + kuboDataSource, + peerCount: 3, + requestTimeoutMs: 5000, + maxCarBytes, + staticPeers, + }); + + // Stub the two HTTP calls the source makes: the CAR GET from a peer, and the + // dag/import POST to the local Kubo. `importResults` is consumed in order, one + // per dag/import call, so we can simulate peer1-fails-then-peer2-succeeds. + const stub = ({ + carStatus = 200, + importResults, + }: { + carStatus?: number; + importResults: Array<{ status: number; body: string }>; + }) => { + let importIdx = 0; + interceptorId = axios.interceptors.request.use((config) => { + config.adapter = () => { + const url = config.url ?? ''; + if (url.includes('/api/v0/dag/import')) { + const r = + importResults[Math.min(importIdx, importResults.length - 1)]; + importIdx++; + return Promise.resolve({ + status: r.status, + statusText: '', + headers: {}, + config, + data: r.body, + }); + } + // CAR GET from the peer + return Promise.resolve({ + status: carStatus, + statusText: '', + headers: { 'content-type': 'application/vnd.ipld.car' }, + config, + data: Readable.from([Buffer.from('CAR-BYTES')]), + }); + }; + return config; + }); + }; + + const okImport = { + status: 200, + body: `{"Root":{"Cid":{"/":"${CID}"},"PinErrorMsg":""}}`, + }; + // Kubo returns 200 with an EMPTY body when pin-roots is false (the default) — + // it only echoes the Root when pinning. Success is confirmed via block/stat. + const emptyImport = { status: 200, body: '' }; + const tamperImport = { + status: 500, + body: `{"Message":"import failed: mismatch in content integrity, expected: ${CID}, got: bafkOther","Type":"error"}`, + }; + + it('happy path: fetches a CAR, imports it, re-serves, and reports peer success', async () => { + stub({ importResults: [okImport] }); + const src = build(['http://peer-a:3000']); + + const successBefore = await labeledCounter(metrics.ipfsPeerFetchTotal, { + result: 'success', + }); + const result = await src.getContent({ cidString: CID }); + + assert.equal(result.statusCode, 200); + // Success metric incremented. + assert.equal( + (await labeledCounter(metrics.ipfsPeerFetchTotal, { + result: 'success', + })) - successBefore, + 1, + ); + // Re-served via the local Kubo (gateway) after import. + assert.equal((kuboDataSource.getContent as any).mock.calls.length, 1); + const reportSuccess = (peerManager.reportSuccess as any).mock; + assert.equal(reportSuccess.calls.length, 1); + assert.equal(reportSuccess.calls[0].arguments[0], IPFS_PEER_CATEGORY); + assert.equal(reportSuccess.calls[0].arguments[1], 'http://peer-a:3000'); + result.stream.destroy(); + }); + + it('treats an empty 200 dag/import body (pin-roots=false) as success when the root is now held', async () => { + // Regression: dag/import with pin-roots=false returns 200 + empty body. + // Success must be confirmed by presence (block/stat), not by parsing "Root". + stub({ importResults: [emptyImport] }); + const src = build(['http://peer-a:3000']); + + const result = await src.getContent({ cidString: CID }); + + assert.equal(result.statusCode, 200); + assert.equal((peerManager.reportSuccess as any).mock.calls.length, 1); + assert.equal((kuboDataSource.isHeldLocally as any).mock.calls.length, 1); + }); + + it('rejects a valid CAR whose root is NOT actually held after import (wrong-content peer)', async () => { + // A peer could return a CAR whose blocks all hash correctly but that does + // not contain the requested root. block/stat then reports the root absent. + kuboDataSource.isHeldLocally = mock.fn(async () => false) as any; + stub({ importResults: [emptyImport] }); + const src = build(['http://peer-a:3000']); + + await assert.rejects( + () => src.getContent({ cidString: CID }), + (e: any) => e.name === 'IpfsNotFoundError', + ); + assert.equal((peerManager.reportFailure as any).mock.calls.length, 1); + assert.equal((peerManager.reportSuccess as any).mock.calls.length, 0); + }); + + it('tamper: a CAR that fails Kubo verification is rejected and the next peer is tried', async () => { + // peer-a's import fails verification (mismatch), peer-b's succeeds. + stub({ importResults: [tamperImport, okImport] }); + const src = build(['http://peer-a:3000', 'http://peer-b:3000']); + + const verifyFailedBefore = await labeledCounter( + metrics.ipfsPeerFetchPeerAttemptsTotal, + { result: 'import_verify_failed' }, + ); + const result = await src.getContent({ cidString: CID }); + + assert.equal(result.statusCode, 200); + // The lying peer's tampered CAR was metered as a verify failure. + assert.ok( + (await labeledCounter(metrics.ipfsPeerFetchPeerAttemptsTotal, { + result: 'import_verify_failed', + })) - + verifyFailedBefore >= + 1, + ); + // The lying peer was reported failed; the good peer reported success. + const fail = (peerManager.reportFailure as any).mock; + const ok = (peerManager.reportSuccess as any).mock; + assert.equal(fail.calls.length, 1); + assert.equal(fail.calls[0].arguments[1], 'http://peer-a:3000'); + assert.equal(ok.calls.length, 1); + assert.equal(ok.calls[0].arguments[1], 'http://peer-b:3000'); + result.stream.destroy(); + }); + + it('local-only: throws IpfsNotFoundError immediately with no peer calls (recursion guard)', async () => { + stub({ importResults: [okImport] }); + const src = build(['http://peer-a:3000']); + + await assert.rejects( + () => src.getContent({ cidString: CID, localOnly: true }), + (e: any) => e.name === 'IpfsNotFoundError', + ); + // No import, no re-serve, no peer reporting happened. + assert.equal((kuboDataSource.getContent as any).mock.calls.length, 0); + assert.equal((peerManager.reportSuccess as any).mock.calls.length, 0); + assert.equal((peerManager.reportFailure as any).mock.calls.length, 0); + }); + + it('all peers fail → IpfsNotFoundError (composite will fall through to public IPFS)', async () => { + stub({ importResults: [tamperImport, tamperImport] }); + const src = build(['http://peer-a:3000', 'http://peer-b:3000']); + + await assert.rejects( + () => src.getContent({ cidString: CID }), + (e: any) => e.name === 'IpfsNotFoundError', + ); + assert.equal((peerManager.reportFailure as any).mock.calls.length, 2); + assert.equal((kuboDataSource.getContent as any).mock.calls.length, 0); + }); + + it('no peers available → IpfsNotFoundError', async () => { + stub({ importResults: [okImport] }); + const src = build([]); // no static peers; peerManager returns none + + await assert.rejects( + () => src.getContent({ cidString: CID }), + (e: any) => e.name === 'IpfsNotFoundError', + ); + }); + + it('a non-200 from a peer is treated as a failure and the next peer is tried', async () => { + stub({ carStatus: 404, importResults: [okImport] }); + const src = build(['http://peer-a:3000']); + + await assert.rejects( + () => src.getContent({ cidString: CID }), + (e: any) => e.name === 'IpfsNotFoundError', + ); + assert.equal((peerManager.reportFailure as any).mock.calls.length, 1); + }); +}); diff --git a/src/ipfs/ipfs-peer-data-source.ts b/src/ipfs/ipfs-peer-data-source.ts new file mode 100644 index 000000000..fc68117af --- /dev/null +++ b/src/ipfs/ipfs-peer-data-source.ts @@ -0,0 +1,334 @@ +/** + * AR.IO Gateway + * Copyright (C) 2022-2025 Permanent Data Solutions, Inc. All Rights Reserved. + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +import { default as axios } from 'axios'; +import FormData from 'form-data'; +import { Readable, Transform } from 'node:stream'; +import winston from 'winston'; + +import { headerNames } from '../constants.js'; +import { ArIOPeerManager } from '../peers/ar-io-peer-manager.js'; +import { startChildSpan } from '../tracing.js'; +import * as metrics from '../metrics.js'; +import { + IpfsContentSource, + IpfsContentSourceOptions, +} from './ipfs-content-source.js'; +import { + IpfsContentResult, + IpfsNotFoundError, + KuboDataSource, +} from './kubo-data-source.js'; + +// Peer-selection weight category registered in ArIOPeerManager for IPFS fleet +// fetches (self-registers on first use). Distinct from 'data' / 'chunk'. +export const IPFS_PEER_CATEGORY = 'ipfs'; + +// A dag/import response (JSON or NDJSON) that indicates a verification/import +// failure. Kubo verifies every block against its CID on import, so a +// tampered/lying peer's CAR trips this. Confirmed against kubo v0.32.1: +// a good import is `{"Root":{"Cid":{"/":""},"PinErrorMsg":""}}` (HTTP 200); +// a byte-tampered CAR is `{"Message":"import failed: mismatch in content +// integrity, expected: , got: ","Type":"error"}` (HTTP 500). +const IMPORT_ERROR_RE = + /mismatch in content integrity|import failed|"Type"\s*:\s*"error"/i; + +/** + * Tier 2 — fetch a CID this gateway lacks from a peer AR.IO gateway that holds + * it, as a verifiable CAR, and import it into the local Kubo. Kubo verifies + * every block against the CID on `dag/import`, so a peer is NEVER trusted: a + * lying/tampered CAR fails to import and we move to the next peer. After a + * verified import the content is local, so we re-serve it through the normal + * KuboDataSource (gateway) path — giving the correct sniffed Content-Type and a + * local-first serve, not the offline octet-stream default. + */ +export class IpfsPeerDataSource implements IpfsContentSource { + private log: winston.Logger; + private peerManager: ArIOPeerManager; + private kuboApiUrl: string; + private kuboDataSource: KuboDataSource; + private peerCount: number; + private requestTimeoutMs: number; + private maxCarBytes: number; + private pinRoots: boolean; + private staticPeers: string[]; + + constructor({ + log, + peerManager, + kuboApiUrl, + kuboDataSource, + peerCount, + requestTimeoutMs, + maxCarBytes, + pinRoots = false, + staticPeers = [], + }: { + log: winston.Logger; + peerManager: ArIOPeerManager; + kuboApiUrl: string; + kuboDataSource: KuboDataSource; + peerCount: number; + requestTimeoutMs: number; + maxCarBytes: number; + pinRoots?: boolean; + // Deterministic peer override (private fleets / integration tests); when set, + // used instead of ArIOPeerManager selection. + staticPeers?: string[]; + }) { + this.log = log.child({ class: this.constructor.name }); + this.peerManager = peerManager; + this.kuboApiUrl = kuboApiUrl.replace(/\/$/, ''); + this.kuboDataSource = kuboDataSource; + this.peerCount = peerCount; + this.requestTimeoutMs = requestTimeoutMs; + this.maxCarBytes = maxCarBytes; + this.pinRoots = pinRoots; + this.staticPeers = staticPeers; + } + + async getContent(opts: IpfsContentSourceOptions): Promise { + const { cidString, path, signal, range, format, parentSpan } = opts; + + // Recursion guard (belt-and-suspenders with the composite's gating): a peer + // source must NEVER run under local-only — that mode is tier-1-only. + if (opts.localOnly === true) { + throw new IpfsNotFoundError( + 'peer fetch is unavailable in local-only mode', + ); + } + + const span = startChildSpan( + 'IpfsPeerDataSource.getContent', + { attributes: { 'ipfs.cid': cidString } }, + parentSpan, + ); + + try { + const peers = this.selectPeers(cidString); + if (peers.length === 0) { + throw new IpfsNotFoundError('no IPFS fleet peers available'); + } + + const deadline = Date.now() + this.requestTimeoutMs; + for (const peer of peers) { + if (signal?.aborted) { + const err = new Error('client aborted'); + err.name = 'AbortError'; + throw err; + } + if (Date.now() >= deadline) { + this.log.debug('IPFS peer-fetch deadline reached', { cidString }); + break; + } + + try { + const carBytes = await this.fetchAndImport( + peer, + cidString, + signal, + deadline, + ); + metrics.ipfsPeerFetchPeerAttemptsTotal.inc({ result: 'success' }); + metrics.ipfsPeerFetchCarBytesTotal.inc(carBytes); + this.peerManager.reportSuccess(IPFS_PEER_CATEGORY, peer); + this.log.debug('IPFS peer-fetch import verified', { + cidString, + peer, + }); + + // Content is now local (verified). Re-serve via the normal gateway + // path: correct Content-Type + local-first (no public walk since the + // blocks are present). localOnly is intentionally left false here. + const result = await this.kuboDataSource.getContent({ + cidString, + path, + signal, + parentSpan: span, + range, + format, + }); + metrics.ipfsPeerFetchTotal.inc({ result: 'success' }); + span.end(); + return result; + } catch (err: any) { + // A genuine client disconnect short-circuits the whole cascade. + if (err?.name === 'AbortError' && signal?.aborted) { + throw err; + } + metrics.ipfsPeerFetchPeerAttemptsTotal.inc({ + // A failed block-integrity check (a lying/tampered peer) vs. a + // transport/non-200/cap error — operators want these separated. + result: + err?.verifyFailed === true + ? 'import_verify_failed' + : 'peer_error', + }); + this.peerManager.reportFailure(IPFS_PEER_CATEGORY, peer); + this.log.debug('IPFS peer-fetch attempt failed, trying next peer', { + cidString, + peer, + message: err?.message, + }); + } + } + + // No peer held it (or all lied / timed out / exceeded the cap). The + // composite falls through to public IPFS (tier 3). + metrics.ipfsPeerFetchTotal.inc({ result: 'miss' }); + throw new IpfsNotFoundError(`no fleet peer holds ${cidString}`); + } catch (error: any) { + if (error?.name !== 'AbortError') { + span.recordException(error); + } + span.end(); + throw error; + } + } + + // Prefer hash-ring selection (cache locality: the same CID tends to hit the + // same peers, warming them); fall back to weighted selection, then to the + // static override. Selection throws when no peers exist — treat as empty. + private selectPeers(cidString: string): string[] { + if (this.staticPeers.length > 0) { + return this.staticPeers.slice(0, this.peerCount); + } + try { + const byKey = this.peerManager.selectPeersForKey( + IPFS_PEER_CATEGORY, + cidString, + this.peerCount, + ); + if (byKey.length > 0) return byKey; + } catch { + // fall through to weighted selection + } + try { + return this.peerManager.selectPeers(IPFS_PEER_CATEGORY, this.peerCount); + } catch { + return []; + } + } + + // Fetch the CAR from a peer (local-only, byte-capped) and stream it into + // Kubo's dag/import (which verifies every block against the CID). Throws on + // any failure so the caller can try the next peer. + private async fetchAndImport( + peer: string, + cidString: string, + signal: AbortSignal | undefined, + deadline: number, + ): Promise { + const remainingMs = Math.max(0, deadline - Date.now()); + const peerUrl = `${peer.replace(/\/$/, '')}/ipfs/${cidString}?format=car`; + + // 1) Fetch the CAR from the peer with the local-only hint so the peer serves + // ONLY from its local store (no recursion/amplification across the fleet). + const carResponse = await axios.get(peerUrl, { + responseType: 'stream', + signal, + timeout: remainingMs, + headers: { + [headerNames.ipfsLocalOnly]: 'true', + Accept: 'application/vnd.ipld.car', + 'Accept-Encoding': 'identity', + }, + maxRedirects: 2, + validateStatus: () => true, + }); + + if (carResponse.status !== 200) { + (carResponse.data as Readable).destroy(); + throw new Error(`peer ${peer} returned status ${carResponse.status}`); + } + + // 2) Byte-cap the CAR (a too-large file falls through to public IPFS) and + // stream it into dag/import as multipart/form-data. + const carBytes = { value: 0 }; + const cappedCar = capStream( + carResponse.data as Readable, + this.maxCarBytes, + carBytes, + ); + const form = new FormData(); + form.append('file', cappedCar, { + filename: `${cidString}.car`, + contentType: 'application/vnd.ipld.car', + }); + + const importResponse = await axios.post( + `${this.kuboApiUrl}/api/v0/dag/import`, + form, + { + params: { 'pin-roots': this.pinRoots, progress: false }, + headers: form.getHeaders(), + signal, + timeout: remainingMs, + maxBodyLength: Infinity, + maxContentLength: Infinity, + responseType: 'text', + validateStatus: () => true, + }, + ); + + // 3) Verify the import outcome. Kubo verifies every block against its CID on + // import; a tampered/lying CAR yields an error body (and HTTP 500) which + // we detect from the BODY (robust even if a future Kubo streams the error + // in an NDJSON line with HTTP 200). + const body = String(importResponse.data ?? ''); + if (IMPORT_ERROR_RE.test(body)) { + // Tag block-integrity failures so the caller can meter a lying/tampered + // peer distinctly from a transport error. + const err = new Error( + `dag/import verification failed for ${cidString}: ${body.slice(0, 200)}`, + ); + (err as Error & { verifyFailed?: boolean }).verifyFailed = true; + throw err; + } + if (importResponse.status !== 200) { + throw new Error( + `dag/import HTTP ${importResponse.status} for ${cidString}: ${body.slice(0, 200)}`, + ); + } + // Confirm the requested root is now held locally. We do NOT parse the + // response for a Root CID: dag/import returns 200 with an EMPTY body when + // pin-roots is false (the default) — it only echoes the Root when pinning. + // An offline block/stat is the authoritative success signal and, crucially, + // also rejects a peer that returned a VALID CAR of the WRONG content (all + // blocks hash correctly, but the requested root isn't among them). + if (!(await this.kuboDataSource.isHeldLocally(cidString, signal))) { + throw new Error( + `imported CAR did not yield the requested root ${cidString}`, + ); + } + return carBytes.value; + } +} + +// Cap a stream at maxBytes, erroring (and tearing down the source) once +// exceeded, while tallying the bytes seen into `counter`. maxBytes <= 0 disables +// the cap but still counts (for the imported-bytes metric). +function capStream( + src: Readable, + maxBytes: number, + counter: { value: number }, +): Readable { + let seen = 0; + const guard = new Transform({ + transform(chunk: Buffer, _enc, cb) { + seen += chunk.length; + counter.value = seen; + if (maxBytes > 0 && seen > maxBytes) { + cb(new Error(`CAR exceeds max size ${maxBytes} bytes`)); + return; + } + cb(null, chunk); + }, + }); + src.on('error', (err) => guard.destroy(err)); + guard.on('error', () => src.destroy()); + return src.pipe(guard); +} diff --git a/src/ipfs/ipfs-service.test.ts b/src/ipfs/ipfs-service.test.ts index 9f288fdb5..7e1c5bcfb 100644 --- a/src/ipfs/ipfs-service.test.ts +++ b/src/ipfs/ipfs-service.test.ts @@ -171,6 +171,103 @@ describe('IpfsService', () => { }); }); + describe('local-only (holding primitive) is isolated from the negative cache', () => { + it('threads localOnly to the data source and does NOT record a negative-cache miss on a local miss', async () => { + const getContent = mock.fn(async () => { + throw new IpfsNotFoundError('not held locally'); + }); + dataSource = { getContent } as unknown as KuboDataSource; + + const service = buildService(); + await assert.rejects( + () => + service.getContent({ + cidString: CID, + format: 'raw', + localOnly: true, + }), + (e: any) => e instanceof IpfsNotFoundError, + ); + + // localOnly is passed through to the source... + assert.equal(getContent.mock.calls[0].arguments[0].localOnly, true); + // ...and a local miss must NOT poison the negative cache (which would + // blackhole the CID for the normal peer/public fallback path). + assert.equal((negativeCache.recordMiss as any).mock.calls.length, 0); + }); + + it('ignores an existing negative-cache entry under local-only (reflects true current local state)', async () => { + negativeCache.isNegativelyCached = mock.fn(() => true) as any; + const getContent = mock.fn(async () => ({ + stream: makeInfiniteStream(), + size: 8, + contentType: 'application/vnd.ipld.raw', + statusCode: 200, + })); + dataSource = { getContent } as unknown as KuboDataSource; + + const service = buildService(); + const result = await service.getContent({ + cidString: CID, + format: 'raw', + localOnly: true, + }); + + // Reached the data source despite the negative-cache entry. + assert.equal(getContent.mock.calls.length, 1); + result.stream.destroy(); + }); + + it('does NOT write the on-disk cache under local-only (avoids persisting the octet-stream default)', async () => { + const source = makeInfiniteStream(); + dataSource = { + getContent: mock.fn(async () => ({ + stream: source, + size: 8, + // no format => this would normally be tee'd to the on-disk cache + contentType: 'application/octet-stream', + statusCode: 200, + })), + } as unknown as KuboDataSource; + + const service = buildService(); + const result = await service.getContent({ + cidString: CID, + localOnly: true, + }); + result.stream.destroy(); + + assert.equal((cache.putFromFile as any).mock.calls.length, 0); + }); + + it('serves an on-disk cache hit as a local hold and leaves the negative cache untouched', async () => { + cache.get = mock.fn(async () => ({ + stream: makeInfiniteStream(), + size: 8, + contentType: 'image/png', + digest: undefined, + })) as any; + dataSource = { + getContent: mock.fn(async () => { + throw new Error('must not reach the data source on a cache hit'); + }), + } as unknown as KuboDataSource; + + const service = buildService(); + const result = await service.getContent({ + cidString: CID, + localOnly: true, + }); + + assert.equal(result.cached, true); + assert.equal(result.contentType, 'image/png'); + // Negative-cache health is not touched under local-only. + assert.equal((negativeCache.evict as any).mock.calls.length, 0); + assert.equal((negativeCache.recordSuccess as any).mock.calls.length, 0); + result.stream.destroy(); + }); + }); + describe('L1: cache hit/miss counters increment once, in the service', () => { it('increments the miss counter exactly once per uncached fetch', async () => { dataSource = { diff --git a/src/ipfs/ipfs-service.ts b/src/ipfs/ipfs-service.ts index a072f077b..0cda74b6e 100644 --- a/src/ipfs/ipfs-service.ts +++ b/src/ipfs/ipfs-service.ts @@ -15,8 +15,8 @@ import { startChildSpan } from '../tracing.js'; import { IpfsFsCache } from './ipfs-cache.js'; import { DataBlockListValidator } from '../types.js'; import { NegativeDataCache } from '../data/negative-data-cache.js'; +import { IpfsContentSource } from './ipfs-content-source.js'; import { - KuboDataSource, IpfsBlockedError, IpfsNotFoundError, IpfsSizeLimitError, @@ -43,7 +43,7 @@ export interface IpfsGetContentResult { export class IpfsService { private log: winston.Logger; - private dataSource: KuboDataSource; + private dataSource: IpfsContentSource; private cache: IpfsFsCache; private blockListValidator: DataBlockListValidator; private maxResponseSizeBytes: number; @@ -67,7 +67,7 @@ export class IpfsService { negativeCache, }: { log: winston.Logger; - dataSource: KuboDataSource; + dataSource: IpfsContentSource; cache: IpfsFsCache; blockListValidator: DataBlockListValidator; maxResponseSizeBytes: number; @@ -88,6 +88,7 @@ export class IpfsService { parentSpan, range, format, + localOnly = false, }: { cidString: string; path?: string; @@ -95,6 +96,14 @@ export class IpfsService { parentSpan?: Span; range?: string; format?: 'raw' | 'car'; + // Serve only from local Kubo (offline) — the fleet-durability / holding + // primitive. Local-only requests bypass the negative cache entirely (read + // AND write) so a probe reflects true current local state and can never + // poison a normal request's later peer/public fallback, and they do NOT + // write the on-disk cache (an offline UnixFS body lacks a sniffed + // Content-Type; the normal :8080 path populates the cache properly). They + // DO read the on-disk cache — a cached object is a legitimate local hold. + localOnly?: boolean; }): Promise { const span = startChildSpan( 'IpfsService.getContent', @@ -102,6 +111,7 @@ export class IpfsService { attributes: { 'ipfs.cid': cidString, 'ipfs.path': path ?? '', + 'ipfs.local_only': localOnly, }, }, parentSpan, @@ -162,7 +172,10 @@ export class IpfsService { // (absent or unpinned) so they don't re-hit Kubo on every request // (latency / DoS amplification) — mirrors the Arweave path's negative data // cache. Only trips after repeated misses (count + duration thresholds). - if (this.negativeCache?.isNegativelyCached(negKey) === true) { + if ( + !localOnly && + this.negativeCache?.isNegativelyCached(negKey) === true + ) { span.setAttribute('ipfs.negative_cache', 'hit'); span.end(); throw new IpfsNotFoundError( @@ -199,8 +212,11 @@ export class IpfsService { // Content is available — clear any negative-cache entry and record a // success so a transiently-unavailable CID that later pins isn't kept in // a negative-cache blackout, and IPFS health isn't skewed miss-only. - this.negativeCache?.evict(negKey); - this.negativeCache?.recordSuccess(); + // (Skipped under local-only, which is isolated from the negative cache.) + if (!localOnly) { + this.negativeCache?.evict(negKey); + this.negativeCache?.recordSuccess(); + } this.log.debug('IPFS cache hit', { cid: normalizedCid, path }); metrics.ipfsCacheHitTotal.inc(); span.setAttributes({ @@ -231,8 +247,15 @@ export class IpfsService { parentSpan: span, range, format, + localOnly, }) .catch((err) => { + // Local-only is isolated from the negative cache: a miss here just + // means "not held locally right now" and must not blackhole the CID + // for the normal peer/public fallback path. + if (localOnly) { + throw err; + } // Record a negative-cache miss for content Kubo could not retrieve: // a 404, or a retrieval TIMEOUT (the "no provider on the network" // case — Kubo never returns 404 for absent network content, it times @@ -278,16 +301,23 @@ export class IpfsService { // A successful fetch means the content is available — clear any // negative-cache entry and record health (see the cache-hit path). - this.negativeCache?.evict(negKey); - this.negativeCache?.recordSuccess(); + // (Skipped under local-only, which is isolated from the negative cache.) + if (!localOnly) { + this.negativeCache?.evict(negKey); + this.negativeCache?.recordSuccess(); + } // Stream directly to the client while writing to a temp file on disk for // caching. No memory buffering — handles files of any size. Partial (206) - // responses are NOT cached — only full objects. + // responses are NOT cached — only full objects. Local-only responses are + // NOT cached either: an offline UnixFS body carries no sniffed + // Content-Type (defaults to octet-stream), so persisting it would poison + // the cache — the normal :8080 path caches it with the correct type. if ( range === undefined && format === undefined && - result.statusCode === 200 + result.statusCode === 200 && + !localOnly ) { this.streamToCache( normalizedCid, diff --git a/src/ipfs/kubo-data-source.test.ts b/src/ipfs/kubo-data-source.test.ts index 2035aae20..b0ebaee6b 100644 --- a/src/ipfs/kubo-data-source.test.ts +++ b/src/ipfs/kubo-data-source.test.ts @@ -260,6 +260,160 @@ describe('KuboDataSource', () => { }); }); + describe('local-only (offline RPC)', () => { + const CID = 'bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi'; + let interceptorId: number; + afterEach(() => axios.interceptors.request.eject(interceptorId)); + + const offlineDs = () => + new KuboDataSource({ + log, + kuboUrl: 'http://localhost:8080', + kuboApiUrl: 'http://localhost:5001', + requestTimeoutMs: 5000, + streamStallTimeoutMs: 5000, + }); + + // Stub the Kubo RPC: capture the request config and return `status` with an + // optional body. `body` is a string (error JSON) or a Buffer (content). + const stubRpc = ( + status: number, + body: string | Buffer, + headers: Record = {}, + ) => { + let captured: any; + interceptorId = axios.interceptors.request.use((config) => { + captured = config; + config.adapter = () => + Promise.resolve({ + status, + statusText: '', + headers, + config, + data: Readable.from([ + typeof body === 'string' ? Buffer.from(body) : body, + ]), + }); + return config; + }); + return () => captured; + }; + + it('raw local-only issues block/get with offline=true', async () => { + const get = stubRpc(200, Buffer.alloc(8)); + const result = await offlineDs().getContent({ + cidString: CID, + format: 'raw', + localOnly: true, + }); + + const cfg = get(); + assert.equal(cfg.method, 'post'); + assert.match(cfg.url, /\/api\/v0\/block\/get$/); + assert.equal(cfg.params.arg, CID); + assert.equal(cfg.params.offline, true); + assert.equal(result.statusCode, 200); + assert.equal(result.contentType, 'application/vnd.ipld.raw'); + result.stream.destroy(); + }); + + it('car local-only issues dag/export with offline=true', async () => { + const get = stubRpc(200, Buffer.alloc(8)); + const result = await offlineDs().getContent({ + cidString: CID, + format: 'car', + localOnly: true, + }); + + const cfg = get(); + assert.match(cfg.url, /\/api\/v0\/dag\/export$/); + assert.equal(cfg.params.offline, true); + assert.equal(result.contentType, 'application/vnd.ipld.car'); + result.stream.destroy(); + }); + + it('no-format local-only issues cat with offline=true and encodes the path into arg', async () => { + const get = stubRpc(200, Buffer.alloc(8)); + const result = await offlineDs().getContent({ + cidString: CID, + path: 'images/logo one.png', + localOnly: true, + }); + + const cfg = get(); + assert.match(cfg.url, /\/api\/v0\/cat$/); + assert.equal(cfg.params.arg, `${CID}/images/logo%20one.png`); + assert.equal(cfg.params.offline, true); + assert.equal(result.contentType, 'application/octet-stream'); + result.stream.destroy(); + }); + + it('maps an offline local miss (500 + "not found locally") to IpfsNotFoundError', async () => { + stubRpc( + 500, + JSON.stringify({ + Message: `block was not found locally (offline): ipld: could not find ${CID}`, + Code: 0, + Type: 'error', + }), + ); + + await assert.rejects( + () => + offlineDs().getContent({ + cidString: CID, + format: 'raw', + localOnly: true, + }), + (error: any) => { + assert.equal(error.name, 'IpfsNotFoundError'); + return true; + }, + ); + }); + + it('maps a non-miss 500 to IpfsUnavailableError (a real Kubo fault is not masked as a benign miss)', async () => { + stubRpc( + 500, + JSON.stringify({ + Message: 'datastore io error', + Code: 0, + Type: 'error', + }), + ); + + await assert.rejects( + () => + offlineDs().getContent({ + cidString: CID, + format: 'raw', + localOnly: true, + }), + (error: any) => { + assert.equal(error.name, 'IpfsUnavailableError'); + return true; + }, + ); + }); + + it('throws IpfsUnavailableError when the RPC API URL is not configured', async () => { + const ds = new KuboDataSource({ + log, + kuboUrl: 'http://localhost:8080', + requestTimeoutMs: 5000, + streamStallTimeoutMs: 5000, + }); + + await assert.rejects( + () => ds.getContent({ cidString: CID, format: 'raw', localOnly: true }), + (error: any) => { + assert.equal(error.name, 'IpfsUnavailableError'); + return true; + }, + ); + }); + }); + describe('error types', () => { it('IpfsNotFoundError has correct name', () => { const error = new IpfsNotFoundError('not found'); diff --git a/src/ipfs/kubo-data-source.ts b/src/ipfs/kubo-data-source.ts index f34353d9d..ce3de20a5 100644 --- a/src/ipfs/kubo-data-source.ts +++ b/src/ipfs/kubo-data-source.ts @@ -22,9 +22,38 @@ export interface IpfsContentResult { contentRange?: string; } +// Kubo RPC error text for an offline (local-only) read that misses the local +// blockstore. Confirmed against ipfs/kubo:v0.32.1 (§3.1a spike): block/get, +// dag/export, and cat all return HTTP 500 with a body of the form +// `{"Message":"block was not found locally (offline): ipld: could not find +// ", ...}` when the content isn't held locally. We map exactly this to +// IpfsNotFoundError; any other non-200 stays an error so a real Kubo fault +// isn't masked as a benign miss. +const OFFLINE_MISS_RE = + /not found locally|could not find|not found|key not found/i; + +interface GetContentOptions { + cidString: string; + path?: string; + signal?: AbortSignal; + parentSpan?: Span; + range?: string; + // Trustless response format passed through to Kubo: a single verifiable + // block (`raw`) or a verifiable DAG archive (`car`). Absent = UnixFS proxy. + format?: 'raw' | 'car'; + // Serve ONLY from the local Kubo blockstore/pinset — never touch public + // IPFS/DHT. Routed through the Kubo RPC API with `offline=true`; a local miss + // returns fast as IpfsNotFoundError. This is the load-bearing primitive for + // peer-fetch recursion prevention and trustless holding measurement. + localOnly?: boolean; +} + export class KuboDataSource { private log: winston.Logger; private kuboUrl: string; + // Kubo RPC API base (:5001). Required only for local-only (offline) reads; the + // read-only gateway (:8080) has no per-request offline flag. + private kuboApiUrl?: string; private requestTimeoutMs: number; private streamStallTimeoutMs: number; private maxConcurrent: number; @@ -34,6 +63,7 @@ export class KuboDataSource { constructor({ log, kuboUrl, + kuboApiUrl, requestTimeoutMs, streamStallTimeoutMs, maxConcurrent = 0, @@ -41,6 +71,7 @@ export class KuboDataSource { }: { log: winston.Logger; kuboUrl: string; + kuboApiUrl?: string; requestTimeoutMs: number; streamStallTimeoutMs: number; maxConcurrent?: number; @@ -48,35 +79,82 @@ export class KuboDataSource { }) { this.log = log.child({ class: this.constructor.name }); this.kuboUrl = kuboUrl.replace(/\/$/, ''); + this.kuboApiUrl = kuboApiUrl?.replace(/\/$/, ''); this.requestTimeoutMs = requestTimeoutMs; this.streamStallTimeoutMs = streamStallTimeoutMs; this.maxConcurrent = maxConcurrent; this.maxRequestMs = maxRequestMs; } - async getContent({ - cidString, - path, - signal, - parentSpan, - range, - format, - }: { - cidString: string; - path?: string; - signal?: AbortSignal; - parentSpan?: Span; - range?: string; - // Trustless response format passed through to Kubo: a single verifiable - // block (`raw`) or a verifiable DAG archive (`car`). Absent = UnixFS proxy. - format?: 'raw' | 'car'; - }): Promise { - signal?.throwIfAborted(); + async getContent(opts: GetContentOptions): Promise { + opts.signal?.throwIfAborted(); // Concurrency cap: bound in-flight Kubo fetches so cheap-to-issue requests // (HEAD, tiny Range) can't amplify into unbounded upstream/DHT load — excess // requests fail fast instead of piling onto Kubo. The slot is released when - // the returned stream closes (below) or on any error (catch). + // the returned stream closes (via finalizeStream) or on any error (below). + const release = this.acquireSlot(); + + const span = startChildSpan( + 'KuboDataSource.getContent', + { + attributes: { + 'ipfs.cid': opts.cidString, + 'ipfs.path': opts.path ?? '', + 'ipfs.local_only': opts.localOnly === true, + }, + }, + opts.parentSpan, + ); + + try { + return opts.localOnly === true + ? await this.getContentOffline(opts, span, release) + : await this.getContentFromGateway(opts, span, release); + } catch (error: any) { + // The branch methods map upstream errors but leave slot/span teardown to + // here so success (stream lifecycle) and failure share one owner. release + // is idempotent, so this is safe even if a branch already released. + release(); + if (error.name !== 'AbortError') { + span.recordException(error); + } + span.end(); + throw error; + } + } + + // Cheap offline presence check: is the ROOT block of this CID in the local + // blockstore? Uses the RPC `block/stat?offline=true` (fast; never a DHT walk). + // The composite uses it to decide "serve locally via the gateway" vs "acquire + // from a fleet peer" WITHOUT first triggering a public-IPFS walk. Best-effort + // and root-only (a full-DAG guarantee would need `cat --offline`), which is + // sufficient for routing; returns false on any error or if the RPC API is + // unconfigured. + async isHeldLocally( + cidString: string, + signal?: AbortSignal, + ): Promise { + if (this.kuboApiUrl === undefined) return false; + try { + const response = await axios.post( + `${this.kuboApiUrl}/api/v0/block/stat`, + undefined, + { + params: { arg: cidString, offline: true }, + signal, + timeout: this.requestTimeoutMs, + validateStatus: () => true, + }, + ); + return response.status === 200; + } catch { + return false; + } + } + + // Reserve a concurrency slot; returns an idempotent release fn. + private acquireSlot(): () => void { if (this.maxConcurrent > 0 && this.inFlight >= this.maxConcurrent) { throw new IpfsUnavailableError( `Too many concurrent IPFS fetches (${this.inFlight}/${this.maxConcurrent})`, @@ -84,13 +162,105 @@ export class KuboDataSource { } this.inFlight++; let released = false; - const release = () => { + return () => { if (!released) { released = true; this.inFlight--; } }; + } + + // Wire a successfully-opened upstream stream into an IpfsContentResult: + // switch to the stall timeout, end the span exactly once, and release the + // concurrency slot when the stream terminates ('close' covers the + // destroy()-without-error paths — HEAD, client abort, rate-limited teardown — + // that emit only 'close', not 'end'/'error'). + private finalizeStream( + stream: Readable, + meta: { + contentLength: number; + contentType: string; + statusCode: number; + contentRange?: string; + }, + release: () => void, + span: Span, + logContext: Record, + ): IpfsContentResult { + attachStallTimeout(stream, this.streamStallTimeoutMs, this.maxRequestMs); + + span.setAttributes({ + 'ipfs.content_length': meta.contentLength, + 'ipfs.content_type': meta.contentType, + }); + span.addEvent('Kubo fetch successful'); + this.log.debug('Kubo fetch successful', { + ...logContext, + contentLength: meta.contentLength, + contentType: meta.contentType, + }); + + let spanEnded = false; + const endSpan = () => { + if (spanEnded) return; + spanEnded = true; + span.end(); + }; + stream.on('end', endSpan); + stream.on('error', (err) => { + span.recordException(err); + endSpan(); + }); + stream.once('close', () => { + endSpan(); + release(); + }); + + return { + stream, + size: meta.contentLength, + contentType: meta.contentType, + statusCode: meta.statusCode, + contentRange: meta.contentRange, + }; + } + + // Connection-phase timeout + client-abort plumbing shared by both fetch + // paths. Returns the AbortController to pass to axios and a `detach` that + // clears the timer and removes the client-abort listener (call on both + // success and error). + private setupAbort(signal?: AbortSignal): { + controller: AbortController; + detach: () => void; + } { + const controller = new AbortController(); + const connectionTimer = setTimeout(() => { + controller.abort(new Error('Kubo connection timeout')); + }, this.requestTimeoutMs); + + const onClientAbort = () => controller.abort(signal?.reason); + if (signal?.aborted) { + onClientAbort(); + } else if (signal) { + signal.addEventListener('abort', onClientAbort, { once: true }); + } + + return { + controller, + detach: () => { + clearTimeout(connectionTimer); + signal?.removeEventListener('abort', onClientAbort); + }, + }; + } + // Non-local-only path: fetch via the read-only Kubo gateway (:8080), which may + // reach public IPFS/DHT. Behavior unchanged from the original implementation. + private async getContentFromGateway( + { cidString, path, signal, range, format }: GetContentOptions, + span: Span, + release: () => void, + ): Promise { // URL-encode path segments to prevent breaking the upstream request const encodedPath = path !== undefined && path !== '' @@ -104,38 +274,11 @@ export class KuboDataSource { const url = `${this.kuboUrl}/ipfs/${ipfsPath}${ format !== undefined ? `?format=${format}` : '' }`; + span.setAttribute('ipfs.url', url); - const span = startChildSpan( - 'KuboDataSource.getContent', - { - attributes: { - 'ipfs.cid': cidString, - 'ipfs.path': path ?? '', - 'ipfs.url': url, - }, - }, - parentSpan, - ); - - this.log.debug('Fetching IPFS content from Kubo', { - cidString, - path, - url, - }); + this.log.debug('Fetching IPFS content from Kubo', { cidString, path, url }); - // Connection-phase timeout - const controller = new AbortController(); - const connectionTimer = setTimeout(() => { - controller.abort(new Error('Kubo connection timeout')); - }, this.requestTimeoutMs); - - // Forward client abort to our controller - const onClientAbort = () => controller.abort(signal?.reason); - if (signal?.aborted) { - onClientAbort(); - } else if (signal) { - signal.addEventListener('abort', onClientAbort, { once: true }); - } + const { controller, detach } = this.setupAbort(signal); try { const response = await axios.get(url, { @@ -158,8 +301,7 @@ export class KuboDataSource { validateStatus: (status) => status < 500 || status === 504, }); - clearTimeout(connectionTimer); - signal?.removeEventListener('abort', onClientAbort); + detach(); if (response.status === 404) { (response.data as Readable).destroy(); @@ -201,90 +343,183 @@ export class KuboDataSource { const contentType = response.headers['content-type'] ?? 'application/octet-stream'; - // Switch from connection timeout to stall timeout - attachStallTimeout(stream, this.streamStallTimeoutMs, this.maxRequestMs); + return this.finalizeStream( + stream, + { + contentLength, + contentType, + statusCode: response.status, + contentRange: response.headers['content-range'], + }, + release, + span, + { cidString, path }, + ); + } catch (error: any) { + detach(); + // axios rejects for 5xx (validateStatus accepts <500 or 504). With + // responseType 'stream', error.response.data is an open Readable — destroy + // it so the socket/fd isn't leaked while Kubo returns 500/502/503. + const errStream = error?.response?.data; + if (errStream !== undefined && typeof errStream.destroy === 'function') { + errStream.destroy(); + } - span.setAttributes({ - 'ipfs.content_length': contentLength, - 'ipfs.content_type': contentType, - }); - span.addEvent('Kubo fetch successful'); + if (error instanceof IpfsNotFoundError) throw error; + if (error instanceof IpfsTimeoutError) throw error; + if (error instanceof IpfsRangeNotSatisfiableError) throw error; + + if (error.name === 'AbortError' || error.code === 'ERR_CANCELED') { + if (signal?.aborted) { + throw error; // Client disconnected + } + throw new IpfsTimeoutError( + `Kubo request timed out for /ipfs/${ipfsPath}`, + ); + } + + if (error.code === 'ECONNREFUSED') { + throw new IpfsUnavailableError( + `Kubo service unavailable at ${this.kuboUrl}`, + ); + } - this.log.debug('Kubo fetch successful', { + this.log.error('Failed to fetch from Kubo', { cidString, path, - contentLength, - contentType, + message: error.message, }); + throw error; + } + } - // End span when the stream terminates. 'close' is included because a - // destroy()-without-error (HEAD, rate-limited teardown, client abort) emits - // only 'close' — not 'end'/'error' — so without it the span never - // ends/exports. endSpan() is idempotent so a normal 'end'-then-'close' - // sequence ends exactly once. - let spanEnded = false; - const endSpan = () => { - if (spanEnded) return; - spanEnded = true; - span.end(); - }; - stream.on('end', endSpan); - stream.on('error', (err) => { - span.recordException(err); - endSpan(); - }); + // Local-only path: serve strictly from the local blockstore via the Kubo RPC + // API with `offline=true`, so a request NEVER triggers a public-IPFS/DHT walk. + // raw -> block/get (single verifiable block) + // car -> dag/export (verifiable DAG archive of the root) + // none -> cat (UnixFS bytes; offline cat also fails unless the WHOLE + // file's blocks are local — a free "holds the whole + // thing", not just the root, signal) + // A local miss returns fast (HTTP 500 with an "offline"/"not found locally" + // body) and is mapped to IpfsNotFoundError. Range is intentionally not honored + // here (the caller nulls it out under local-only); none of the local-only + // consumers (observer raw probe, peer CAR fetch) use Range. + private async getContentOffline( + { cidString, path, signal, format }: GetContentOptions, + span: Span, + release: () => void, + ): Promise { + if (this.kuboApiUrl === undefined) { + throw new IpfsUnavailableError( + 'Kubo RPC API URL is not configured; local-only fetch requires IPFS_KUBO_API_URL', + ); + } + + let endpoint: string; + let contentType: string; + let arg = cidString; + if (format === 'raw') { + endpoint = 'block/get'; + contentType = 'application/vnd.ipld.raw'; + } else if (format === 'car') { + endpoint = 'dag/export'; + contentType = 'application/vnd.ipld.car'; + } else { + endpoint = 'cat'; + // cat is the only offline endpoint that resolves a sub-path within the DAG. + if (path !== undefined && path !== '') { + const encodedPath = path + .split('/') + .map((seg) => encodeURIComponent(seg)) + .join('/'); + arg = `${cidString}/${encodedPath}`; + } + // Content-Type is not derivable from the RPC cat response; default to a + // safe binary type. Cache hits (served earlier in IpfsService) carry the + // correct type, and the load-bearing raw/car paths set it explicitly. + contentType = 'application/octet-stream'; + } + + const url = `${this.kuboApiUrl}/api/v0/${endpoint}`; + span.setAttribute('ipfs.url', `${url}?arg=${arg}&offline=true`); - // Release the concurrency slot and end the span when the response stream is - // fully consumed or destroyed (covers success, client abort, and errors). - stream.once('close', () => { - endSpan(); - release(); + this.log.debug('Fetching IPFS content from Kubo (local-only)', { + cidString, + path, + endpoint, + }); + + const { controller, detach } = this.setupAbort(signal); + + try { + const response = await axios.post(url, undefined, { + params: { arg, offline: true }, + responseType: 'stream', + signal: controller.signal, + headers: { 'Accept-Encoding': 'identity' }, + maxRedirects: 0, + // Inspect every status ourselves: an offline miss is a 500 we must + // translate to a fast IpfsNotFoundError rather than let axios reject. + validateStatus: () => true, }); - return { + detach(); + + if (response.status !== 200) { + // Drain the (small) error body to classify it and avoid leaking the fd. + const body = await collectStream(response.data as Readable, 8192); + if (OFFLINE_MISS_RE.test(body)) { + throw new IpfsNotFoundError( + `IPFS content not held locally: /ipfs/${arg}`, + ); + } + throw new IpfsUnavailableError( + `Unexpected Kubo RPC status ${response.status} for offline /ipfs/${arg}: ${body.slice(0, 200)}`, + ); + } + + const stream = response.data as Readable; + const rawContentLength = parseInt( + response.headers['content-length'] ?? '0', + 10, + ); + const contentLength = Number.isFinite(rawContentLength) + ? rawContentLength + : 0; + + return this.finalizeStream( stream, - size: contentLength, - contentType, - statusCode: response.status, - contentRange: response.headers['content-range'], - }; + { contentLength, contentType, statusCode: 200 }, + release, + span, + { cidString, path, localOnly: true }, + ); } catch (error: any) { - release(); - clearTimeout(connectionTimer); - signal?.removeEventListener('abort', onClientAbort); - // axios rejects for 5xx (validateStatus accepts <500 or 504). With - // responseType 'stream', error.response.data is an open Readable — destroy - // it so the socket/fd isn't leaked while Kubo returns 500/502/503. + detach(); const errStream = error?.response?.data; if (errStream !== undefined && typeof errStream.destroy === 'function') { errStream.destroy(); } - if (error.name !== 'AbortError') { - span.recordException(error); - } - span.end(); - if (error instanceof IpfsNotFoundError) throw error; - if (error instanceof IpfsTimeoutError) throw error; - if (error instanceof IpfsRangeNotSatisfiableError) throw error; + if (error instanceof IpfsUnavailableError) throw error; if (error.name === 'AbortError' || error.code === 'ERR_CANCELED') { if (signal?.aborted) { throw error; // Client disconnected } throw new IpfsTimeoutError( - `Kubo request timed out for /ipfs/${ipfsPath}`, + `Kubo RPC timed out for offline /ipfs/${arg}`, ); } if (error.code === 'ECONNREFUSED') { throw new IpfsUnavailableError( - `Kubo service unavailable at ${this.kuboUrl}`, + `Kubo RPC unavailable at ${this.kuboApiUrl}`, ); } - this.log.error('Failed to fetch from Kubo', { + this.log.error('Failed to fetch from Kubo (local-only)', { cidString, path, message: error.message, @@ -294,6 +529,31 @@ export class KuboDataSource { } } +// Read a Readable to a UTF-8 string, capped at maxBytes (destroys the stream +// once the cap is reached). Used to classify small Kubo RPC error bodies. +async function collectStream( + stream: Readable, + maxBytes: number, +): Promise { + const chunks: Buffer[] = []; + let total = 0; + try { + for await (const chunk of stream) { + const buf = Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk); + chunks.push(buf); + total += buf.length; + if (total >= maxBytes) { + stream.destroy(); + break; + } + } + } catch { + // A read error while draining the error body is non-fatal — return what we + // have so the caller can still classify it. + } + return Buffer.concat(chunks).toString('utf8'); +} + export class IpfsNotFoundError extends Error { constructor(message: string) { super(message); diff --git a/src/ipfs/sequential-ipfs-source.test.ts b/src/ipfs/sequential-ipfs-source.test.ts new file mode 100644 index 000000000..d9a53b790 --- /dev/null +++ b/src/ipfs/sequential-ipfs-source.test.ts @@ -0,0 +1,135 @@ +/** + * AR.IO Gateway + * Copyright (C) 2022-2025 Permanent Data Solutions, Inc. All Rights Reserved. + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +import { describe, it, beforeEach, mock } from 'node:test'; +import { strict as assert } from 'node:assert'; +import { Readable } from 'node:stream'; + +import { createTestLogger } from '../../test/test-logger.js'; +import { SequentialIpfsSource } from './sequential-ipfs-source.js'; +import { IpfsContentSource } from './ipfs-content-source.js'; +import { + KuboDataSource, + IpfsBlockedError, + IpfsNotFoundError, +} from './kubo-data-source.js'; + +const CID = 'bafybeigdyrzt5sfp7udm7hu76uh7y26nf3efuylqabf3oclgtqy55fbzdi'; + +const result = (contentType: string) => ({ + stream: Readable.from([Buffer.from('x')]), + size: 1, + contentType, + statusCode: 200, +}); + +describe('SequentialIpfsSource', () => { + const log = createTestLogger({ suite: 'SequentialIpfsSource' }); + + let kuboDataSource: KuboDataSource; + let peerDataSource: IpfsContentSource; + + beforeEach(() => { + kuboDataSource = { + getContent: mock.fn(async () => result('text/html')), + isHeldLocally: mock.fn(async () => false), + } as unknown as KuboDataSource; + + peerDataSource = { + getContent: mock.fn(async () => result('image/png')), + } as unknown as IpfsContentSource; + }); + + const build = (withPeer = true) => + new SequentialIpfsSource({ + log, + kuboDataSource, + peerDataSource: withPeer ? peerDataSource : undefined, + }); + + it('local-only runs tier 1 ONLY — never peers, never a presence check', async () => { + const r = await build().getContent({ cidString: CID, localOnly: true }); + r.stream.destroy(); + + assert.equal((kuboDataSource.getContent as any).mock.calls.length, 1); + assert.equal( + (kuboDataSource.getContent as any).mock.calls[0].arguments[0].localOnly, + true, + ); + assert.equal((kuboDataSource.isHeldLocally as any).mock.calls.length, 0); + assert.equal((peerDataSource.getContent as any).mock.calls.length, 0); + }); + + it('peer-fetch disabled → pure passthrough to Kubo (no presence check)', async () => { + const r = await build(false).getContent({ cidString: CID }); + r.stream.destroy(); + + assert.equal((kuboDataSource.getContent as any).mock.calls.length, 1); + assert.equal((kuboDataSource.isHeldLocally as any).mock.calls.length, 0); + }); + + it('normal request, held locally → serves via the gateway, never asks peers', async () => { + kuboDataSource.isHeldLocally = mock.fn(async () => true) as any; + + const r = await build().getContent({ cidString: CID }); + // Served via the gateway (correct content-type), not the offline octet-stream. + assert.equal(r.contentType, 'text/html'); + assert.equal((peerDataSource.getContent as any).mock.calls.length, 0); + assert.equal((kuboDataSource.getContent as any).mock.calls.length, 1); + r.stream.destroy(); + }); + + it('normal request, not held → acquires from a peer and serves', async () => { + const r = await build().getContent({ cidString: CID }); + + assert.equal((peerDataSource.getContent as any).mock.calls.length, 1); + assert.equal(r.contentType, 'image/png'); // returned by the peer source + // Gateway not used for serving (the peer source already re-served). + assert.equal((kuboDataSource.getContent as any).mock.calls.length, 0); + r.stream.destroy(); + }); + + it('normal request, not held, peers miss → falls through to public IPFS', async () => { + peerDataSource.getContent = mock.fn(async () => { + throw new IpfsNotFoundError('no peer holds it'); + }) as any; + + const r = await build().getContent({ cidString: CID }); + // Fell through to the gateway (tier 3, public). + assert.equal(r.contentType, 'text/html'); + assert.equal((kuboDataSource.getContent as any).mock.calls.length, 1); + r.stream.destroy(); + }); + + it('a blocked CID from a tier is re-thrown and does NOT fall through', async () => { + peerDataSource.getContent = mock.fn(async () => { + throw new IpfsBlockedError('blocked'); + }) as any; + + await assert.rejects( + () => build().getContent({ cidString: CID }), + (e: any) => e instanceof IpfsBlockedError, + ); + // Tier 3 (public) must NOT be reached for a blocked CID. + assert.equal((kuboDataSource.getContent as any).mock.calls.length, 0); + }); + + it('a genuine client abort short-circuits the cascade', async () => { + const controller = new AbortController(); + controller.abort(); + peerDataSource.getContent = mock.fn(async () => { + const e = new Error('aborted'); + e.name = 'AbortError'; + throw e; + }) as any; + + await assert.rejects( + () => build().getContent({ cidString: CID, signal: controller.signal }), + (e: any) => e.name === 'AbortError', + ); + assert.equal((kuboDataSource.getContent as any).mock.calls.length, 0); + }); +}); diff --git a/src/ipfs/sequential-ipfs-source.ts b/src/ipfs/sequential-ipfs-source.ts new file mode 100644 index 000000000..2e5babc30 --- /dev/null +++ b/src/ipfs/sequential-ipfs-source.ts @@ -0,0 +1,133 @@ +/** + * AR.IO Gateway + * Copyright (C) 2022-2025 Permanent Data Solutions, Inc. All Rights Reserved. + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +import winston from 'winston'; + +import { + IpfsContentSource, + IpfsContentSourceOptions, +} from './ipfs-content-source.js'; +import { + IpfsBlockedError, + IpfsContentResult, + IpfsRangeNotSatisfiableError, + IpfsSizeLimitError, + KuboDataSource, +} from './kubo-data-source.js'; + +/** + * Composite IPFS source realizing the serving order: + * + * 1. local Kubo (offline) — do we already hold it? fast, no network + * 2. peer AR.IO gateways — does a fleet peer hold it? verified CAR import + * 3. Kubo (public IPFS) — public DHT fallback existing behavior + * + * Design note (serving vs. acquisition): normal (non-local-only) requests are + * always SERVED through the Kubo gateway (:8080) so they get the correct sniffed + * Content-Type and a local-first serve. Tier 2's job is purely ACQUISITION — + * import a verified CAR into local Kubo — after which the gateway serves it + * locally. This avoids the offline-RPC's octet-stream default leaking into + * browser-facing responses, and it touches nothing in the ingress/proxy layer: + * the only added I/O is an internal Kubo RPC presence check and outbound peer + * HTTP fetches (the same shape existing Arweave peer sources already use). + * + * Local-only requests run tier 1 ONLY (genuinely offline) — never peers, never + * public — which is what prevents peer-fetch recursion and makes holding + * trustlessly measurable. When peer-fetch is disabled the composite is a pure + * passthrough to Kubo (zero behavior change). + */ +export class SequentialIpfsSource implements IpfsContentSource { + private log: winston.Logger; + private kuboDataSource: KuboDataSource; + private peerDataSource?: IpfsContentSource; + + constructor({ + log, + kuboDataSource, + peerDataSource, + }: { + log: winston.Logger; + kuboDataSource: KuboDataSource; + // Omit to disable peer-fetch (pure Kubo passthrough). + peerDataSource?: IpfsContentSource; + }) { + this.log = log.child({ class: this.constructor.name }); + this.kuboDataSource = kuboDataSource; + this.peerDataSource = peerDataSource; + } + + async getContent(opts: IpfsContentSourceOptions): Promise { + // Tier 1 ONLY: genuinely offline. Never peers, never public. This is the + // recursion guard and the holding-measurement primitive. + if (opts.localOnly === true) { + return this.kuboDataSource.getContent(opts); + } + + // Peer-fetch disabled → pure passthrough to Kubo (zero behavior change). + if (this.peerDataSource === undefined) { + return this.kuboDataSource.getContent(opts); + } + + opts.signal?.throwIfAborted(); + + // Tier 1: already held locally? Serve via the gateway (local-first, correct + // Content-Type, no public walk for content we hold). + if (await this.kuboDataSource.isHeldLocally(opts.cidString, opts.signal)) { + this.log.debug('IPFS content held locally, serving via gateway', { + cidString: opts.cidString, + }); + return this.kuboDataSource.getContent(opts); + } + + // Tier 2: not held → acquire from a fleet peer (verified CAR import) and + // serve. On a recoverable miss, fall through to public IPFS. + try { + return await this.peerDataSource.getContent(opts); + } catch (error: any) { + this.rethrowIfFatal(error, opts.signal); + this.log.debug('IPFS peer-fetch missed, falling through to public IPFS', { + cidString: opts.cidString, + message: error?.message, + }); + } + + // Tier 3: public IPFS via the gateway (existing behavior). + return this.kuboDataSource.getContent(opts); + } + + // Errors that must NOT fall through to the next tier: + // - a genuine client disconnect (short-circuit the whole cascade); + // - moderation (a blocked CID stays blocked across every tier); + // - size / range errors (not an availability miss). + // Everything else (NotFound / Timeout / Unavailable / transport) is a + // recoverable miss and falls through. + private rethrowIfFatal(error: any, signal?: AbortSignal): void { + if (error?.name === 'AbortError' && signal?.aborted === true) { + throw error; + } + if ( + error instanceof IpfsBlockedError || + error instanceof IpfsSizeLimitError || + error instanceof IpfsRangeNotSatisfiableError + ) { + throw error; + } + } +} + +/** + * Wraps a KuboDataSource to force `localOnly: true` on every call — a fixed + * tier-1 (offline) source. Retained as a reusable primitive (e.g. for an + * explicit tier list or an observer holding-probe wiring); the composite above + * reaches tier 1 directly via KuboDataSource, so this is not required by it. + */ +export class LocalOnlyKuboSource implements IpfsContentSource { + constructor(private inner: KuboDataSource) {} + + getContent(opts: IpfsContentSourceOptions): Promise { + return this.inner.getContent({ ...opts, localOnly: true }); + } +} diff --git a/src/metrics.ts b/src/metrics.ts index 06131a69e..29d29ff2a 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -1715,6 +1715,31 @@ export const ipfsBlockedTotal = new promClient.Counter({ help: 'IPFS requests blocked by CID blocklist', }); +// Peer-fetch (fleet durability layer) metrics. +export const ipfsPeerFetchTotal = new promClient.Counter({ + name: 'ipfs_peer_fetch_total', + help: 'IPFS peer-fetch outcomes: a CID acquired from a fleet peer, or a miss', + labelNames: ['result'] as const, // 'success' | 'miss' +}); + +export const ipfsPeerFetchPeerAttemptsTotal = new promClient.Counter({ + name: 'ipfs_peer_fetch_peer_attempts_total', + help: 'Per-peer IPFS peer-fetch attempts by outcome', + // 'success' | 'import_verify_failed' | 'peer_error' + labelNames: ['result'] as const, +}); + +export const ipfsPeerFetchCarBytesTotal = new promClient.Counter({ + name: 'ipfs_peer_fetch_car_bytes_total', + help: 'Total CAR bytes imported into local Kubo via IPFS peer-fetch', +}); + +export const ipfsLocalOnlyServeTotal = new promClient.Counter({ + name: 'ipfs_local_only_serve_total', + help: 'Inbound local-only IPFS serve outcomes (holding probes / peer serves)', + labelNames: ['result'] as const, // 'hit' | 'miss' +}); + // // Chunk metadata anchor (offset → tx + data_root via reference peer // `/chunk/{offset}/data` headers, cross-checked against the chain). diff --git a/src/routes/ipfs.test.ts b/src/routes/ipfs.test.ts index 331b0d039..354831f5a 100644 --- a/src/routes/ipfs.test.ts +++ b/src/routes/ipfs.test.ts @@ -38,14 +38,17 @@ function makeInfiniteStream(): Readable { function makeApp({ service, setArns = false, + paymentProcessor, }: { service: Partial; setArns?: boolean; + paymentProcessor?: any; }): express.Express { const app = express(); const handler = createIpfsHandler({ log, ipfsService: service as IpfsService, + paymentProcessor, }); const setCtx: express.Handler = (req, _res, next) => { (req as any).ipfsCid = (req.params as any).cid; @@ -65,6 +68,18 @@ async function counterValue(counter: { return m.values.reduce((sum, v) => sum + v.value, 0); } +async function labeledCounter( + counter: { get: () => Promise<{ values: { labels: any; value: number }[] }> }, + labels: Record, +): Promise { + const m = await counter.get(); + return m.values + .filter((v) => + Object.entries(labels).every(([k, val]) => v.labels[k] === val), + ) + .reduce((sum, v) => sum + v.value, 0); +} + const waitFor = async (pred: () => boolean, ms = 2000) => { const start = Date.now(); while (!pred() && Date.now() - start < ms) { @@ -221,4 +236,110 @@ describe('IPFS route handler', () => { await new Promise((r) => server.close(() => r())); }); }); + + describe('local-only serve mode', () => { + it('threads localOnly:true and nulls Range from the X-Ar-Io-Local-Only header, echoing the marker on a hit', async () => { + const getContent = mock.fn(async () => okResult()); + const hitBefore = await labeledCounter(metrics.ipfsLocalOnlyServeTotal, { + result: 'hit', + }); + const res = await request(makeApp({ service: { getContent } })) + .get(`/c/${CID}`) + .set('X-Ar-Io-Local-Only', 'true') + .set('Range', 'bytes=0-9') + .expect(200); + + const args = getContent.mock.calls[0].arguments[0] as any; + assert.equal(args.localOnly, true); + // Range is not honored under local-only. + assert.equal(args.range, undefined); + // The marker is echoed so a peer/observer can assert the mode was honored. + assert.equal(res.headers['x-ar-io-local-only'], 'true'); + // A local-only hit is metered. + assert.equal( + (await labeledCounter(metrics.ipfsLocalOnlyServeTotal, { + result: 'hit', + })) - hitBefore, + 1, + ); + }); + + it('accepts ?local=1 as a local-only trigger', async () => { + const getContent = mock.fn(async () => okResult()); + await request(makeApp({ service: { getContent } })) + .get(`/c/${CID}?local=1`) + .expect(200); + + assert.equal( + (getContent.mock.calls[0].arguments[0] as any).localOnly, + true, + ); + }); + + it('returns 404 on a local miss with no fallback, metering the miss', async () => { + const service = { + getContent: mock.fn(async () => { + throw new IpfsNotFoundError('not held locally'); + }), + }; + const missBefore = await labeledCounter(metrics.ipfsLocalOnlyServeTotal, { + result: 'miss', + }); + await request(makeApp({ service })) + .get(`/c/${CID}`) + .set('X-Ar-Io-Local-Only', 'true') + .expect(404); + assert.equal( + (await labeledCounter(metrics.ipfsLocalOnlyServeTotal, { + result: 'miss', + })) - missBefore, + 1, + ); + }); + + it('does not set localOnly or echo the marker for a normal request', async () => { + const getContent = mock.fn(async () => okResult()); + const res = await request(makeApp({ service: { getContent } })) + .get(`/c/${CID}`) + .expect(200); + + assert.equal( + (getContent.mock.calls[0].arguments[0] as any).localOnly, + false, + ); + assert.equal(res.headers['x-ar-io-local-only'], undefined); + }); + + it('bypasses payment for local-only but still consults it for a normal request', async () => { + const paymentProcessor = { + isBrowserRequest: mock.fn(() => false), + calculateRequirements: mock.fn(() => ({ maxAmountRequired: '0' })), + extractPayment: mock.fn(() => undefined), + }; + + // Normal request: the payment processor IS consulted. + await request( + makeApp({ + service: { getContent: mock.fn(async () => okResult()) }, + paymentProcessor, + }), + ) + .get(`/c/${CID}`) + .expect(200); + assert.ok(paymentProcessor.calculateRequirements.mock.calls.length >= 1); + + // Local-only request: the processor is passed undefined, so it is bypassed. + paymentProcessor.calculateRequirements.mock.resetCalls(); + await request( + makeApp({ + service: { getContent: mock.fn(async () => okResult()) }, + paymentProcessor, + }), + ) + .get(`/c/${CID}`) + .set('X-Ar-Io-Local-Only', 'true') + .expect(200); + assert.equal(paymentProcessor.calculateRequirements.mock.calls.length, 0); + }); + }); }); diff --git a/src/routes/ipfs.ts b/src/routes/ipfs.ts index d7e6ad38d..41f60097b 100644 --- a/src/routes/ipfs.ts +++ b/src/routes/ipfs.ts @@ -222,6 +222,12 @@ async function handleIpfsRequest({ : undefined; // Trustless format takes precedence over Range: verifiable block/CAR retrieval. const format = parseIpfsFormat(req); + // Local-only serve mode: resolve ONLY from the local blockstore, never public + // IPFS. Used by peer gateways (peer-fetch recursion guard) and the observer + // holding-probe. `?local=1` is accepted as a testing convenience. + const localOnly = + req.headers[headerNames.ipfsLocalOnly.toLowerCase()] === 'true' || + req.query.local === '1'; const ipfsPath = path !== undefined ? `${cidString}/${path}` : cidString; parentLog.debug('Handling IPFS request', { @@ -229,6 +235,7 @@ async function handleIpfsRequest({ path, routeType, format, + localOnly, }); try { @@ -236,8 +243,11 @@ async function handleIpfsRequest({ cidString, path, signal: req.signal, - range: format !== undefined ? undefined : rangeForKubo, + // Range is not honored under local-only (the offline RPC path serves the + // full object); none of the local-only consumers combine Range with it. + range: format !== undefined || localOnly ? undefined : rangeForKubo, format, + localOnly, }); // Check payment and rate limits (x402 + rate limiting in one call). @@ -256,7 +266,10 @@ async function handleIpfsRequest({ clientIps: extractAllClientIPs(req).clientIps, }, rateLimiter, - paymentProcessor, + // Local-only requests are intra-fleet (peers) / observation traffic and + // bypass payment (402); rate limiting still applies (passing undefined + // disables only the payment check, mirroring the no-payment config). + paymentProcessor: localOnly ? undefined : paymentProcessor, }); if (!limitCheck.allowed) { @@ -286,6 +299,12 @@ async function handleIpfsRequest({ res.setHeader('ETag', `"${etag}"`); res.setHeader('X-Ipfs-Path', `/ipfs/${ipfsPath}`); res.setHeader(headerNames.arIoSource, 'ipfs'); + // Echo the local-only marker on a hit so a peer/observer can assert the + // server honored the mode (a proxy would have 404'd on a local miss). + if (localOnly) { + res.setHeader(headerNames.ipfsLocalOnly, 'true'); + metrics.ipfsLocalOnlyServeTotal.inc({ result: 'hit' }); + } // The representation is content-negotiated: the same URL yields a UnixFS // proxy body, a raw block, or a CAR depending on the Accept header (see // parseIpfsFormat). Tell shared caches to key on Accept so they don't serve @@ -447,6 +466,11 @@ async function handleIpfsRequest({ } if (error instanceof IpfsNotFoundError) { + // A local-only miss is the "gateway does not hold this" signal (the + // observer's holding probe / a peer's local-only fetch). + if (localOnly) { + metrics.ipfsLocalOnlyServeTotal.inc({ result: 'miss' }); + } metrics.ipfsRequestsTotal.inc({ route_type: routeType, status: 'not_found', diff --git a/src/system.ts b/src/system.ts index 37b114d7a..956422988 100644 --- a/src/system.ts +++ b/src/system.ts @@ -1882,6 +1882,8 @@ import { IpfsFsCache } from './ipfs/ipfs-cache.js'; import { IpfsService } from './ipfs/ipfs-service.js'; import { IpfsPinner } from './ipfs/ipfs-pinner.js'; import { createIpfsRateLimiter } from './ipfs/ipfs-rate-limiter.js'; +import { IpfsPeerDataSource } from './ipfs/ipfs-peer-data-source.js'; +import { SequentialIpfsSource } from './ipfs/sequential-ipfs-source.js'; import { RateLimiter } from './limiter/types.js'; export let ipfsService: IpfsService | undefined; @@ -1894,6 +1896,9 @@ if (config.IPFS_ENABLED) { const kuboDataSource = new KuboDataSource({ log, kuboUrl: config.IPFS_KUBO_URL, + // RPC API base — required for local-only (offline) reads; the read-only + // :8080 gateway has no per-request offline flag. + kuboApiUrl: config.IPFS_KUBO_API_URL, requestTimeoutMs: config.IPFS_KUBO_REQUEST_TIMEOUT_MS, streamStallTimeoutMs: config.IPFS_STREAM_STALL_TIMEOUT_MS, maxConcurrent: config.IPFS_KUBO_MAX_CONCURRENT_REQUESTS, @@ -1934,9 +1939,33 @@ if (config.IPFS_ENABLED) { metricsSource: 'ipfs', }); + // Peer-fetch durability layer (tier 2). Built only when enabled; otherwise the + // composite below is a pure passthrough to Kubo (zero behavior change). + const ipfsPeerDataSource = config.IPFS_PEER_FETCH_ENABLED + ? new IpfsPeerDataSource({ + log, + peerManager: arIOPeerManager, + kuboApiUrl: config.IPFS_KUBO_API_URL, + kuboDataSource, + peerCount: config.IPFS_PEER_FETCH_COUNT, + requestTimeoutMs: config.IPFS_PEER_FETCH_TIMEOUT_MS, + maxCarBytes: config.IPFS_PEER_FETCH_MAX_CAR_BYTES, + pinRoots: config.IPFS_PIN_ARNS_CONTENT, + staticPeers: config.IPFS_PEER_FETCH_STATIC_PEERS, + }) + : undefined; + + // Composite: local Kubo (offline) → fleet peers (verified CAR import) → public + // IPFS. Local-only requests run tier 1 only. + const ipfsCompositeSource = new SequentialIpfsSource({ + log, + kuboDataSource, + peerDataSource: ipfsPeerDataSource, + }); + ipfsService = new IpfsService({ log, - dataSource: kuboDataSource, + dataSource: ipfsCompositeSource, cache: ipfsCache, blockListValidator: dataBlockListValidator, maxResponseSizeBytes: config.IPFS_MAX_RESPONSE_SIZE_BYTES, diff --git a/test/end-to-end/ipfs-peer-fetch.test.ts b/test/end-to-end/ipfs-peer-fetch.test.ts new file mode 100644 index 000000000..d3db23d43 --- /dev/null +++ b/test/end-to-end/ipfs-peer-fetch.test.ts @@ -0,0 +1,250 @@ +/** + * AR.IO Gateway + * Copyright (C) 2022-2025 Permanent Data Solutions, Inc. All Rights Reserved. + * + * SPDX-License-Identifier: AGPL-3.0-or-later + */ +import { strict as assert } from 'node:assert'; +import { after, before, describe, it } from 'node:test'; +import { + GenericContainer, + StartedTestContainer, + Network, + StartedNetwork, + Wait, +} from 'testcontainers'; +import axios from 'axios'; + +import { getCoreContainer } from './utils.js'; + +const KUBO_IMAGE = 'ipfs/kubo:v0.32.1'; +const SEED_CONTENT = 'hello-fleet-durability-peer-fetch'; + +// A tiny HTTP server (run on the built `core` image so no extra image is pulled) +// that plays a LYING peer: it answers every /ipfs/... request with 200 + a CAR +// that does NOT hash to the requested CID, and counts how many times it was +// asked so tests can prove (a) the fleet fell through past it, and (b) local-only +// requests never recurse to it. +const MALICIOUS_PEER_SCRIPT = ` +const http = require('http'); +let carHits = 0; +http.createServer((req, res) => { + if (req.url.startsWith('/__count')) { + res.writeHead(200, { 'content-type': 'application/json' }); + res.end(JSON.stringify({ carHits })); + return; + } + if (req.url.startsWith('/ipfs/')) { + carHits++; + res.writeHead(200, { 'content-type': 'application/vnd.ipld.car' }); + res.end('not-a-valid-car-block-'.repeat(64)); + return; + } + res.writeHead(404); + res.end(); +}).listen(3000, '0.0.0.0', () => console.log('malicious peer listening on 3000')); +`; + +const extractCid = (execOutput: string): string => { + const match = execOutput.match(/\bba[a-z2-7]{20,}\b/); + if (match === null) { + throw new Error(`could not parse CID from: ${execOutput}`); + } + return match[0]; +}; + +// Heavy: builds the core image and starts 2 core + 2 kubo + 1 malicious-peer +// containers. Gated behind `test:e2e`. Proves fleet durability independent of +// public IPFS (kubos run --offline, so a 200 on the cold node can ONLY have come +// from a peer), trustless import (a lying peer is rejected), the holding flip +// (local-only 404 -> 200 after a fetch), and the recursion guard. +describe('IPFS peer-fetch durability layer', () => { + let network: StartedNetwork; + let kuboA: StartedTestContainer; + let kuboB: StartedTestContainer; + let coreA: StartedTestContainer; + let coreB: StartedTestContainer; + let malicious: StartedTestContainer; + let coreAUrl: string; + let coreBUrl: string; + let maliciousUrl: string; + let cid: string; + let unheldCid: string; + + before(async () => { + network = await new Network().start(); + + // Two Kubo nodes, both --offline so they CANNOT reach public IPFS. The only + // way cold kubo-b can obtain content is an HTTP CAR fetch from a peer gateway. + const startKubo = (alias: string) => + new GenericContainer(KUBO_IMAGE) + .withNetwork(network) + .withNetworkAliases(alias) + .withCommand(['daemon', '--offline']) + .withWaitStrategy(Wait.forLogMessage(/Daemon is ready/)) + .withStartupTimeout(120_000) + .start(); + + [kuboA, kuboB] = await Promise.all([ + startKubo('kubo-a'), + startKubo('kubo-b'), + ]); + + // Seed CID X into kubo-a only. + const add = await kuboA.exec([ + 'sh', + '-c', + `echo -n "${SEED_CONTENT}" | ipfs add -q --cid-version=1`, + ]); + cid = extractCid(add.output); + + // A valid CID that was never added anywhere (for the recursion-guard test). + const onlyHash = await kuboB.exec([ + 'sh', + '-c', + `echo -n "never-added-to-any-node-xyz" | ipfs add -qn --cid-version=1`, + ]); + unheldCid = extractCid(onlyHash.output); + + // Build the core image from the worktree (includes the peer-fetch changes) + // and tag it `core`; reused for both cores and the malicious peer. + await getCoreContainer(); + + const baseCoreEnv = { + START_WRITERS: 'false', + ADMIN_API_KEY: 'secret', + IPFS_ENABLED: 'true', + LOG_LEVEL: 'info', + }; + + coreA = await new GenericContainer('core') + .withEnvironment({ + ...baseCoreEnv, + IPFS_KUBO_URL: 'http://kubo-a:8080', + IPFS_KUBO_API_URL: 'http://kubo-a:5001', + // core-a only serves what it holds locally; no peer-fetch needed. + IPFS_PEER_FETCH_ENABLED: 'false', + }) + .withNetwork(network) + .withNetworkAliases('core-a') + .withExposedPorts(4000) + .withWaitStrategy(Wait.forHttp('/ar-io/info', 4000)) + .withStartupTimeout(180_000) + .start(); + + malicious = await new GenericContainer('core') + // The core image is distroless (no node on PATH); use the absolute path + // its own entrypoint uses. + .withEntrypoint(['/nodejs/bin/node', '-e', MALICIOUS_PEER_SCRIPT]) + .withNetwork(network) + .withNetworkAliases('malicious') + .withExposedPorts(3000) + // forListeningPorts() does an internal port probe via a shell command the + // distroless image can't run; wait on the readiness log line instead. + .withWaitStrategy(Wait.forLogMessage(/malicious peer listening/)) + .withStartupTimeout(60_000) + .start(); + + coreB = await new GenericContainer('core') + .withEnvironment({ + ...baseCoreEnv, + IPFS_KUBO_URL: 'http://kubo-b:8080', + IPFS_KUBO_API_URL: 'http://kubo-b:5001', + IPFS_PEER_FETCH_ENABLED: 'true', + // The lying peer is tried FIRST, then the honest core-a. A correct + // result therefore also proves the untrusted peer is rejected safely. + IPFS_PEER_FETCH_STATIC_PEERS: + 'http://malicious:3000,http://core-a:4000', + IPFS_PEER_FETCH_TIMEOUT_MS: '20000', + }) + .withNetwork(network) + .withNetworkAliases('core-b') + .withExposedPorts(4000) + .withWaitStrategy(Wait.forHttp('/ar-io/info', 4000)) + .withStartupTimeout(180_000) + .start(); + + coreAUrl = `http://localhost:${coreA.getMappedPort(4000)}`; + coreBUrl = `http://localhost:${coreB.getMappedPort(4000)}`; + maliciousUrl = `http://localhost:${malicious.getMappedPort(3000)}`; + }); + + after(async () => { + await coreB?.stop(); + await malicious?.stop(); + await coreA?.stop(); + await kuboB?.stop(); + await kuboA?.stop(); + await network?.stop(); + }); + + const maliciousCarHits = async (): Promise => { + const res = await axios.get(`${maliciousUrl}/__count`); + return res.data.carHits as number; + }; + + it('core-a holds and serves the seeded CID; core-b does not hold it yet', async () => { + // core-a serves it (it holds it locally). + const onA = await axios.get(`${coreAUrl}/ipfs/${cid}`, { + responseType: 'text', + validateStatus: () => true, + }); + assert.equal(onA.status, 200); + assert.equal(onA.data, SEED_CONTENT); + + // core-b, probed local-only, does NOT hold it yet → 404. + const probeBefore = await axios.get(`${coreBUrl}/ipfs/${cid}`, { + headers: { 'X-Ar-Io-Local-Only': 'true' }, + validateStatus: () => true, + }); + assert.equal(probeBefore.status, 404); + }); + + it('core-b peer-fetches the CID from the fleet, past a lying peer, and serves correct bytes', async () => { + const hitsBefore = await maliciousCarHits(); + + const res = await axios.get(`${coreBUrl}/ipfs/${cid}`, { + responseType: 'text', + validateStatus: () => true, + }); + + // Correct bytes — even though kubo-b is offline (no public IPFS), so this + // could ONLY have come from the fleet (core-a). + assert.equal(res.status, 200); + assert.equal(res.data, SEED_CONTENT); + + // The lying peer was tried first (its tampered CAR failed Kubo's import + // verification) and the fleet fell through to the honest peer. + const hitsAfter = await maliciousCarHits(); + assert.ok( + hitsAfter > hitsBefore, + 'expected the malicious peer to have been tried and rejected', + ); + }); + + it('core-b now holds the CID: the local-only probe flips 404 → 200', async () => { + const probeAfter = await axios.get(`${coreBUrl}/ipfs/${cid}`, { + headers: { 'X-Ar-Io-Local-Only': 'true' }, + responseType: 'text', + validateStatus: () => true, + }); + assert.equal(probeAfter.status, 200); + assert.equal(probeAfter.data, SEED_CONTENT); + // The server honored local-only (echoed the marker). + assert.equal(probeAfter.headers['x-ar-io-local-only'], 'true'); + }); + + it('recursion guard: a local-only miss returns 404 and never contacts peers', async () => { + const hitsBefore = await maliciousCarHits(); + + const res = await axios.get(`${coreBUrl}/ipfs/${unheldCid}`, { + headers: { 'X-Ar-Io-Local-Only': 'true' }, + validateStatus: () => true, + }); + assert.equal(res.status, 404); + + // A local-only request must run tier 1 ONLY — no outbound peer fetch. + const hitsAfter = await maliciousCarHits(); + assert.equal(hitsAfter, hitsBefore); + }); +});