foxglove · snosenzo · Apr 21, 2026 · Apr 21, 2026 · Apr 21, 2026 · claude
@@ -6,6 +6,8 @@ import type { IReadable } from "@mcap/core";
 export class BlobReadable implements IReadable {
   #blob: Blob;
 
+  public readonly supportsConcurrentReads = true;
+
   public constructor(blob: Blob) {
     this.#blob = blob;
   }

@@ -10,6 +10,7 @@ type ChunkCursorParams = {
   startTime: bigint | undefined;
   endTime: bigint | undefined;
   reverse: boolean;
+  messageIndexCache?: ReadonlyMap<bigint, Uint8Array>;
 };
 
 /**
@@ -26,6 +27,7 @@ export class ChunkCursor {
   #startTime: bigint | undefined;
   #endTime: bigint | undefined;
   #reverse: boolean;
+  #messageIndexCache?: ReadonlyMap<bigint, Uint8Array>;
 
   // List of message offsets (across all channels) sorted by logTime.
   #orderedMessageOffsets?: [logTime: bigint, offset: bigint][];
@@ -38,6 +40,7 @@ export class ChunkCursor {
     this.#startTime = params.startTime;
     this.#endTime = params.endTime;
     this.#reverse = params.reverse;
+    this.#messageIndexCache = params.messageIndexCache;
 
     if (this.chunkIndex.messageIndexLength === 0n) {
       // Chunk has no message indexes.
@@ -134,10 +137,16 @@ export class ChunkCursor {
 
     // Future optimization: read only message indexes for given channelIds, not all message indexes for the chunk
     const messageIndexEndOffset = messageIndexStartOffset + this.chunkIndex.messageIndexLength;
-    const messageIndexes = await readable.read(
-      relevantMessageIndexStartOffset,
-      messageIndexEndOffset - relevantMessageIndexStartOffset,
-    );
+    const cachedMessageIndexes = this.#messageIndexCache?.get(this.chunkIndex.chunkStartOffset);
+    const messageIndexes = cachedMessageIndexes
+      ? cachedMessageIndexes.subarray(
+          Number(relevantMessageIndexStartOffset - messageIndexStartOffset),
+          Number(messageIndexEndOffset - messageIndexStartOffset),
+        )
+      : await readable.read(
+          relevantMessageIndexStartOffset,
+          messageIndexEndOffset - relevantMessageIndexStartOffset,
+        );
     const messageIndexesView = new DataView(
       messageIndexes.buffer,
       messageIndexes.byteOffset,

@@ -1180,4 +1180,118 @@ describe("McapIndexedReader", () => {
     const reader = await McapIndexedReader.Initialize({ readable: makeReadable(builder.buffer) });
     await expect(collect(reader.readMessages())).resolves.toEqual([message1, message2]);
   });
+
+  describe("prefetchMessageIndexes", () => {
+    // Tracking readable that allocates a fresh Uint8Array per call so it is safe to call
+    // concurrently. The module-level `makeReadable` above reuses a single buffer across reads and
+    // cannot stand in for a concurrency-safe readable.
+    function makeTrackingReadable(
+      data: Uint8Array,
+      opts: { supportsConcurrentReads: boolean },
+    ) {
+      let readCalls = 0;
+      return {
+        get readCalls() {
+          return readCalls;
+        },
+        supportsConcurrentReads: opts.supportsConcurrentReads,
+        size: async () => BigInt(data.length),
+        read: async (offset: bigint, size: bigint) => {
+          ++readCalls;
+          const out = new Uint8Array(Number(size));
+          out.set(new Uint8Array(data.buffer, data.byteOffset + Number(offset), Number(size)));
+          return out;
+        },
+      };
+    }
+
+    // Build a minimal two-chunk MCAP with a single channel and one message per chunk. This is
+    // shared by all prefetch tests and keeps the read-count assertions easy to reason about.
+    function buildTwoChunkFile() {
+      const channel: TypedMcapRecord = {
+        type: "Channel",
+        id: 1,
+        schemaId: 0,
+        topic: "a",
+        messageEncoding: "utf12",
+        metadata: new Map(),
+      };
+      const message1: TypedMcapRecords["Message"] = {
+        type: "Message",
+        channelId: channel.id,
+        sequence: 1,
+        logTime: 1n,
+        publishTime: 0n,
+        data: new Uint8Array(),
+      };
+      const message2: TypedMcapRecords["Message"] = {
+        type: "Message",
+        channelId: channel.id,
+        sequence: 2,
+        logTime: 2n,
+        publishTime: 0n,
+        data: new Uint8Array(),
+      };
+
+      const chunk1 = new ChunkBuilder({ useMessageIndex: true });
+      chunk1.addChannel(channel);
+      chunk1.addMessage(message1);
+
+      const chunk2 = new ChunkBuilder({ useMessageIndex: true });
+      chunk2.addChannel(channel);
+      chunk2.addMessage(message2);
+
+      const builder = new McapRecordBuilder();
+      builder.writeMagic();
+      builder.writeHeader({ profile: "", library: "" });
+
+      const chunkIndexes: TypedMcapRecords["ChunkIndex"][] = [];
+      chunkIndexes.push(writeChunkWithMessageIndexes(builder, chunk1));
+      chunkIndexes.push(writeChunkWithMessageIndexes(builder, chunk2));
+
+      builder.writeDataEnd({ dataSectionCrc: 0 });
+      const summaryStart = BigInt(builder.length);
+      builder.writeChannel(channel);
+      for (const index of chunkIndexes) {
+        builder.writeChunkIndex(index);
+      }
+      builder.writeFooter({ summaryStart, summaryOffsetStart: 0n, summaryCrc: 0 });
+      builder.writeMagic();
+
+      return { buffer: builder.buffer, messages: [message1, message2] };
+    }
+
+    it("yields the same messages as the default path", async () => {
+      const { buffer, messages } = buildTwoChunkFile();
+      const reader = await McapIndexedReader.Initialize({
+        readable: makeReadable(buffer),
+        prefetchMessageIndexes: true,
+      });
+      await expect(collect(reader.readMessages())).resolves.toEqual(messages);
+    });
+
+    it.each([
+      { label: "sequential readable", concurrent: false },
+      { label: "concurrent readable", concurrent: true },
+    ])(
+      "loads all message indexes during Initialize and issues no message-index reads during readMessages ($label)",
+      async ({ concurrent }) => {
+        const { buffer, messages } = buildTwoChunkFile();
+        const readable = makeTrackingReadable(buffer, { supportsConcurrentReads: concurrent });
+
+        const reader = await McapIndexedReader.Initialize({
+          readable,
+          prefetchMessageIndexes: true,
+        });
+        const readsAfterInit = readable.readCalls;
+
+        await expect(collect(reader.readMessages())).resolves.toEqual(messages);
+
+        // With prefetch enabled, readMessages must only issue one chunk-data read per chunk (2)
+        // — no message-index reads should happen after Initialize. Without prefetch the same
+        // iteration would additionally issue one message-index read per chunk (4 reads total).
+        expect(readable.readCalls - readsAfterInit).toEqual(2);
+      },
+    );
+  });
 });
@@ -21,6 +21,7 @@
   footer: TypedMcapRecords["Footer"];
   dataEndOffset: bigint;
   dataSectionCrc?: number;
+  messageIndexCache?: ReadonlyMap<bigint, Uint8Array>;
 };
 
 export class McapIndexedReader {
@@ -39,6 +40,7 @@
 
   #readable: IReadable;
   #decompressHandlers?: DecompressHandlers;
+  #messageIndexCache?: ReadonlyMap<bigint, Uint8Array>;
 
   #messageStartTime: bigint | undefined;
   #messageEndTime: bigint | undefined;
@@ -59,6 +61,7 @@
     this.footer = args.footer;
     this.dataEndOffset = args.dataEndOffset;
     this.dataSectionCrc = args.dataSectionCrc;
+    this.#messageIndexCache = args.messageIndexCache;
 
     for (const chunk of args.chunkIndexes) {
       if (this.#messageStartTime == undefined || chunk.messageStartTime < this.#messageStartTime) {
@@ -89,6 +92,7 @@
   static async Initialize({
     readable,
     decompressHandlers,
+    prefetchMessageIndexes = false,
   }: {
     readable: IReadable;
 
@@ -97,6 +101,39 @@
      * compression will be called to decompress the chunk data.
      */
     decompressHandlers?: DecompressHandlers;
+
+    /**
+     * When `true`, every chunk's MessageIndex records are read and cached in memory during
+     * `Initialize`. Subsequent calls to `readMessages` reuse the cache instead of issuing one
+     * network/disk read per chunk to load message indexes.
+     *
+     * Use this when:
+     * - Message reads are latency-bound (e.g. a remote `IReadable` over HTTP, S3, or similar) and
+     *   you want to parallelize the many small message-index reads up front instead of serializing
+     *   them inside `readMessages`.
+     * - You will call `readMessages` multiple times on the same reader (e.g. seeking, topic
+     *   filtering, playback UIs) and want consistently low per-call latency.
+     *
+     * Avoid this when:
+     * - The file is very large and memory is constrained. The cache holds the entirety of every
+     *   chunk's MessageIndex bytes until the reader is discarded.
+     * - You only need to read a small, known slice of the file once. The prefetch will read
+     *   message indexes for chunks you never iterate, which can be wasteful.
+     * - The underlying `IReadable` is already low-latency (e.g. a local mmapped file) where the
+     *   per-chunk await cost is negligible.
+     *
+     * Memory cost: roughly 16 bytes per message in the file, held for the reader's lifetime
+     * (each MessageIndex entry is a `(logTime, offset)` pair of `uint64`s).
+     *
+     * Parallelism: if the provided `IReadable` advertises `supportsConcurrentReads`, the prefetch
+     * issues reads with a small bounded concurrency. Otherwise the reads are serialized so that
+     * implementations reusing an internal buffer (e.g. `FileHandleReadable`) remain correct; in
+     * that case the main benefit is that the cost is amortized once at init time and reused
+     * across every `readMessages` call.
+     *
+     * Defaults to `false`.
+     */
+    prefetchMessageIndexes?: boolean;
   }): Promise<McapIndexedReader> {
     const size = await readable.size();
 
@@ -318,6 +355,73 @@
       throw errorWithLibrary(`${indexReader.bytesRemaining()} bytes remaining in index section`);
     }
 
+    let messageIndexCache: Map<bigint, Uint8Array> | undefined;
+    if (prefetchMessageIndexes && chunkIndexes.length > 0) {
+      messageIndexCache = new Map<bigint, Uint8Array>();
+
+      // Resolve each chunk's message-index byte range up front so we can either fan out reads in
+      // parallel or iterate them sequentially based on the readable's advertised capability.
+      const indexRequests: { chunkStartOffset: bigint; offset: bigint; length: bigint }[] = [];
+      for (const chunkIndex of chunkIndexes) {
+        if (chunkIndex.messageIndexLength === 0n) {
+          messageIndexCache.set(chunkIndex.chunkStartOffset, new Uint8Array());
+          continue;
+        }
+        let messageIndexStartOffset: bigint | undefined;
+        for (const offset of chunkIndex.messageIndexOffsets.values()) {
+          if (messageIndexStartOffset == undefined || offset < messageIndexStartOffset) {
+            messageIndexStartOffset = offset;
+          }
+        }
+        if (messageIndexStartOffset == undefined) {
+          messageIndexCache.set(chunkIndex.chunkStartOffset, new Uint8Array());
+          continue;
+        }
+        indexRequests.push({
+          chunkStartOffset: chunkIndex.chunkStartOffset,
+          offset: messageIndexStartOffset,
+          length: chunkIndex.messageIndexLength,
+        });
+      }
+
+      if (readable.supportsConcurrentReads) {
+        // Fan out reads with a bounded worker pool. Concurrency-safe readables guarantee the
+        // returned Uint8Arrays won't be mutated or aliased by any subsequent read. We cap the
+        // in-flight count so that files with many chunks don't overwhelm the underlying transport
+        // (e.g. HTTP connection limits) when reading from a remote IReadable.
+        const MAX_CONCURRENT_READS = 6;
+        const indexResults = new Array<Uint8Array>(indexRequests.length);
+        let nextIndex = 0;
+        const workers = Array.from(
+          { length: Math.min(MAX_CONCURRENT_READS, indexRequests.length) }, // iterable
+          async () => {
+            // map function
+            for (;;) {
+              // Safe: JS is single-threaded; each worker awaits between iterations,
+              // so nextIndex++ never races.
+              const i = nextIndex++;
+              if (i >= indexRequests.length) {
+                return;
+              }
+              const { offset, length } = indexRequests[i]!;
+              indexResults[i] = await readable.read(offset, length);
+            }
+          },
+        );
+        await Promise.all(workers);
+        for (let i = 0; i < indexRequests.length; i++) {
+          messageIndexCache.set(indexRequests[i]!.chunkStartOffset, indexResults[i]!);
+        }
+      } else {
+        // Readable may reuse an internal buffer across read() calls; serialize reads and copy each
+        // result into a fresh Uint8Array before the next read starts.
+        for (const { chunkStartOffset, offset, length } of indexRequests) {
+          const bytes = await readable.read(offset, length);
+          messageIndexCache.set(chunkStartOffset, new Uint8Array(bytes));
+        }
+      }
+    }
+
     return new McapIndexedReader({
       readable,
       chunkIndexes,
@@ -332,6 +436,7 @@
       footer,
       dataEndOffset,
       dataSectionCrc,
+      messageIndexCache,
     });
   }
 
@@ -372,7 +477,14 @@
     for (const chunkIndex of this.chunkIndexes) {
       if (chunkIndex.messageStartTime <= endTime && chunkIndex.messageEndTime >= startTime) {
         chunkCursors.push(
-          new ChunkCursor({ chunkIndex, relevantChannels, startTime, endTime, reverse }),
+          new ChunkCursor({
+            chunkIndex,
+            relevantChannels,
+            startTime,
+            endTime,
+            reverse,
+            messageIndexCache: this.#messageIndexCache,
+          }),
         );
         if (chunksOrdered && prevChunkEndTime != undefined) {
           chunksOrdered = chunkIndex.messageStartTime >= prevChunkEndTime;

@@ -9,6 +9,11 @@ export class TempBuffer implements IReadable, IWritable, ISeekableWriter {
   #buffer = new ArrayBuffer(0);
   #position = 0;
 
+  /** Concurrent read() calls are safe: reads only return views and never mutate state. Note that
+   * this guarantee only holds when no write() / seek() / truncate() is interleaved with the reads.
+   */
+  readonly supportsConcurrentReads = true;
+
   constructor(source?: ArrayBufferView | ArrayBuffer) {
     if (source instanceof ArrayBuffer) {
       this.#buffer = source;

@@ -139,4 +139,23 @@ export type DecompressHandlers = {
 export interface IReadable {
   size(): Promise<bigint>;
   read(offset: bigint, size: bigint): Promise<Uint8Array>;
+
+  /**
+   * Optional capability flag advertised by implementations whose `read()` is safe to invoke
+   * concurrently (e.g. via `Promise.all`).
+   *
+   * When `true`, consumers may:
+   * - Issue multiple `read()` calls without awaiting each one sequentially.
+   * - Retain the returned `Uint8Array` across subsequent `read()` calls; the bytes will not be
+   *   mutated or aliased by any later read.
+   *
+   * When omitted or `false`, consumers MUST:
+   * - Serialize `read()` calls (await each one before issuing the next).
+   * - Treat the returned `Uint8Array` as valid only until the next `read()` call — copy the bytes
+   *   out before the next read if they need to outlive it.
+   *
+   * Defaults to `false` (i.e. omitted) for safety; existing implementations that reuse an internal
+   * buffer do not need to opt in.
+   */
+  readonly supportsConcurrentReads?: boolean;
 }