diff --git a/src/arweave/composite-client.test.ts b/src/arweave/composite-client.test.ts index 3a771fdcf..a08fc0165 100644 --- a/src/arweave/composite-client.test.ts +++ b/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. diff --git a/src/arweave/composite-client.ts b/src/arweave/composite-client.ts index 8459e2758..e0b9b7508 100644 --- a/src/arweave/composite-client.ts +++ b/src/arweave/composite-client.ts @@ -497,6 +497,11 @@ export class ArweaveCompositeClient } private async postChunkToPeer(task: ChunkPostTask): Promise { + // 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, }; diff --git a/src/data/rebroadcasting-chunk-source.test.ts b/src/data/rebroadcasting-chunk-source.test.ts index 895654035..4a6af1cf2 100644 --- a/src/data/rebroadcasting-chunk-source.test.ts +++ b/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: [ diff --git a/src/metrics.ts b/src/metrics.ts index 4f28e414a..56a392e48 100644 --- a/src/metrics.ts +++ b/src/metrics.ts @@ -603,10 +603,26 @@ export const arweaveTxFetchCounter = new promClient.Counter({ labelNames: ['node_type'], }); +/** + * `reason` carries the outcome behind `status`: + * + * - success: the accepted HTTP status, "200" (peer will store it long-term) or + * "303" (peer parked it in its disk pool), or "dry_run" when posting is + * simulated. The 303 subset is also counted by + * `arweave_chunk_post_temporary_total`. + * - fail: the peer's HTTP status as a string when it answered ("400", "429", + * "503"); otherwise "timeout" (our response or abort deadline), "canceled" + * (the caller aborted), "network" (unreachable), or "invalid_chunk" / + * "invalid_proof" for dry-run validation failures. + * + * Without it, a peer rejecting chunks is indistinguishable from one + * rate-limiting us or one we cannot reach, and telling them apart otherwise + * takes the peer operator's own logs. + */ export const arweaveChunkPostCounter = new promClient.Counter({ name: 'arweave_chunk_post_total', help: 'Counts individual POST request to endpoint', - labelNames: ['endpoint', 'status', 'role'], + labelNames: ['endpoint', 'status', 'role', 'reason'], }); /** diff --git a/src/types.d.ts b/src/types.d.ts index 623ef1039..3b1cff88b 100644 --- a/src/types.d.ts +++ b/src/types.d.ts @@ -936,6 +936,8 @@ type BroadcastChunkResponses = { statusCode: number; canceled: boolean; timedOut: boolean; + /** Peer answered 303: stored in its disk pool, not its long-term home. */ + temporary?: boolean; skipped?: boolean; skipReason?: | 'success_threshold' @@ -947,6 +949,22 @@ interface BroadcastChunkResult { successCount: number; preferredSuccessCount: number; failureCount: number; + /** + * How many accepting peers answered 303 ("temporary"): they + * persisted the chunk into their disk pool but are not the long-term home for + * that offset. `longTermSuccessCount` is the 200 remainder. Both outcomes are + * successful propagation — a chunk whose transaction is still pending has no + * absolute offset yet, so 303 is the expected answer even from the tip nodes + * — but the split is the difference between "peers that will keep this" and + * "peers that will drop it when their disk pool matures", which callers + * cannot otherwise see. + * + * Both are derived from `results`, which is authoritative, rather than from + * the early-termination counters above — so their sum can differ slightly + * from `successCount`, which is deliberately racy. + */ + temporarySuccessCount: number; + longTermSuccessCount: number; results: BroadcastChunkResponses[]; }