fix(arweave): reject pending transactions in the tx-data source

CHANGELOG.md

@@ -59,6 +59,14 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
 ### Fixed
 
+- The `tx-data` retrieval source no longer treats an unmined transaction as
+  data. A node answers `202 Pending` for a transaction it has not mined, and
+  that text was decoded as the transaction's data with a `NaN` size, which then
+  broke x402 pricing, let the request skip rate-limit token accounting, and
+  made a metrics call throw from inside the stream's `end` handler. The source
+  now accepts only `200` answers, requires a whole-number `data_size`, and
+  rejects data whose length differs from it, so the request moves on to the
+  next source.
 - The trusted-gateways root TX lookup no longer records a 1-byte payload size
   when a gateway rejects its HEAD request. The lookup then falls back to a
   `Range: bytes=0-0` GET, whose `Content-Length` is 1; that value was taken as

src/arweave/composite-client.test.ts

@@ -11,6 +11,7 @@ import { AddressInfo } from 'node:net';
 import { default as Arweave } from 'arweave';
 
 import { ArweaveCompositeClient } from './composite-client.js';
+import { toB64Url } from '../lib/encoding.js';
 import { UniformFailureSimulator } from '../lib/chaos.js';
 import { ArweavePeerManager } from '../peers/arweave-peer-manager.js';
 import * as config from '../config.js';
@@ -678,4 +679,126 @@ describe('ArweaveCompositeClient', () => {
       assert.equal(result.successCount, live.length);
     });
   });
+
+  describe('getData', () => {
+    const BASE_URL = 'https://test.example.com';
+
+    /** A client whose trusted node answers /data and /data_size as given. */
+    const clientAnswering = (
+      data: { status: number; data: unknown },
+      dataSize: { status: number; data: unknown },
+    ) => {
+      const client = createTestClient();
+      (client as any).trustedNodeRequestBucket = 10;
+      (client as any).trustedNodeAxios = mock.fn(
+        async (request: { url: string }) => ({
+          ...(request.url.endsWith('/data_size') ? dataSize : data),
+          config: { baseURL: BASE_URL },
+        }),
+      );
+      return client;
+    };
+
+    const readAll = async (stream: NodeJS.ReadableStream) => {
+      const chunks: Buffer[] = [];
+      for await (const chunk of stream) {
+        chunks.push(Buffer.from(chunk as Buffer));
+      }
+      return Buffer.concat(chunks);
+    };
+
+    const payload = Buffer.from('hello world');
+
+    it('serves a mined transaction and lets later end listeners run', async () => {
+      // axios parses the plain-number /data_size body as JSON.
+      const client = clientAnswering(
+        { status: 200, data: toB64Url(payload) },
+        { status: 200, data: payload.length },
+      );
+
+      const result = await client.getData({ id: 'mined-tx' });
+      let laterEndListenerRan = false;
+      result.stream.on('end', () => {
+        laterEndListenerRan = true;
+      });
+
+      assert.equal(result.size, payload.length);
+      assert.deepEqual(await readAll(result.stream), payload);
+      assert.equal(laterEndListenerRan, true);
+    });
+
+    it('serves the requested region of a mined transaction', async () => {
+      const client = clientAnswering(
+        { status: 200, data: toB64Url(payload) },
+        { status: 200, data: String(payload.length) },
+      );
+
+      const result = await client.getData({
+        id: 'mined-tx',
+        region: { offset: 6, size: 5 },
+      });
+
+      assert.equal(result.size, 5);
+      assert.equal((await readAll(result.stream)).toString(), 'world');
+    });
+
+    it('keeps later end listeners running when a region size is not finite', async () => {
+      // Recording NaN would throw from inside the 'end' listener, and a
+      // throwing listener stops the ones registered after it.
+      const client = clientAnswering(
+        { status: 200, data: toB64Url(payload) },
+        { status: 200, data: payload.length },
+      );
+
+      const result = await client.getData({
+        id: 'mined-tx',
+        region: { offset: 0, size: Number.NaN },
+      });
+      let laterEndListenerRan = false;
+      result.stream.on('end', () => {
+        laterEndListenerRan = true;
+      });
+      await readAll(result.stream);
+
+      assert.equal(laterEndListenerRan, true);
+    });
+
+    it('rejects a transaction the node reports as pending', async () => {
+      // An unmined transaction: the node answers 202 with the text "Pending"
+      // on both routes. Serving it would yield junk bytes and a NaN size.
+      const client = clientAnswering(
+        { status: 202, data: 'Pending' },
+        { status: 202, data: 'Pending' },
+      );
+
+      await assert.rejects(
+        client.getData({ id: 'pending-tx' }),
+        /Transaction data unavailable \(data 202, data_size 202\)/,
+      );
+    });
+
+    it('rejects a size that is not a non-negative integer', async () => {
+      const client = clientAnswering(
+        { status: 200, data: toB64Url(payload) },
+        { status: 200, data: 'not-a-number' },
+      );
+
+      await assert.rejects(
+        client.getData({ id: 'mined-tx' }),
+        /Invalid transaction data size: not-a-number/,
+      );
+    });
+
+    it('rejects data whose length differs from data_size', async () => {
+      const client = clientAnswering(
+        { status: 200, data: toB64Url(payload) },
+        { status: 200, data: payload.length + 1 },
+      );
+
+      await assert.rejects(
+        client.getData({ id: 'mined-tx' }),
+        /Transaction data is 11 bytes but data_size is 12/,
+      );
+    });
+  });
 });

src/arweave/composite-client.ts

@@ -1756,6 +1756,22 @@ export class ArweaveCompositeClient
     }
   }
 
+  /**
+   * Fetches a transaction's data from the trusted node via `/tx/{id}/data`
+   * and `/tx/{id}/data_size`.
+   *
+   * Both requests must answer `200`. Any other status, including the `202
+   * Pending` a node returns for an unmined transaction, throws so the caller
+   * can fall through to the next data source. The size must be a non-negative
+   * safe integer, and the decoded data must be exactly that many bytes.
+   *
+   * @param id - Transaction ID.
+   * @param region - Optional byte range. When given, the stream carries only
+   * that slice of the data and the reported size is `region.size`.
+   * @param signal - Aborts the request when triggered.
+   * @returns The data as an unverified, trusted, uncached stream.
+   * @throws When the node has no usable data or the size check fails.
+   */
   async getData({
     id,
     region,
@@ -1782,12 +1798,33 @@ export class ArweaveCompositeClient
         }),
       ]);
 
+      // A node answers 202 "Pending" for a transaction it knows about but has
+      // not mined. That body is a status string, not data, so anything but a
+      // 200 is not a usable answer: decoding "Pending" would yield a few junk
+      // bytes and a NaN size.
+      if (dataResponse.status !== 200 || dataSizeResponse.status !== 200) {
+        throw new Error(
+          `Transaction data unavailable (data ${dataResponse.status}, data_size ${dataSizeResponse.status})`,
+        );
+      }
+
       if (!dataResponse.data) {
         throw Error('No transaction data');
       }
 
-      const size = +dataSizeResponse.data;
+      const size = Number(dataSizeResponse.data);
+      if (!Number.isSafeInteger(size) || size < 0) {
+        throw new Error(
+          `Invalid transaction data size: ${String(dataSizeResponse.data).slice(0, 32)}`,
+        );
+      }
+
       let txData = fromB64Url(dataResponse.data);
+      if (txData.length !== size) {
+        throw new Error(
+          `Transaction data is ${txData.length} bytes but data_size is ${size}`,
+        );
+      }
 
       if (region) {
         txData = txData.subarray(region.offset, region.offset + region.size);
@@ -1812,8 +1849,13 @@ export class ArweaveCompositeClient
           request_type: requestType,
         });
 
-        // Track bytes streamed
+        // Track bytes streamed. A non-finite value would poison the counter
+        // and make the histogram throw from inside this 'end' listener, which
+        // also skips any 'end' listeners registered after it.
         const bytesStreamed = region ? region.size : size;
+        if (!Number.isFinite(bytesStreamed)) {
+          return;
+        }
         metrics.getDataStreamBytesTotal.inc(
           {
             class: this.constructor.name,