fix(desktop): gate transport retries to idempotent or provably-unsent requests
Follow-up hardening on #92977 (issue #92976). The cherry-picked retry wrapped every verb, so an ECONNRESET arriving after the backend had already processed a POST (prompt submitted, session created) would silently double-submit on retry. - Extract the transport policy into electron/api-transport.ts so it is unit-testable without Electron: keep-alive agent pools, transient error classification, and a verb-gated withRetry. - Retry rule: GET/HEAD/OPTIONS retry on any transient transport error; POST/PUT/PATCH/DELETE retry only when the request provably never reached the server (connect-phase failures like ECONNREFUSED / ENOTFOUND, or an error thrown before the body was flushed — requestState.bodySent === false). Ambiguous resets after the body went out surface to the caller; when in doubt, don't retry. - Separate keep-alive pools for JSON calls vs streaming downloads so long downloads can't starve latency-sensitive JSON calls. - Destroy pooled agents on app will-quit. - Tests: shouldRetryRequest truth table, withRetry behavior, plus LIVE transport tests against real misbehaving node HTTP servers: a GET burst where the server resets keep-alive sockets (bare attempt fails, retried succeeds) and a POST whose socket is RST after server-side processing (hit counter stays 1 — no double submit).
This commit is contained in:
Teknium
@@ -0,0 +1,352 @@
|
||||
/**
|
||||
* Unit + live-transport tests for the Electron main process's Hermes REST
|
||||
* retry policy (#92976 / PR #92977 salvage).
|
||||
*
|
||||
* The live tests run REAL node http servers that misbehave the way the
|
||||
* reported backend does (closing sockets under burst keep-alive traffic) and
|
||||
* prove two things end to end:
|
||||
*
|
||||
* - idempotent GETs that die with ECONNRESET are retried and succeed, where
|
||||
* a single bare attempt (pre-PR behavior) surfaces the raw reset;
|
||||
* - a POST whose socket is reset AFTER the server processed it is NOT
|
||||
* retried: the server-side hit counter stays at 1 and the error surfaces.
|
||||
*/
|
||||
import http from 'node:http'
|
||||
import type { AddressInfo } from 'node:net'
|
||||
import { afterAll, describe, expect, it } from 'vitest'
|
||||
|
||||
import {
|
||||
destroyKeepaliveAgents,
|
||||
downloadAgentFor,
|
||||
isIdempotentMethod,
|
||||
isTransientTransportError,
|
||||
jsonAgentFor,
|
||||
shouldRetryRequest,
|
||||
withRetry
|
||||
} from './api-transport'
|
||||
|
||||
function errWithCode(code: string, message = code): NodeJS.ErrnoException {
|
||||
const e: NodeJS.ErrnoException = new Error(message)
|
||||
e.code = code
|
||||
return e
|
||||
}
|
||||
|
||||
afterAll(() => {
|
||||
destroyKeepaliveAgents()
|
||||
})
|
||||
|
||||
describe('isTransientTransportError', () => {
|
||||
it('accepts transient socket-level codes and messages', () => {
|
||||
for (const code of ['ECONNRESET', 'ECONNREFUSED', 'EPIPE', 'ETIMEDOUT', 'ENOTFOUND', 'EAI_AGAIN']) {
|
||||
expect(isTransientTransportError(errWithCode(code))).toBe(true)
|
||||
}
|
||||
expect(isTransientTransportError(new Error('socket hang up'))).toBe(true)
|
||||
expect(isTransientTransportError(new Error('read ECONNRESET'))).toBe(true)
|
||||
})
|
||||
|
||||
it('rejects non-transport errors', () => {
|
||||
expect(isTransientTransportError(new Error('404: not found'))).toBe(false)
|
||||
expect(isTransientTransportError(new Error('Invalid JSON from http://x'))).toBe(false)
|
||||
expect(isTransientTransportError(null)).toBe(false)
|
||||
expect(isTransientTransportError(undefined)).toBe(false)
|
||||
})
|
||||
})
|
||||
|
||||
describe('isIdempotentMethod', () => {
|
||||
it.each([
|
||||
['GET', true],
|
||||
['get', true],
|
||||
['HEAD', true],
|
||||
['OPTIONS', true],
|
||||
['POST', false],
|
||||
['PUT', false],
|
||||
['PATCH', false],
|
||||
['DELETE', false],
|
||||
[undefined, true] // node http defaults omitted method to GET
|
||||
])('%s -> %s', (method, expected) => {
|
||||
expect(isIdempotentMethod(method)).toBe(expected)
|
||||
})
|
||||
})
|
||||
|
||||
describe('shouldRetryRequest truth table', () => {
|
||||
const reset = () => errWithCode('ECONNRESET', 'read ECONNRESET')
|
||||
const refused = () => errWithCode('ECONNREFUSED', 'connect ECONNREFUSED 127.0.0.1:1')
|
||||
const hangUp = () => new Error('socket hang up')
|
||||
|
||||
it('GET: retries any transient error regardless of body state', () => {
|
||||
expect(shouldRetryRequest(reset(), 'GET', { bodySent: true })).toBe(true)
|
||||
expect(shouldRetryRequest(reset(), 'GET', { bodySent: false })).toBe(true)
|
||||
expect(shouldRetryRequest(hangUp(), 'HEAD', { bodySent: true })).toBe(true)
|
||||
})
|
||||
|
||||
it('GET: never retries non-transport errors (HTTP 4xx/5xx surfaced as Error)', () => {
|
||||
expect(shouldRetryRequest(new Error('500: boom'), 'GET', { bodySent: true })).toBe(false)
|
||||
})
|
||||
|
||||
it('POST: retries when the connection provably never happened', () => {
|
||||
expect(shouldRetryRequest(refused(), 'POST', { bodySent: false })).toBe(true)
|
||||
expect(shouldRetryRequest(refused(), 'POST', { bodySent: true })).toBe(true) // refused == nothing sent
|
||||
expect(shouldRetryRequest(errWithCode('ENOTFOUND'), 'PUT', { bodySent: false })).toBe(true)
|
||||
})
|
||||
|
||||
it('POST: retries transient errors thrown before the body was flushed', () => {
|
||||
expect(shouldRetryRequest(reset(), 'POST', { bodySent: false })).toBe(true)
|
||||
expect(shouldRetryRequest(hangUp(), 'DELETE', { bodySent: false })).toBe(true)
|
||||
})
|
||||
|
||||
it('POST: does NOT retry ambiguous resets after the body went out', () => {
|
||||
expect(shouldRetryRequest(reset(), 'POST', { bodySent: true })).toBe(false)
|
||||
expect(shouldRetryRequest(hangUp(), 'POST', { bodySent: true })).toBe(false)
|
||||
expect(shouldRetryRequest(errWithCode('EPIPE'), 'PUT', { bodySent: true })).toBe(false)
|
||||
expect(shouldRetryRequest(errWithCode('ETIMEDOUT'), 'DELETE', { bodySent: true })).toBe(false)
|
||||
})
|
||||
|
||||
it('POST: conservative when request state is unknown', () => {
|
||||
// No bodySent flag at all — treat as "may have been sent", don't retry.
|
||||
expect(shouldRetryRequest(reset(), 'POST', {})).toBe(false)
|
||||
expect(shouldRetryRequest(reset(), 'POST')).toBe(false)
|
||||
})
|
||||
})
|
||||
|
||||
describe('withRetry', () => {
|
||||
const noDelay = { delayFn: () => Promise.resolve() }
|
||||
|
||||
it('retries a GET through transient failures and resolves', async () => {
|
||||
let attempts = 0
|
||||
const result = await withRetry(
|
||||
() => {
|
||||
attempts += 1
|
||||
if (attempts < 3) return Promise.reject(errWithCode('ECONNRESET'))
|
||||
return Promise.resolve('ok')
|
||||
},
|
||||
{ method: 'GET', ...noDelay }
|
||||
)
|
||||
expect(result).toBe('ok')
|
||||
expect(attempts).toBe(3)
|
||||
})
|
||||
|
||||
it('gives each attempt a fresh requestState', async () => {
|
||||
const seen: boolean[] = []
|
||||
let attempts = 0
|
||||
await withRetry(
|
||||
(state: any) => {
|
||||
seen.push(state.bodySent)
|
||||
state.bodySent = true
|
||||
attempts += 1
|
||||
if (attempts < 2) return Promise.reject(errWithCode('ECONNREFUSED'))
|
||||
return Promise.resolve(null)
|
||||
},
|
||||
{ method: 'POST', ...noDelay }
|
||||
)
|
||||
expect(seen).toEqual([false, false])
|
||||
})
|
||||
|
||||
it('does not retry a POST that failed after the body was flushed', async () => {
|
||||
let attempts = 0
|
||||
await expect(
|
||||
withRetry(
|
||||
(state: any) => {
|
||||
attempts += 1
|
||||
state.bodySent = true
|
||||
return Promise.reject(errWithCode('ECONNRESET', 'read ECONNRESET'))
|
||||
},
|
||||
{ method: 'POST', ...noDelay }
|
||||
)
|
||||
).rejects.toThrow('read ECONNRESET')
|
||||
expect(attempts).toBe(1)
|
||||
})
|
||||
|
||||
it('retries a POST on ECONNREFUSED (never reached the server)', async () => {
|
||||
let attempts = 0
|
||||
await expect(
|
||||
withRetry(
|
||||
() => {
|
||||
attempts += 1
|
||||
return Promise.reject(errWithCode('ECONNREFUSED'))
|
||||
},
|
||||
{ method: 'POST', maxRetries: 2, ...noDelay }
|
||||
)
|
||||
).rejects.toThrow('ECONNREFUSED')
|
||||
expect(attempts).toBe(3)
|
||||
})
|
||||
|
||||
it('bounds retries at maxRetries even for GET', async () => {
|
||||
let attempts = 0
|
||||
await expect(
|
||||
withRetry(
|
||||
() => {
|
||||
attempts += 1
|
||||
return Promise.reject(errWithCode('ECONNRESET'))
|
||||
},
|
||||
{ method: 'GET', maxRetries: 2, ...noDelay }
|
||||
)
|
||||
).rejects.toThrow()
|
||||
expect(attempts).toBe(3)
|
||||
})
|
||||
|
||||
it('never retries non-transient errors', async () => {
|
||||
let attempts = 0
|
||||
await expect(
|
||||
withRetry(
|
||||
() => {
|
||||
attempts += 1
|
||||
return Promise.reject(new Error('500: internal'))
|
||||
},
|
||||
{ method: 'GET', ...noDelay }
|
||||
)
|
||||
).rejects.toThrow('500')
|
||||
expect(attempts).toBe(1)
|
||||
})
|
||||
})
|
||||
|
||||
describe('keep-alive agent pools', () => {
|
||||
it('separates JSON and download pools per protocol', () => {
|
||||
expect(jsonAgentFor('http:')).not.toBe(jsonAgentFor('https:'))
|
||||
expect(jsonAgentFor('http:')).not.toBe(downloadAgentFor('http:'))
|
||||
expect(jsonAgentFor('https:')).not.toBe(downloadAgentFor('https:'))
|
||||
// Stable across calls (a real pool, not a factory).
|
||||
expect(jsonAgentFor('http:')).toBe(jsonAgentFor('http:'))
|
||||
})
|
||||
})
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// LIVE transport tests against real misbehaving HTTP servers.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/** Minimal single-attempt GET mirroring the pre-PR fetchJson (no retry). */
|
||||
function bareJsonGet(url: string): Promise<any> {
|
||||
return new Promise((resolve, reject) => {
|
||||
const req = http.request(new URL(url), { agent: jsonAgentFor('http:'), method: 'GET' }, res => {
|
||||
const chunks: Buffer[] = []
|
||||
res.on('error', reject)
|
||||
res.on('data', c => chunks.push(c))
|
||||
res.on('end', () => resolve(JSON.parse(Buffer.concat(chunks).toString('utf8'))))
|
||||
})
|
||||
req.on('error', reject)
|
||||
req.end()
|
||||
})
|
||||
}
|
||||
|
||||
/** The head behavior: same request under the verb-gated retry policy. */
|
||||
function retriedJsonGet(url: string): Promise<any> {
|
||||
return withRetry(() => bareJsonGet(url), { method: 'GET', delayFn: () => Promise.resolve() })
|
||||
}
|
||||
|
||||
function listen(server: http.Server): Promise<string> {
|
||||
return new Promise(resolve => {
|
||||
server.listen(0, '127.0.0.1', () => {
|
||||
resolve(`http://127.0.0.1:${(server.address() as AddressInfo).port}`)
|
||||
})
|
||||
})
|
||||
}
|
||||
|
||||
describe('live: GET burst against a server that resets keep-alive sockets', () => {
|
||||
it('bare attempts fail with ECONNRESET/hang-up; retried GETs all succeed', async () => {
|
||||
// Deterministic misbehavior: every other request gets its socket
|
||||
// destroyed instead of a response — the observable client-side effect of
|
||||
// a backend killing idle keep-alive sockets mid-burst.
|
||||
let hits = 0
|
||||
const server = http.createServer((req, res) => {
|
||||
hits += 1
|
||||
if (hits % 2 === 1) {
|
||||
req.socket.destroy()
|
||||
return
|
||||
}
|
||||
res.setHeader('content-type', 'application/json')
|
||||
res.end(JSON.stringify({ n: hits }))
|
||||
})
|
||||
const base = await listen(server)
|
||||
|
||||
try {
|
||||
// BASE (pre-PR, single attempt): the burst surfaces raw transport errors.
|
||||
let baseFailures = 0
|
||||
for (let i = 0; i < 6; i++) {
|
||||
try {
|
||||
await bareJsonGet(`${base}/api/sessions`)
|
||||
} catch (error: any) {
|
||||
baseFailures += 1
|
||||
expect(isTransientTransportError(error)).toBe(true)
|
||||
}
|
||||
}
|
||||
expect(baseFailures).toBeGreaterThan(0)
|
||||
|
||||
// HEAD (retry policy): the same burst fully succeeds. Sequential so the
|
||||
// server's alternating destroy/respond pattern is deterministic per
|
||||
// request (first attempt reset, retry served).
|
||||
for (let i = 0; i < 6; i++) {
|
||||
const r = await retriedJsonGet(`${base}/api/sessions`)
|
||||
expect(r).toHaveProperty('n')
|
||||
}
|
||||
} finally {
|
||||
server.close()
|
||||
}
|
||||
}, 20_000)
|
||||
})
|
||||
|
||||
describe('live: POST reset after server-side processing', () => {
|
||||
it('does not double-submit: server hit count stays 1, error surfaces', async () => {
|
||||
// The server fully receives and "processes" the POST (counter increments),
|
||||
// then RSTs the socket before responding — the dangerous ambiguous case.
|
||||
let posts = 0
|
||||
const server = http.createServer((req, res) => {
|
||||
const chunks: Buffer[] = []
|
||||
req.on('data', c => chunks.push(c))
|
||||
req.on('end', () => {
|
||||
posts += 1 // processed: prompt submitted / session created
|
||||
req.socket.resetAndDestroy()
|
||||
void res
|
||||
})
|
||||
})
|
||||
const base = await listen(server)
|
||||
|
||||
const postOnce = () =>
|
||||
withRetry(
|
||||
(state: any) =>
|
||||
new Promise((resolve, reject) => {
|
||||
const body = Buffer.from(JSON.stringify({ prompt: 'hello' }))
|
||||
const req = http.request(
|
||||
new URL(`${base}/api/prompt`),
|
||||
{
|
||||
agent: jsonAgentFor('http:'),
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/json', 'content-length': String(body.length) }
|
||||
},
|
||||
res => {
|
||||
res.resume()
|
||||
res.on('end', () => resolve(null))
|
||||
}
|
||||
)
|
||||
req.on('error', reject)
|
||||
state.bodySent = true
|
||||
req.write(body)
|
||||
req.end()
|
||||
}),
|
||||
{ method: 'POST', delayFn: () => Promise.resolve() }
|
||||
)
|
||||
|
||||
try {
|
||||
await expect(postOnce()).rejects.toSatisfy((error: any) => isTransientTransportError(error))
|
||||
expect(posts).toBe(1) // exactly one server-side submission — no retry
|
||||
} finally {
|
||||
server.close()
|
||||
}
|
||||
}, 20_000)
|
||||
|
||||
it('sanity: an identical GET-shaped retry WOULD have re-hit the server', async () => {
|
||||
// Companion proof that the verb gate (not luck) is what kept posts === 1:
|
||||
// the same reset-after-processing server sees multiple hits under GET.
|
||||
let gets = 0
|
||||
const server = http.createServer(req => {
|
||||
gets += 1
|
||||
req.socket.resetAndDestroy()
|
||||
})
|
||||
const base = await listen(server)
|
||||
|
||||
try {
|
||||
await expect(retriedJsonGet(`${base}/api/thing`)).rejects.toThrow()
|
||||
expect(gets).toBeGreaterThan(1) // retried — proves the machinery fires
|
||||
} finally {
|
||||
server.close()
|
||||
}
|
||||
}, 20_000)
|
||||
})
|
||||
@@ -0,0 +1,144 @@
|
||||
/**
|
||||
* Shared HTTP transport policy for the Electron main process's Hermes REST
|
||||
* helpers (fetchJson / fetchPublicJson / downloadViaTokenToFile).
|
||||
*
|
||||
* Two concerns live here so they can be unit-tested without Electron:
|
||||
*
|
||||
* 1. Connection-pooled keep-alive agents. Opening a fresh TCP socket per call
|
||||
* is what produced the burst-traffic ECONNRESET storms (#92976): the
|
||||
* backend closes idle keep-alive sockets and the next write on a reused
|
||||
* raw socket dies with 'socket hang up'. JSON calls and streaming
|
||||
* downloads get SEPARATE pools so a handful of long-lived download
|
||||
* streams can never starve the small, latency-sensitive JSON calls out of
|
||||
* the socket pool.
|
||||
*
|
||||
* 2. A retry policy that is safe for non-idempotent verbs. A transient
|
||||
* transport error does NOT mean the server didn't process the request —
|
||||
* an ECONNRESET can arrive after the backend already handled a POST
|
||||
* (created the session, submitted the prompt) and merely lost the socket
|
||||
* before the response was read. Blindly retrying every verb double-submits.
|
||||
*
|
||||
* The rule implemented by shouldRetryRequest():
|
||||
* - Idempotent verbs (GET / HEAD / OPTIONS) retry on any transient
|
||||
* transport error — replaying them is harmless by definition.
|
||||
* - Non-idempotent verbs (POST / PUT / PATCH / DELETE) retry ONLY when
|
||||
* the request provably never reached the server:
|
||||
* a) connection-establishment failures (ECONNREFUSED, ENOTFOUND,
|
||||
* EAI_AGAIN, EHOSTUNREACH, ENETUNREACH) — no connection means no
|
||||
* request; or
|
||||
* b) a transient error thrown before we started flushing the
|
||||
* request (requestState.bodySent === false).
|
||||
* Anything ambiguous — ECONNRESET / EPIPE / 'socket hang up' after
|
||||
* the body went out — is NOT retried; the error surfaces to the
|
||||
* caller. When in doubt, don't retry a non-idempotent request.
|
||||
*/
|
||||
|
||||
import http from 'node:http'
|
||||
import https from 'node:https'
|
||||
|
||||
// JSON pool: many small concurrent calls (session lists, config, prompts).
|
||||
const HTTP_JSON_AGENT = new http.Agent({ keepAlive: true, maxSockets: 50 })
|
||||
const HTTPS_JSON_AGENT = new https.Agent({ keepAlive: true, maxSockets: 50 })
|
||||
|
||||
// Download pool: few long-lived streaming bodies. Isolated from the JSON pool
|
||||
// so saturating it with large file downloads can't block interactive calls.
|
||||
const HTTP_DOWNLOAD_AGENT = new http.Agent({ keepAlive: true, maxSockets: 8 })
|
||||
const HTTPS_DOWNLOAD_AGENT = new https.Agent({ keepAlive: true, maxSockets: 8 })
|
||||
|
||||
function jsonAgentFor(protocol) {
|
||||
return protocol === 'https:' ? HTTPS_JSON_AGENT : HTTP_JSON_AGENT
|
||||
}
|
||||
|
||||
function downloadAgentFor(protocol) {
|
||||
return protocol === 'https:' ? HTTPS_DOWNLOAD_AGENT : HTTP_DOWNLOAD_AGENT
|
||||
}
|
||||
|
||||
// Close pooled sockets so lingering keep-alive connections can't hold the
|
||||
// process open (or leak FDs) across quit. Wired to app 'will-quit' in main.ts.
|
||||
function destroyKeepaliveAgents() {
|
||||
for (const agent of [HTTP_JSON_AGENT, HTTPS_JSON_AGENT, HTTP_DOWNLOAD_AGENT, HTTPS_DOWNLOAD_AGENT]) {
|
||||
agent.destroy()
|
||||
}
|
||||
}
|
||||
|
||||
// Transient transport errors: retry MAY be safe (subject to verb gating).
|
||||
const TRANSIENT_CODES = new Set(['ECONNRESET', 'ECONNREFUSED', 'EPIPE', 'ETIMEDOUT', 'EAI_AGAIN', 'ENOTFOUND', 'EHOSTUNREACH', 'ENETUNREACH'])
|
||||
|
||||
// Errors that prove the request never reached the server: the TCP connection
|
||||
// (or name resolution) failed outright, so nothing was submitted.
|
||||
const NEVER_SENT_CODES = new Set(['ECONNREFUSED', 'ENOTFOUND', 'EAI_AGAIN', 'EHOSTUNREACH', 'ENETUNREACH'])
|
||||
|
||||
const IDEMPOTENT_METHODS = new Set(['GET', 'HEAD', 'OPTIONS'])
|
||||
|
||||
function isIdempotentMethod(method) {
|
||||
return IDEMPOTENT_METHODS.has(String(method || 'GET').toUpperCase())
|
||||
}
|
||||
|
||||
function isTransientTransportError(error) {
|
||||
if (!error) return false
|
||||
if (TRANSIENT_CODES.has(error.code)) return true
|
||||
const msg = String(error.message || '')
|
||||
return msg.includes('socket hang up') || msg.includes('read ECONNRESET')
|
||||
}
|
||||
|
||||
/**
|
||||
* The verb-gated retry decision.
|
||||
*
|
||||
* @param error the transport error from the failed attempt
|
||||
* @param method HTTP verb of the request ('GET', 'POST', ...)
|
||||
* @param requestState per-attempt state; requestState.bodySent is set true by
|
||||
* the caller just BEFORE the first byte of the request is
|
||||
* flushed, so a `false` here proves nothing went out.
|
||||
*/
|
||||
function shouldRetryRequest(error, method, requestState: any = {}) {
|
||||
if (!isTransientTransportError(error)) return false
|
||||
if (isIdempotentMethod(method)) return true
|
||||
|
||||
// Non-idempotent: only when the request provably never reached the server.
|
||||
if (NEVER_SENT_CODES.has(error && error.code)) return true
|
||||
if (requestState.bodySent === false) return true
|
||||
|
||||
// Ambiguous (reset/hang-up after the body was flushed): the server may have
|
||||
// processed it. Surface the error rather than risk a double submit.
|
||||
return false
|
||||
}
|
||||
|
||||
/**
|
||||
* Run `makeAttempt` with bounded retries under the policy above.
|
||||
*
|
||||
* `makeAttempt(requestState)` must return a Promise and should set
|
||||
* `requestState.bodySent = true` immediately before flushing the request
|
||||
* (before the first req.write()/req.end()). Each attempt gets a fresh state
|
||||
* object initialized to { bodySent: false }.
|
||||
*/
|
||||
async function withRetry(makeAttempt, options: any = {}) {
|
||||
const method = String(options.method || 'GET').toUpperCase()
|
||||
const maxRetries = Number.isInteger(options.maxRetries) ? options.maxRetries : 2
|
||||
const delayFn = options.delayFn || (attempt => new Promise(r => setTimeout(r, Math.min(200 * Math.pow(2, attempt), 2000))))
|
||||
|
||||
let lastError
|
||||
for (let attempt = 0; attempt <= maxRetries; attempt++) {
|
||||
const requestState = { bodySent: false }
|
||||
try {
|
||||
return await makeAttempt(requestState)
|
||||
} catch (error) {
|
||||
lastError = error
|
||||
if (attempt < maxRetries && shouldRetryRequest(error, method, requestState)) {
|
||||
await delayFn(attempt)
|
||||
continue
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
throw lastError
|
||||
}
|
||||
|
||||
export {
|
||||
destroyKeepaliveAgents,
|
||||
downloadAgentFor,
|
||||
isIdempotentMethod,
|
||||
isTransientTransportError,
|
||||
jsonAgentFor,
|
||||
shouldRetryRequest,
|
||||
withRetry
|
||||
}
|
||||
@@ -168,6 +168,7 @@ import {
|
||||
pumpStreamToFile
|
||||
} from './gateway-file-download'
|
||||
import { probeGatewayWebSocket } from './gateway-ws-probe'
|
||||
import { destroyKeepaliveAgents, downloadAgentFor, jsonAgentFor, withRetry } from './api-transport'
|
||||
import { registerGitIpc } from './git-ipc'
|
||||
import { clearStaleGitLocks } from './gitlock'
|
||||
import { readAndConsumeHandoffResult } from './handoff-result'
|
||||
@@ -4856,40 +4857,11 @@ function multipartBody(upload) {
|
||||
return { body, contentType: `multipart/form-data; boundary=${boundary}` }
|
||||
}
|
||||
|
||||
// Connection-pooled HTTP agents to avoid ECONNRESET from per-call TCP sockets.
|
||||
// Each fetchJson / fetchPublicJson / downloadViaTokenToFile call shares these
|
||||
// instead of opening a fresh connection every time.
|
||||
const HTTP_KEEPALIVE_AGENT = new http.Agent({ keepAlive: true, maxSockets: 50 })
|
||||
const HTTPS_KEEPALIVE_AGENT = new https.Agent({ keepAlive: true, maxSockets: 50 })
|
||||
|
||||
const RETRYABLE_CODES = new Set(['ECONNRESET', 'ECONNREFUSED', 'EPIPE', 'ERR_NETWORK'])
|
||||
|
||||
function isRetryableError(error) {
|
||||
if (!error) return false
|
||||
if (RETRYABLE_CODES.has(error.code)) return true
|
||||
const msg = String(error.message || '')
|
||||
return msg.includes('socket hang up') || msg.includes('read ECONNRESET')
|
||||
}
|
||||
|
||||
async function withRetry(fn, maxRetries = 2) {
|
||||
let lastError
|
||||
for (let attempt = 0; attempt <= maxRetries; attempt++) {
|
||||
try {
|
||||
return await fn()
|
||||
} catch (error) {
|
||||
lastError = error
|
||||
if (attempt < maxRetries && isRetryableError(error)) {
|
||||
await new Promise(r => setTimeout(r, Math.min(200 * Math.pow(2, attempt), 2000)))
|
||||
continue
|
||||
}
|
||||
throw error
|
||||
}
|
||||
}
|
||||
throw lastError
|
||||
}
|
||||
|
||||
function fetchJson(url, token, options: any = {}) {
|
||||
return withRetry(() => new Promise((resolve, reject) => {
|
||||
// Retry policy lives in api-transport.ts: idempotent verbs retry on any
|
||||
// transient transport error; POST/PUT/DELETE only when the request provably
|
||||
// never reached the server (see shouldRetryRequest) — never double-submit.
|
||||
return withRetry((requestState: any) => new Promise((resolve, reject) => {
|
||||
const { body, contentType } = options.upload
|
||||
? multipartBody(options.upload)
|
||||
: {
|
||||
@@ -4899,7 +4871,7 @@ function fetchJson(url, token, options: any = {}) {
|
||||
|
||||
const parsed = new URL(url)
|
||||
const client = parsed.protocol === 'https:' ? https : http
|
||||
const agent = parsed.protocol === 'https:' ? HTTPS_KEEPALIVE_AGENT : HTTP_KEEPALIVE_AGENT
|
||||
const agent = jsonAgentFor(parsed.protocol)
|
||||
const timeoutMs = resolveTimeoutMs(options.timeoutMs, DEFAULT_FETCH_TIMEOUT_MS)
|
||||
|
||||
if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') {
|
||||
@@ -4978,12 +4950,17 @@ function fetchJson(url, token, options: any = {}) {
|
||||
req.destroy(new Error(`Timed out connecting to Hermes backend after ${timeoutMs}ms`))
|
||||
})
|
||||
|
||||
// From here the request goes on the wire: a later transport error can no
|
||||
// longer prove the server didn't process it, so non-idempotent verbs must
|
||||
// not be retried past this point.
|
||||
requestState.bodySent = true
|
||||
|
||||
if (body) {
|
||||
req.write(body)
|
||||
}
|
||||
|
||||
req.end()
|
||||
}))
|
||||
}), { method: options.method || 'GET' })
|
||||
}
|
||||
|
||||
// Token-auth download that streams the response body straight to a
|
||||
@@ -5010,7 +4987,7 @@ function downloadViaTokenToFile(url, token, ctx, options: any = {}) {
|
||||
}
|
||||
|
||||
const client = parsed.protocol === 'https:' ? https : http
|
||||
const agent = parsed.protocol === 'https:' ? HTTPS_KEEPALIVE_AGENT : HTTP_KEEPALIVE_AGENT
|
||||
const agent = downloadAgentFor(parsed.protocol)
|
||||
const timeoutMs = resolveTimeoutMs(options.timeoutMs, DEFAULT_FETCH_TIMEOUT_MS)
|
||||
|
||||
const req = client.request(
|
||||
@@ -5051,7 +5028,7 @@ function fetchPublicJson(url, options: any = {}) {
|
||||
// NO ``X-Hermes-Session-Token`` header — used by the auth-mode probe before
|
||||
// any credentials exist, and any time we must not leak a token to an
|
||||
// endpoint that doesn't need one.
|
||||
return withRetry(() => new Promise((resolve, reject) => {
|
||||
return withRetry((requestState: any) => new Promise((resolve, reject) => {
|
||||
const body = options.body === undefined ? undefined : Buffer.from(JSON.stringify(options.body))
|
||||
let parsed
|
||||
|
||||
@@ -5064,7 +5041,7 @@ function fetchPublicJson(url, options: any = {}) {
|
||||
}
|
||||
|
||||
const client = parsed.protocol === 'https:' ? https : http
|
||||
const agent = parsed.protocol === 'https:' ? HTTPS_KEEPALIVE_AGENT : HTTP_KEEPALIVE_AGENT
|
||||
const agent = jsonAgentFor(parsed.protocol)
|
||||
const timeoutMs = resolveTimeoutMs(options.timeoutMs, DEFAULT_FETCH_TIMEOUT_MS)
|
||||
|
||||
if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') {
|
||||
@@ -5131,12 +5108,15 @@ function fetchPublicJson(url, options: any = {}) {
|
||||
req.destroy(new Error(`Timed out connecting to Hermes backend after ${timeoutMs}ms`))
|
||||
})
|
||||
|
||||
// Past this point the request is on the wire — see fetchJson.
|
||||
requestState.bodySent = true
|
||||
|
||||
if (body) {
|
||||
req.write(body)
|
||||
}
|
||||
|
||||
req.end()
|
||||
}))
|
||||
}), { method: options.method || 'GET' })
|
||||
}
|
||||
|
||||
function mimeTypeForPath(filePath) {
|
||||
@@ -14520,6 +14500,12 @@ app.on('before-quit', () => {
|
||||
}
|
||||
})
|
||||
|
||||
// Close the pooled keep-alive sockets on quit so lingering connections can't
|
||||
// hold the event loop open or leak FDs past app teardown.
|
||||
app.on('will-quit', () => {
|
||||
destroyKeepaliveAgents()
|
||||
})
|
||||
|
||||
// Answered synchronously so preload can publish the verdict before the
|
||||
// renderer's first script — see the note there on why it cannot decide this
|
||||
// itself. Registered at module scope, which runs long before any window.
|
||||
|
||||
Reference in New Issue
Block a user