feat: indexed aggregates behind --indexed-aggregates

packages/cli/src/aggregate-stream.ts

@@ -0,0 +1,164 @@
+/**
+ * Stream an indexed aggregate's full byte stream to a provider pull.
+ *
+ * The response is the exact byte layout the aggregate root commits to (see
+ * `ipfs2foc-core/indexed-aggregate`): each sub-piece's canonical CAR at its
+ * unpadded offset with zero fill, then the embedded data segment index. The
+ * provider recomputes the piece commitment over these bytes on pull, so every
+ * sub-piece is hash-verified here as it passes through — a drifted gateway
+ * response aborts the stream instead of feeding the provider bytes that can
+ * only fail its commP check after the whole download.
+ */
+
+import { once } from 'node:events'
+import { createReadStream } from 'node:fs'
+import type { ServerResponse } from 'node:http'
+import * as Hasher from '@web3-storage/data-segment/multihash'
+import { buildIndexedAggregate, type IndexedAggregate } from 'ipfs2foc-core/indexed-aggregate'
+import type { CID } from 'multiformats/cid'
+import * as Raw from 'multiformats/codecs/raw'
+import * as Link from 'multiformats/link'
+import type { MigrationDB } from './db.ts'
+import { fetchCanonicalCar } from './gateway-blocks.ts'
+import { log } from './util.ts'
+
+const ZERO_CHUNK = new Uint8Array(64 * 1024)
+
+async function writeChunk(res: ServerResponse, chunk: Uint8Array): Promise<void> {
+  if (!res.write(chunk)) {
+    await once(res, 'drain')
+  }
+}
+
+async function writeZeros(res: ServerResponse, length: number): Promise<void> {
+  let remaining = length
+  while (remaining > 0) {
+    const n = Math.min(remaining, ZERO_CHUNK.length)
+    await writeChunk(res, n === ZERO_CHUNK.length ? ZERO_CHUNK : ZERO_CHUNK.subarray(0, n))
+    remaining -= n
+  }
+}
+
+/**
+ * Recompute the indexed aggregate layout for a planned aggregate row. The
+ * layout is deterministic from the member set (the builder re-sorts and picks
+ * the smallest fitting deal size), so nothing beyond the members needs to be
+ * persisted; the recomputed root is checked against the stored one before any
+ * byte is served.
+ */
+export function layoutForAggregate(db: MigrationDB, idx: number, expectedRoot: string): IndexedAggregate {
+  const members = db.aggregateManifest(idx)
+  const layout = buildIndexedAggregate(members.map((m) => ({ pieceCid: m.pieceCid, rawSize: m.rawSize })))
+  if (layout.rootPieceCid !== expectedRoot) {
+    throw new Error(
+      `recomputed indexed aggregate root ${layout.rootPieceCid} does not match stored root ${expectedRoot}`
+    )
+  }
+  return layout
+}
+
+/**
+ * Pipe one sub-piece's canonical CAR into the response while recomputing its
+ * piece commitment. Bytes come from the assembled CAR on disk when the
+ * sub-piece has one, otherwise from the block-verified canonical gateway
+ * stream (`fetchCanonicalCar`) — the same bytes the plan-time commitment was
+ * computed over. Returns the byte count written.
+ */
+async function pipeSubPiece(
+  db: MigrationDB,
+  res: ServerResponse,
+  pieceCid: string,
+  expectedLength: number
+): Promise<number> {
+  const subPiece = db.subPieceByCid(pieceCid)
+  if (subPiece == null || subPiece.status !== 'built') {
+    throw new Error(`sub-piece ${pieceCid} is ${subPiece == null ? 'unknown' : 'not built'}`)
+  }
+
+  const hasher = Hasher.create()
+  let written = 0
+
+  const consume = async (chunks: AsyncIterable<Uint8Array>): Promise<void> => {
+    for await (const chunk of chunks) {
+      hasher.write(chunk)
+      written += chunk.length
+      if (written > expectedLength) {
+        throw new Error(`sub-piece ${pieceCid} produced more than the expected ${expectedLength} bytes`)
+      }
+      await writeChunk(res, chunk)
+    }
+  }
+
+  if (subPiece.carPath != null) {
+    await consume(createReadStream(subPiece.carPath))
+  } else if (subPiece.url != null && subPiece.url !== '') {
+    const carUrl = new URL(subPiece.url)
+    const sourceCid = carUrl.pathname.replace(/^\/ipfs\//, '')
+    const { body } = await fetchCanonicalCar(carUrl.origin, sourceCid)
+    await consume(body as unknown as AsyncIterable<Uint8Array>)
+  } else {
+    throw new Error(`sub-piece ${pieceCid} has neither a CAR file nor a gateway URL`)
+  }
+
+  if (written !== expectedLength) {
+    throw new Error(`sub-piece ${pieceCid} produced ${written} bytes, expected ${expectedLength}`)
+  }
+  const recomputed = (Link.create(Raw.code, hasher.digest()) as CID).toString()
+  if (recomputed !== pieceCid) {
+    throw new Error(`sub-piece bytes recomputed to ${recomputed}, expected ${pieceCid}`)
+  }
+  return written
+}
+
+/**
+ * Answer a pull for an indexed aggregate root: `Content-Length` up front, then
+ * the regions in stream order. A mid-stream failure destroys the socket — the
+ * provider sees a transport abort, never a clean-looking short body.
+ */
+export async function serveIndexedAggregate(
+  db: MigrationDB,
+  idx: number,
+  expectedRoot: string,
+  res: ServerResponse,
+  head: boolean
+): Promise<void> {
+  let layout: IndexedAggregate
+  try {
+    layout = layoutForAggregate(db, idx, expectedRoot)
+  } catch (err) {
+    const message = err instanceof Error ? err.message : String(err)
+    log(`aggregate ${idx}: ${message}`)
+    res.writeHead(500, { 'content-type': 'text/plain' })
+    res.end('aggregate layout unavailable')
+    return
+  }
+
+  res.writeHead(200, {
+    'content-type': 'application/octet-stream',
+    'content-length': String(layout.streamLength),
+    'cache-control': 'no-store',
+  })
+  if (head) {
+    res.end()
+    return
+  }
+
+  try {
+    for (const region of layout.regions) {
+      if (region.kind === 'zero') {
+        await writeZeros(res, region.length)
+      } else if (region.kind === 'index') {
+        await writeChunk(res, layout.indexBytes)
+      } else {
+        const member = layout.members[region.memberIndex]
+        const written = await pipeSubPiece(db, res, member.pieceCid, region.payloadLength)
+        await writeZeros(res, region.length - written)
+      }
+    }
+    res.end()
+  } catch (err) {
+    const message = err instanceof Error ? err.message : String(err)
+    log(`aggregate ${idx}: stream aborted — ${message}`)
+    res.destroy()
+  }
+}

packages/cli/src/aggregate.ts

@@ -9,17 +9,22 @@
  * re-derives on add.
  */
 
+import { Index } from '@web3-storage/data-segment'
 import * as Piece from '@web3-storage/data-segment/piece'
+import { buildIndexedAggregate } from 'ipfs2foc-core/indexed-aggregate'
 import { pieceAggregateCommP } from 'ipfs2foc-core/piece-aggregate'
 import type { PieceResult } from './piece.ts'
 
 const NODE_SIZE = 32n
+const INDEX_ENTRY_SIZE = 64n
 
 export interface AggregatePlan {
   index: number
   /** Aggregate root PieceCID v2 (aggregate piece commitment). */
   rootPieceCid: string
   members: PieceResult[]
+  /** True when the root is an indexed (data segment) aggregate. */
+  indexed: boolean
 }
 
 export interface PackResult {
@@ -51,7 +56,7 @@ export function packAggregates(pieces: PieceResult[], aggregateSizeBytes: bigint
       return
     }
     const root = pieceAggregateCommP(members.map((m) => ({ pieceCid: m.pieceCid, rawSize: m.rawSize }))).rootPieceCid
-    aggregates.push({ index: aggregates.length, rootPieceCid: root, members })
+    aggregates.push({ index: aggregates.length, rootPieceCid: root, members, indexed: false })
     members = []
     used = 0n
   }
@@ -72,3 +77,55 @@ export function packAggregates(pieces: PieceResult[], aggregateSizeBytes: bigint
   flush()
   return { aggregates, oversized }
 }
+
+/**
+ * Pack pieces into indexed (data segment) aggregates. The packing budget
+ * reserves the index area up front — `maxIndexEntriesInDeal(dealSize) * 64`
+ * padded bytes at the deal's tail — so a full group still fits the deal size
+ * with its index. Groups of one skip the wrapper entirely: the piece's own
+ * CID is the aggregate root and it is pulled/added as itself (an index over
+ * a single entry is rejected by the provider's aggregate parsing).
+ */
+export function packIndexedAggregates(pieces: PieceResult[], aggregateSizeBytes: bigint): PackResult {
+  const maxEntries = BigInt(Index.maxIndexEntriesInDeal(aggregateSizeBytes))
+  const budget = aggregateSizeBytes - maxEntries * INDEX_ENTRY_SIZE
+
+  const aggregates: AggregatePlan[] = []
+  const oversized: PieceResult[] = []
+
+  let members: PieceResult[] = []
+  let used = 0n
+
+  const flush = (): void => {
+    if (members.length === 0) {
+      return
+    }
+    if (members.length === 1) {
+      aggregates.push({ index: aggregates.length, rootPieceCid: members[0].pieceCid, members, indexed: false })
+    } else {
+      const root = buildIndexedAggregate(
+        members.map((m) => ({ pieceCid: m.pieceCid, rawSize: m.rawSize })),
+        { maxDealSize: aggregateSizeBytes }
+      ).rootPieceCid
+      aggregates.push({ index: aggregates.length, rootPieceCid: root, members, indexed: true })
+    }
+    members = []
+    used = 0n
+  }
+
+  for (const piece of pieces) {
+    const size = paddedSize(piece.pieceCid)
+    if (size > budget) {
+      oversized.push(piece)
+      continue
+    }
+    if (used + size > budget || BigInt(members.length) >= maxEntries) {
+      flush()
+    }
+    members.push(piece)
+    used += size
+  }
+
+  flush()
+  return { aggregates, oversized }
+}

packages/cli/src/db.ts

@@ -103,6 +103,12 @@ export interface AggregateRow {
   /** Aggregate root PieceCID v2 — the parent CID added on-chain for the aggregate. */
   rootPieceCid: string
   pieceSizeBytes: string
+  /**
+   * True when the aggregate embeds a data segment index (single-piece add;
+   * the provider pulls one assembled stream). False is the bare layout whose
+   * sub-pieces are pulled individually.
+   */
+  indexed: boolean
   status: AggregateStatus
   /** Synthetic per-aggregate pull marker, set when submission begins. */
   pullId: string | null
@@ -162,6 +168,7 @@ export class MigrationDB {
         idx              INTEGER PRIMARY KEY,
         root_piece_cid   TEXT NOT NULL,
         piece_size_bytes TEXT NOT NULL,
+        indexed          INTEGER NOT NULL DEFAULT 0,
         status           TEXT NOT NULL DEFAULT 'planned',
         pull_id          TEXT,
         data_set_id      TEXT,
@@ -366,10 +373,10 @@ export class MigrationDB {
    * a sub-piece — single-asset source CIDs become 1-member passthrough
    * sub-pieces at plan time, so the pull/add path has one canonical shape.
    */
-  saveAggregate(idx: number, rootPieceCid: string, pieceSizeBytes: bigint, members: string[]): void {
+  saveAggregate(idx: number, rootPieceCid: string, pieceSizeBytes: bigint, members: string[], indexed = false): void {
     const aggregateStmt = this.#db.prepare(
-      `INSERT INTO aggregates (idx, root_piece_cid, piece_size_bytes, status, created_at)
-       VALUES (?, ?, ?, 'planned', ?)`
+      `INSERT INTO aggregates (idx, root_piece_cid, piece_size_bytes, indexed, status, created_at)
+       VALUES (?, ?, ?, ?, 'planned', ?)`
     )
     const memberStmt = this.#db.prepare(
       `INSERT INTO aggregate_members (aggregate_idx, segment_index, sub_piece_cid)
@@ -381,7 +388,7 @@ export class MigrationDB {
     // recordBuiltSubPiece / recordPassthroughSubPiece writers.
     this.#db.exec('BEGIN')
     try {
-      aggregateStmt.run(idx, rootPieceCid, pieceSizeBytes.toString(), new Date().toISOString())
+      aggregateStmt.run(idx, rootPieceCid, pieceSizeBytes.toString(), indexed ? 1 : 0, new Date().toISOString())
       members.forEach((subPieceCid, segmentIndex) => {
         memberStmt.run(idx, segmentIndex, subPieceCid)
       })
@@ -392,13 +399,25 @@ export class MigrationDB {
     }
   }
 
+  /**
+   * The indexed aggregate whose root is `rootPieceCid`, if any. Serves the
+   * `/piece/{pcidv2}` route's aggregate dispatch: an indexed root answers
+   * with the assembled stream, not a sub-piece redirect.
+   */
+  indexedAggregateByRoot(rootPieceCid: string): { idx: number } | null {
+    const row = this.#db
+      .prepare(`SELECT idx FROM aggregates WHERE root_piece_cid = ? AND indexed = 1`)
+      .get(rootPieceCid) as { idx: number | bigint } | undefined
+    return row == null ? null : { idx: Number(row.idx) }
+  }
+
   aggregates(): AggregateRow[] {
     // `member_count` is the *source-CID* count, expanding packed sub-pieces
     // through `sub_piece_members`. A 48-CID packed aggregate reports 48, not
     // 1, so operator-facing counters (report, status) reflect input shape.
     const rows = this.#db
       .prepare(
-        `SELECT a.idx, a.root_piece_cid, a.piece_size_bytes, a.status, a.pull_id,
+        `SELECT a.idx, a.root_piece_cid, a.piece_size_bytes, a.indexed, a.status, a.pull_id,
                 a.data_set_id, a.piece_id, a.tx_hash, a.committed_block, a.error,
                 a.submitted_at, a.parked_at, a.committed_at,
                 (
@@ -417,6 +436,7 @@ export class MigrationDB {
         idx: Number(row.idx),
         rootPieceCid: String(row.root_piece_cid),
         pieceSizeBytes: String(row.piece_size_bytes),
+        indexed: Number(row.indexed) === 1,
         status: row.status as AggregateStatus,
         pullId: str(row.pull_id),
         dataSetId: str(row.data_set_id),

packages/cli/src/index.ts

@@ -56,6 +56,9 @@ Usage:
   ipfs2foc commp  <cid> [--gateway URL]...
   ipfs2foc plan   --cids <file> [--db <file>] [--gateway URL]... [--piece-size 32GiB]
                      [--concurrency 8]
+                     [--indexed-aggregates]  (experimental: embed a data segment index so each
+                     aggregate is pulled as one stream and added as one self-describing piece;
+                     requires a provider whose indexing understands embedded indexes)
   ipfs2foc import-manifest <manifest.json> [--db <file>] [--network mainnet|calibration]
   ipfs2foc export [--db <file>] [--out <file>] [--network mainnet|calibration] [--source-relay <https-url>]
                      [--piece-size 32GiB] [--no-auto-pack]
@@ -262,6 +265,7 @@ async function cmdPlan(argv: string[]): Promise<void> {
       'ipfs-fallback-mode': { type: 'string' },
       'ipfs-fallback-timeout-seconds': { type: 'string' },
       'no-auto-pack': { type: 'boolean', default: false },
+      'indexed-aggregates': { type: 'boolean', default: false },
     },
   })
   if (values.cids == null) {
@@ -283,6 +287,7 @@ async function cmdPlan(argv: string[]): Promise<void> {
       ipfsFallback: fallback.ipfsFallback,
       fallbackTimeoutMs: fallback.fallbackTimeoutMs,
       autoPack: values['no-auto-pack'] !== true,
+      indexedAggregates: values['indexed-aggregates'] === true,
     })
 
     log('')