fix(http): stop buffering whole bodies on the range-GET header fallback

src/data/chunk-metadata-anchor-source.test.ts

@@ -5,6 +5,8 @@
  * SPDX-License-Identifier: AGPL-3.0-or-later
  */
 import { strict as assert } from 'node:assert';
+import * as http from 'node:http';
+import { AddressInfo } from 'node:net';
 import { afterEach, describe, it } from 'node:test';
 import type { AxiosInstance } from 'axios';
 
@@ -504,4 +506,110 @@ describe('ChunkMetadataAnchorSource', () => {
       assert.notStrictEqual(result, null);
     });
   });
+
+  describe('range-GET fallback against a real server', () => {
+    /**
+     * A peer whose HEAD carries no chunk headers, so every probe takes the
+     * range-GET fallback. Records each request's client port so tests can
+     * tell whether connections were reused.
+     */
+    const startPeer = async (onGet: (res: http.ServerResponse) => void) => {
+      const clientPorts: number[] = [];
+      const server = http.createServer((req, res) => {
+        clientPorts.push(req.socket.remotePort ?? -1);
+        if (req.method === 'HEAD') {
+          res.writeHead(200, { 'Content-Length': '0' });
+          res.end();
+          return;
+        }
+        onGet(res);
+      });
+      await new Promise<void>((resolve) =>
+        server.listen(0, '127.0.0.1', resolve),
+      );
+      const { port } = server.address() as AddressInfo;
+      return {
+        url: `http://127.0.0.1:${port}`,
+        clientPorts,
+        close: async () => {
+          server.closeAllConnections();
+          await new Promise((resolve) => server.close(resolve));
+        },
+      };
+    };
+
+    /** A source using its own real axios instance. */
+    const makeRealSource = (peerUrl: string) =>
+      new ChunkMetadataAnchorSource({
+        log,
+        peerUrls: [peerUrl],
+        requestTimeoutMs: 5000,
+        cacheSize: 32,
+        cacheTtlMs: 60_000,
+        fetchTxOffset: async () => matchingChainOffset,
+        fetchTransaction: async () => ({ data_root: dataRoot }),
+      });
+
+    it('does not download the whole body when the peer ignores the range', async () => {
+      const total = 64 * 1024 * 1024;
+      const progress = { written: 0, finished: false };
+      const peer = await startPeer((res) => {
+        res.writeHead(200, {
+          'Content-Length': String(total),
+          ...(chunkHeaders() as Record<string, string>),
+        });
+        const chunk = Buffer.alloc(64 * 1024);
+        const write = () => {
+          while (progress.written < total) {
+            progress.written += chunk.length;
+            if (!res.write(chunk)) {
+              res.once('drain', write);
+              return;
+            }
+          }
+          res.end();
+          progress.finished = true;
+        };
+        write();
+      });
+      try {
+        const result = await makeRealSource(peer.url).getTxBoundary(
+          inRangeOffset,
+        );
+
+        assert.notEqual(result, null);
+        assert.strictEqual(result!.id, txId);
+        await new Promise((resolve) => setTimeout(resolve, 100));
+        assert.equal(progress.finished, false);
+        assert.ok(
+          progress.written < total / 2,
+          `peer sent ${progress.written} of ${total} bytes`,
+        );
+      } finally {
+        await peer.close();
+      }
+    });
+
+    it('keeps reusing the connection when the peer honours the range', async () => {
+      const peer = await startPeer((res) => {
+        res.writeHead(206, {
+          'Content-Length': '1',
+          'Content-Range': 'bytes 0-0/262144',
+          ...(chunkHeaders() as Record<string, string>),
+        });
+        res.end(Buffer.from('x'));
+      });
+      try {
+        const source = makeRealSource(peer.url);
+        assert.notEqual(await source.getTxBoundary(inRangeOffset), null);
+        assert.notEqual(await source.getTxBoundary(inRangeOffset + 1n), null);
+
+        // HEAD and GET for each probe, all over one keep-alive connection.
+        assert.equal(peer.clientPorts.length, 4);
+        assert.equal(new Set(peer.clientPorts).size, 1);
+      } finally {
+        await peer.close();
+      }
+    });
+  });
 });

src/data/chunk-metadata-anchor-source.ts

@@ -17,7 +17,7 @@ import {
   anchorChunkMetadata,
 } from '../lib/chunk-metadata-anchor.js';
 import { createAgentPair } from '../lib/http-agent.js';
-import { normalizeAbortError } from '../lib/http-utils.js';
+import { discardResponseBody, normalizeAbortError } from '../lib/http-utils.js';
 
 // Largest absolute weave offset this source will probe via the
 // number-typed cache + chain cross-check path. The chain-anchored
@@ -301,13 +301,21 @@ export class ChunkMetadataAnchorSource implements TxBoundarySource {
     }
 
     // `bytes=0-0` is the smallest legal range; the server returns a 1-
-    // byte body which we discard. Headers are the only thing we want.
+    // byte body which we discard. Headers are the only thing we want. The
+    // body arrives as a stream so a peer that ignores the range has its
+    // connection closed instead of its whole response buffered.
     const getResponse = await this.axiosInstance.get(url, {
       signal,
       timeout: this.requestTimeoutMs,
       headers: { Range: 'bytes=0-0' },
-      responseType: 'arraybuffer',
+      responseType: 'stream',
     });
+    await discardResponseBody(getResponse.data, {
+      timeoutMs: this.requestTimeoutMs,
+    });
+    // A buffered read rejected when aborted mid-body; keep that behaviour now
+    // that the body is discarded separately.
+    signal?.throwIfAborted();
     return getResponse.headers as Record<string, string | string[] | undefined>;
   }
 

src/discovery/gateways-root-tx-index.test.ts

@@ -5,6 +5,8 @@
  * SPDX-License-Identifier: AGPL-3.0-or-later
  */
 import { strict as assert } from 'node:assert';
+import * as http from 'node:http';
+import { AddressInfo } from 'node:net';
 import { afterEach, describe, it, mock } from 'node:test';
 import { LRUCache } from 'lru-cache';
 import { GatewaysRootTxIndex } from './gateways-root-tx-index.js';
@@ -824,4 +826,154 @@ describe('GatewaysRootTxIndex', () => {
       );
     });
   });
+
+  describe('range-GET fallback against a real server', () => {
+    const ROOT_HEADERS = {
+      'X-AR-IO-Root-Transaction-Id': 'root-tx-456',
+      'X-AR-IO-Root-Data-Item-Offset': '1000',
+      'X-AR-IO-Root-Data-Offset': '1500',
+    };
+
+    /**
+     * A /raw server that rejects HEAD, as some peers behind CDNs do, so every
+     * lookup takes the range-GET fallback. Records each request's client port
+     * so tests can tell whether connections were reused.
+     */
+    const startServer = async (onGet: (res: http.ServerResponse) => void) => {
+      const clientPorts: number[] = [];
+      const server = http.createServer((req, res) => {
+        clientPorts.push(req.socket.remotePort ?? -1);
+        if (req.method === 'HEAD') {
+          res.writeHead(405, { 'Content-Length': '0' });
+          res.end();
+          return;
+        }
+        onGet(res);
+      });
+      await new Promise<void>((resolve) =>
+        server.listen(0, '127.0.0.1', resolve),
+      );
+      const { port } = server.address() as AddressInfo;
+      return {
+        url: `http://127.0.0.1:${port}`,
+        clientPorts,
+        close: async () => {
+          server.closeAllConnections();
+          await new Promise((resolve) => server.close(resolve));
+        },
+      };
+    };
+
+    /**
+     * Streams `total` bytes with backpressure and reports how much the client
+     * actually accepted before hanging up.
+     */
+    const streamLargeBody = (
+      res: http.ServerResponse,
+      status: number,
+      total: number,
+      headers: Record<string, string>,
+    ) => {
+      const progress = { written: 0, finished: false };
+      res.writeHead(status, { 'Content-Length': String(total), ...headers });
+      const chunk = Buffer.alloc(64 * 1024);
+      const write = () => {
+        while (progress.written < total) {
+          progress.written += chunk.length;
+          if (!res.write(chunk)) {
+            res.once('drain', write);
+            return;
+          }
+        }
+        res.end();
+        progress.finished = true;
+      };
+      write();
+      return progress;
+    };
+
+    const makeIndex = (url: string) => {
+      const index = new GatewaysRootTxIndex({
+        log,
+        trustedGatewaysUrls: { [url]: 1 },
+        requestTimeoutMs: 5000,
+        rateLimitBurstSize: 1000,
+        rateLimitTokensPerInterval: 1000,
+        rateLimitInterval: 'second',
+      });
+      for (const [, limiter] of (index as any)['limiters']) {
+        limiter.content = limiter.bucketSize;
+      }
+      return index;
+    };
+
+    const TOTAL = 64 * 1024 * 1024;
+
+    it('does not download the whole item when the peer ignores the range', async () => {
+      let progress = { written: 0, finished: false };
+      const server = await startServer((res) => {
+        progress = streamLargeBody(res, 200, TOTAL, ROOT_HEADERS);
+      });
+      try {
+        const result = await makeIndex(server.url).getRootTx('item-a');
+
+        assert(result !== undefined);
+        assert.equal(result.rootTxId, 'root-tx-456');
+        // A 200 carries the whole payload, so Content-Length is its size.
+        assert.equal(result.dataSize, TOTAL);
+        await new Promise((resolve) => setTimeout(resolve, 100));
+        assert.equal(progress.finished, false);
+        assert.ok(
+          progress.written < TOTAL / 2,
+          `peer sent ${progress.written} of ${TOTAL} bytes`,
+        );
+      } finally {
+        await server.close();
+      }
+    });
+
+    it('does not download a large error body either', async () => {
+      let progress = { written: 0, finished: false };
+      const server = await startServer((res) => {
+        progress = streamLargeBody(res, 500, TOTAL, {});
+      });
+      try {
+        const result = await makeIndex(server.url).getRootTx('item-a');
+
+        assert.equal(result, undefined);
+        await new Promise((resolve) => setTimeout(resolve, 100));
+        assert.equal(progress.finished, false);
+        assert.ok(
+          progress.written < TOTAL / 2,
+          `peer sent ${progress.written} of ${TOTAL} bytes`,
+        );
+      } finally {
+        await server.close();
+      }
+    });
+
+    it('keeps reusing the connection when the peer honours the range', async () => {
+      const server = await startServer((res) => {
+        res.writeHead(206, {
+          'Content-Length': '1',
+          'Content-Range': 'bytes 0-0/5000',
+          ...ROOT_HEADERS,
+        });
+        res.end(Buffer.from('x'));
+      });
+      try {
+        const index = makeIndex(server.url);
+        const first = await index.getRootTx('item-a');
+        const second = await index.getRootTx('item-b');
+
+        assert.equal(first?.dataSize, 5000);
+        assert.equal(second?.dataSize, 5000);
+        // HEAD and GET for each lookup, all over one keep-alive connection.
+        assert.equal(server.clientPorts.length, 4);
+        assert.equal(new Set(server.clientPorts).size, 1);
+      } finally {
+        await server.close();
+      }
+    });
+  });
 });

src/discovery/gateways-root-tx-index.ts

@@ -10,7 +10,11 @@ import { LRUCache } from 'lru-cache';
 import { TokenBucket } from 'limiter';
 import { DataItemRootIndex } from '../types.js';
 import { shuffleArray } from '../lib/random.js';
-import { parseContentRange, parseNonNegativeInt } from '../lib/http-utils.js';
+import {
+  discardResponseBody,
+  parseContentRange,
+  parseNonNegativeInt,
+} from '../lib/http-utils.js';
 import { createAgentPair } from '../lib/http-agent.js';
 import * as config from '../config.js';
 import * as metrics from '../metrics.js';
@@ -46,6 +50,7 @@ export class GatewaysRootTxIndex implements DataItemRootIndex {
   private readonly axiosInstance: AxiosInstance;
   private readonly cache?: LRUCache<string, CachedGatewayOffsets>;
   private readonly limiters: Map<string, TokenBucket>;
+  private readonly requestTimeoutMs: number;
 
   constructor({
     log,
@@ -66,6 +71,7 @@ export class GatewaysRootTxIndex implements DataItemRootIndex {
   }) {
     this.log = log.child({ class: this.constructor.name });
     this.cache = cache;
+    this.requestTimeoutMs = requestTimeoutMs;
 
     if (Object.keys(trustedGatewaysUrls).length === 0) {
       throw new Error('At least one gateway URL must be provided');
@@ -296,7 +302,9 @@ export class GatewaysRootTxIndex implements DataItemRootIndex {
    * those behind CDNs or proxies — don't support HEAD on this route even
    * though the upstream gateway would. `bytes=0-0` is the smallest legal
    * range; the server returns 1 byte of body we discard, and the headers
-   * are what we want.
+   * are what we want. The body is received as a stream and discarded by
+   * {@link discardResponseBody}: a peer that ignores the range and sends the
+   * whole item has its connection closed instead of the item being buffered.
    *
    * 404 is treated as a definitive "item doesn't exist on this peer" and
    * propagated to the caller — falling back to GET would just hit the
@@ -316,10 +324,22 @@ export class GatewaysRootTxIndex implements DataItemRootIndex {
         throw err;
       }
       // Network error, 405 Method Not Allowed, 5xx, etc. — try GET.
-      return this.axiosInstance.get(url, {
-        headers: { Range: 'bytes=0-0' },
-        responseType: 'arraybuffer',
-      });
+      try {
+        const response = await this.axiosInstance.get(url, {
+          headers: { Range: 'bytes=0-0' },
+          responseType: 'stream',
+        });
+        await discardResponseBody(response.data, {
+          timeoutMs: this.requestTimeoutMs,
+        });
+        return response;
+      } catch (getErr: any) {
+        // An error status still carries an unread body stream.
+        await discardResponseBody(getErr?.response?.data, {
+          timeoutMs: this.requestTimeoutMs,
+        });
+        throw getErr;
+      }
     }
   }
 }