feat(chunks): record why a chunk POST failed and whether peers will keep it

src/arweave/composite-client.test.ts

@@ -15,6 +15,7 @@ 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';
+import * as metrics from '../metrics.js';
 import log from '../log.js';
 
 describe('ArweaveCompositeClient', () => {
@@ -529,6 +530,88 @@ describe('ArweaveCompositeClient', () => {
       assert.equal(result.success, false);
       assert.equal(result.statusCode, 500);
     });
+
+    // The failure counter is what an operator reads when a peer starts
+    // rejecting chunks. Without a reason, a peer refusing the chunk (400), one
+    // rate-limiting us (429) and one we cannot reach are indistinguishable,
+    // and each calls for a different response.
+    const failReasonCount = async (endpoint: string, reason: string) => {
+      const { values } = await metrics.arweaveChunkPostCounter.get();
+      const sample = values.find(
+        (v: any) =>
+          v.labels.endpoint === endpoint &&
+          v.labels.status === 'fail' &&
+          v.labels.reason === reason,
+      );
+      return sample?.value ?? 0;
+    };
+
+    it('labels a peer-rejected post with the peer’s status code', async () => {
+      respond = (res) => res.writeHead(429).end();
+      const client = createTestClient();
+      const before = await failReasonCount(baseUrl, '429');
+      await post(client);
+      assert.equal(await failReasonCount(baseUrl, '429'), before + 1);
+    });
+
+    // The abort deadline is normally the LOWER of the two, so this is the common
+    // timeout path. AbortSignal.timeout() surfaces as ERR_CANCELED, which would
+    // otherwise be reported as a caller cancellation — and aggregateStatusCode()
+    // turns that into 499 (Client Closed Request), blaming the uploader for our
+    // own deadline.
+    it('reports our own abort deadline as a timeout, not a cancellation', async () => {
+      respond = () => undefined; // never answer
+      const client: any = createTestClient();
+      const before = await failReasonCount(baseUrl, 'timeout');
+      const result = await client.postChunkToPeer({
+        peer: baseUrl,
+        chunk: {} as any,
+        abortTimeout: 50, // fires first
+        responseTimeout: 5000,
+        headers: {},
+      });
+      assert.equal(result.success, false);
+      assert.equal(result.timedOut, true);
+      assert.equal(result.canceled, false);
+      assert.equal(await failReasonCount(baseUrl, 'timeout'), before + 1);
+    });
+
+    it('labels accepted posts with the status the peer returned', async () => {
+      const successReasonCount = async (endpoint: string, reason: string) => {
+        const { values } = await metrics.arweaveChunkPostCounter.get();
+        const s = values.find(
+          (v: any) =>
+            v.labels.endpoint === endpoint &&
+            v.labels.status === 'success' &&
+            v.labels.reason === reason,
+        );
+        return s?.value ?? 0;
+      };
+      respond = (res) => res.writeHead(303).end();
+      const client = createTestClient();
+      const before = await successReasonCount(baseUrl, '303');
+      await post(client);
+      assert.equal(await successReasonCount(baseUrl, '303'), before + 1);
+    });
+
+    it('labels a post the peer never answers as a timeout, not a status code', async () => {
+      // Never respond: the request must hit responseTimeout rather than any
+      // HTTP status, so the reason has to come from the error, not a response.
+      respond = () => undefined;
+      const client: any = createTestClient();
+      const before = await failReasonCount(baseUrl, 'timeout');
+      const result = await client.postChunkToPeer({
+        peer: baseUrl,
+        chunk: {} as any,
+        abortTimeout: 5000,
+        responseTimeout: 50,
+        headers: {},
+      });
+      assert.equal(result.success, false);
+      assert.equal(result.timedOut, true);
+      assert.equal(result.statusCode, undefined);
+      assert.equal(await failReasonCount(baseUrl, 'timeout'), before + 1);
+    });
   });
 
   // Verifies the CHUNK_POST_CONTINUE_PAST_THRESHOLD behavior against real
@@ -626,6 +709,46 @@ describe('ArweaveCompositeClient', () => {
       assert.equal(result.successCount, urls.length);
     });
 
+    // successCount alone cannot tell "peers that will keep this chunk" from
+    // "peers that parked it in a disk pool they will drain", because a 303 is
+    // counted as a success (correctly — it is still propagation). The split is
+    // reported so callers can see which they got.
+    it('splits successes into long-term (200) and temporary (303)', async () => {
+      await startServers(2, 200);
+      const longTerm = [...urls];
+      // startServers resets `servers`/`urls`, so hold on to the first batch and
+      // restore both lists afterwards — otherwise afterEach never closes those
+      // listeners and this file leaks handles.
+      const longTermServers = [...servers];
+      await startServers(3, 303);
+      const temporary = [...urls];
+      servers = [...longTermServers, ...servers];
+      urls = [...longTerm, ...temporary];
+      mockPeerManager.getPeerUrls = mock.fn(() => urls);
+      mockPeerManager.selectPeers = mock.fn(() => urls);
+
+      const client = createTestClient();
+      const result = await broadcast(client, true);
+
+      assert.equal(result.successCount, urls.length);
+      assert.equal(result.longTermSuccessCount, longTerm.length);
+      assert.equal(result.temporarySuccessCount, temporary.length);
+      assert.equal(
+        result.longTermSuccessCount + result.temporarySuccessCount,
+        result.successCount,
+      );
+    });
+
+    it('counts a broadcast accepted only into disk pools as entirely temporary', async () => {
+      await startServers(4, 303);
+      const client = createTestClient();
+      const result = await broadcast(client, true);
+
+      assert.equal(result.successCount, urls.length);
+      assert.equal(result.temporarySuccessCount, urls.length);
+      assert.equal(result.longTermSuccessCount, 0);
+    });
+
     it('bails out of the dead peer tail in continuePastThreshold mode', async () => {
       const deadUrls = await makeDeadUrls(25);
       // Three live peers first, then a long dead tail.

src/arweave/composite-client.ts

@@ -497,6 +497,11 @@ export class ArweaveCompositeClient
   }
 
   private async postChunkToPeer(task: ChunkPostTask): Promise<ChunkPostResult> {
+    // Assigned just before the request so the catch can ask whether the abort
+    // was our own deadline firing rather than a caller cancelling. Created
+    // lazily: the dry-run paths below return without posting, and creating the
+    // signal up front would arm a timer per call for nothing.
+    let abortSignal: AbortSignal | undefined;
     try {
       this.failureSimulator.maybeFail();
 
@@ -517,6 +522,7 @@ export class ArweaveCompositeClient
           metrics.arweaveChunkPostCounter.inc({
             endpoint: task.peer,
             status: 'success',
+            reason: 'dry_run',
           });
 
           return {
@@ -548,6 +554,7 @@ export class ArweaveCompositeClient
           metrics.arweaveChunkPostCounter.inc({
             endpoint: task.peer,
             status: 'fail',
+            reason: 'invalid_chunk',
           });
 
           return {
@@ -600,6 +607,7 @@ export class ArweaveCompositeClient
           metrics.arweaveChunkPostCounter.inc({
             endpoint: task.peer,
             status: 'fail',
+            reason: 'invalid_proof',
           });
 
           return {
@@ -611,6 +619,7 @@ export class ArweaveCompositeClient
         metrics.arweaveChunkPostCounter.inc({
           endpoint: task.peer,
           status: 'success',
+          reason: 'dry_run',
         });
 
         return {
@@ -619,11 +628,13 @@ export class ArweaveCompositeClient
         };
       }
 
+      abortSignal = AbortSignal.timeout(task.abortTimeout);
+
       const response = await axios({
         method: 'POST',
         url: `${task.peer}/chunk`,
         data: task.chunk,
-        signal: AbortSignal.timeout(task.abortTimeout),
+        signal: abortSignal,
         timeout: task.responseTimeout,
         headers: task.headers,
         // An arweave node returns 200 when it will store the chunk long-term and
@@ -643,6 +654,7 @@ export class ArweaveCompositeClient
       metrics.arweaveChunkPostCounter.inc({
         endpoint: task.peer,
         status: 'success',
+        reason: String(response.status),
       });
       if (temporary) {
         metrics.arweaveChunkPostTemporaryCounter.inc({ endpoint: task.peer });
@@ -657,24 +669,58 @@ export class ArweaveCompositeClient
       let canceled = false;
       let timedOut = false;
 
-      if (axios.isAxiosError(error)) {
-        timedOut = error.code === 'ECONNABORTED';
-        canceled = error.code === 'ERR_CANCELED';
+      const isAxiosError = axios.isAxiosError(error);
+      if (isAxiosError) {
+        // ECONNABORTED is the response timeout. An AbortSignal.timeout() firing
+        // surfaces as ERR_CANCELED, indistinguishable from a caller cancelling —
+        // but it is our own deadline, not the caller's, and abortTimeout is
+        // normally the lower of the two, so this is the common case. Ask the
+        // signal which it was: AbortSignal.timeout() sets reason to a
+        // TimeoutError DOMException, while an explicit abort does not.
+        // Misreporting it matters beyond the metric: aggregateStatusCode() maps
+        // canceled to 499 (Client Closed Request), blaming the uploader for a
+        // deadline of ours, where timedOut maps to 504.
+        const abortedByOurDeadline =
+          abortSignal?.aborted === true &&
+          abortSignal.reason?.name === 'TimeoutError';
+        timedOut = error.code === 'ECONNABORTED' || abortedByOurDeadline;
+        canceled = error.code === 'ERR_CANCELED' && !abortedByOurDeadline;
       }
 
+      // A peer that answered tells us why it refused the chunk (400 unknown
+      // data root, 429 rate limited, 503 overloaded); one that did not answer
+      // is a timeout, our own abort, or unreachable. These call for completely
+      // different operator responses, so record which it was.
+      const statusCode = error.response?.status;
+      const reason =
+        statusCode !== undefined
+          ? String(statusCode)
+          : timedOut
+            ? 'timeout'
+            : canceled
+              ? 'canceled'
+              : isAxiosError
+                ? 'network'
+                : // Not an HTTP failure at all: a throw from our own code path
+                  // (e.g. the failure simulator). Calling it "network" would
+                  // send an operator looking at the wrong thing.
+                  'error';
+
       metrics.arweaveChunkPostCounter.inc({
         endpoint: task.peer,
         status: 'fail',
+        reason,
       });
 
       this.log.debug('Failed to POST chunk to peer:', {
         peer: task.peer,
         error: error.message,
+        reason,
       });
 
       return {
         success: false,
-        statusCode: error.response?.status,
+        statusCode,
         error: error.message,
         canceled,
         timedOut,
@@ -1965,6 +2011,8 @@ export class ArweaveCompositeClient
         return {
           successCount: 0,
           preferredSuccessCount: 0,
+          temporarySuccessCount: 0,
+          longTermSuccessCount: 0,
           failureCount: 0,
           results: [],
         };
@@ -2158,6 +2206,7 @@ export class ArweaveCompositeClient
               statusCode,
               canceled: result.canceled ?? false,
               timedOut: result.timedOut ?? false,
+              temporary: result.temporary ?? false,
             };
           } catch (error: any) {
             failureCount++;
@@ -2192,10 +2241,28 @@ export class ArweaveCompositeClient
         }
       }
 
+      // Derived from `results` rather than incremented in the workers: the
+      // counters above are deliberately racy (they only gate early
+      // termination), while these are reported to callers.
+      const temporarySuccessCount = results.filter(
+        (r) => r.success && r.temporary === true,
+      ).length;
+      const longTermSuccessCount = results.filter(
+        (r) => r.success && r.temporary !== true,
+      ).length;
+
       const duration = Date.now() - startTime;
 
       span.setAttribute('chunk.broadcast.duration_ms', duration);
       span.setAttribute('chunk.broadcast.success_count', successCount);
+      span.setAttribute(
+        'chunk.broadcast.temporary_success_count',
+        temporarySuccessCount,
+      );
+      span.setAttribute(
+        'chunk.broadcast.long_term_success_count',
+        longTermSuccessCount,
+      );
       span.setAttribute(
         'chunk.broadcast.preferred_success_count',
         preferredSuccessCount,
@@ -2243,6 +2310,8 @@ export class ArweaveCompositeClient
       this.log.debug('Chunk broadcast complete', {
         successCount,
         preferredSuccessCount,
+        temporarySuccessCount,
+        longTermSuccessCount,
         failureCount,
         consecutive4xxFailures,
         totalPeers: sortedPeers.length,
@@ -2262,6 +2331,8 @@ export class ArweaveCompositeClient
       return {
         successCount,
         preferredSuccessCount,
+        temporarySuccessCount,
+        longTermSuccessCount,
         failureCount,
         results,
       };

src/data/rebroadcasting-chunk-source.test.ts

@@ -93,6 +93,8 @@ class MockChunkBroadcaster implements ChunkBroadcaster {
   public shouldFail = false;
   public result: BroadcastChunkResult = {
     successCount: 1,
+    temporarySuccessCount: 0,
+    longTermSuccessCount: 1,
     preferredSuccessCount: 0,
     failureCount: 0,
     results: [
@@ -151,6 +153,9 @@ class MockChunkBroadcaster implements ChunkBroadcaster {
     this.broadcastPromise = null;
     this.result = {
       successCount: 1,
+      preferredSuccessCount: 0,
+      temporarySuccessCount: 0,
+      longTermSuccessCount: 1,
       failureCount: 0,
       results: [
         {
@@ -379,6 +384,9 @@ describe('RebroadcastingChunkSource', () => {
     it('should not cache when success count below threshold', async () => {
       mockBroadcaster.result = {
         successCount: 0,
+        preferredSuccessCount: 0,
+        temporarySuccessCount: 0,
+        longTermSuccessCount: 0,
         failureCount: 1,
         results: [
           {
@@ -406,6 +414,9 @@ describe('RebroadcastingChunkSource', () => {
       // Fix broadcaster result
       mockBroadcaster.result = {
         successCount: 1,
+        preferredSuccessCount: 0,
+        temporarySuccessCount: 0,
+        longTermSuccessCount: 1,
         failureCount: 0,
         results: [
           {
@@ -471,6 +482,8 @@ describe('RebroadcastingChunkSource', () => {
           totalBroadcasts++;
           return {
             successCount: 1,
+            temporarySuccessCount: 0,
+            longTermSuccessCount: 1,
             preferredSuccessCount: 0,
             failureCount: 0,
             results: [