From 060a55e6912508b00ca162a2f7d4de2cc124f0c5 Mon Sep 17 00:00:00 2001 From: Daniel Roe Date: Fri, 10 Jul 2026 13:00:56 +0200 Subject: [PATCH] fix: retry transient knot disconnects within the sync job --- server/utils/git-wire/errors.ts | 27 ++++++++++++ server/utils/splice.ts | 45 +++++++++++++++++-- test/unit/transient-transport.spec.ts | 63 +++++++++++++++++++++++++++ 3 files changed, 131 insertions(+), 4 deletions(-) create mode 100644 test/unit/transient-transport.spec.ts diff --git a/server/utils/git-wire/errors.ts b/server/utils/git-wire/errors.ts index 88eb402..d6830a0 100644 --- a/server/utils/git-wire/errors.ts +++ b/server/utils/git-wire/errors.ts @@ -33,6 +33,33 @@ export class RemoteRejectedError extends WireError { } } +/** + * True for failures that are worth retrying immediately within the same job: a + * flaky knot SSH endpoint that resets or times out the connection at the + * handshake, or drops the channel mid-stream. These clear within seconds, so a + * short in-job retry usually succeeds where waiting out the queue's minutes-long + * backoff would leave the mirror stale. + * + * A `RemoteRejectedError` is never transient-transport: it's a protocol-level + * verdict (CAS loss, repo gone, auth refused, pack too big) that a reconnect + * won't change, so those fall through to the caller's own handling. + */ +export function isTransientTransport(err: unknown): boolean { + if (!(err instanceof WireError) || err instanceof RemoteRejectedError) return false + const lc = err.message.toLowerCase() + return ( + lc.includes('econnreset') + || lc.includes('connection reset') + || lc.includes('timed out') + || lc.includes('handshake') + || lc.includes('no advertisement') + || lc.includes('connection lost') + || lc.includes('stdin closed') + || lc.includes('receive-pack exited') + || lc.includes('empty report-status') + ) +} + /** * Classify ssh / sshd / knot stderr (the child process's stderr band, since * we deliberately do not request side-band multiplexing). Returns null when diff --git a/server/utils/splice.ts b/server/utils/splice.ts index 08835ea..e06939f 100644 --- a/server/utils/splice.ts +++ b/server/utils/splice.ts @@ -1,3 +1,4 @@ +import { isTransientTransport } from './git-wire/errors' import { type ReceivePackFactory, ReceivePackSession, @@ -13,6 +14,36 @@ const DEFAULT_MAX_PACK_BYTES = 1024 * 1024 * 1024 /** Cap haves so a repo with thousands of refs can't bloat the negotiation. */ const MAX_HAVES = 256 +/** + * In-job retry for the flaky knot SSH endpoint. A burst of connection resets at + * the handshake typically clears within a second or two, so retrying a few + * times inside the job lands the push now rather than after the queue's + * minutes-long backoff (or, worse, after the attempt ceiling gives up and + * leaves the mirror stale until the next push). Only transient transport + * failures are retried; protocol verdicts re-throw immediately. + */ +const TRANSPORT_RETRIES = 3 +const TRANSPORT_RETRY_DELAY_MS = 1_500 + +export async function withTransportRetry(op: () => Promise, delayMs = TRANSPORT_RETRY_DELAY_MS): Promise { + let lastErr: unknown + for (let attempt = 0; attempt <= TRANSPORT_RETRIES; attempt++) { + try { + // eslint-disable-next-line no-await-in-loop -- sequential retries by design + return await op() + } + catch (err) { + if (!isTransientTransport(err)) throw err + lastErr = err + if (attempt < TRANSPORT_RETRIES) { + // eslint-disable-next-line no-await-in-loop -- deliberate backoff between retries + await new Promise(resolve => setTimeout(resolve, delayMs)) + } + } + } + throw lastErr +} + function maxPackBytes(): number { const raw = process.env.NUXT_MAX_PACK_BYTES if (!raw) return DEFAULT_MAX_PACK_BYTES @@ -60,8 +91,12 @@ export interface SplicePushResult { * keeps the knot's advertised tip as the authoritative compare-and-swap base. */ export async function splicePush(params: SplicePushParams): Promise { - const factory = await sshFactory(params.installationId, params.knot, params.repoDid) - return runSplice(factory, params) + return withTransportRetry(async () => { + // Fresh factory (and therefore fresh SSH connection) per attempt: a reset + // connection can't be reused. + const factory = await sshFactory(params.installationId, params.knot, params.repoDid) + return runSplice(factory, params) + }) } /** The fetch + push exchange over an open session. Split out for the wire test. */ @@ -118,8 +153,10 @@ export async function spliceDelete(params: { repoDid: string ref: string }): Promise { - const factory = await sshFactory(params.installationId, params.knot, params.repoDid) - return runSpliceDelete(factory, params.ref) + return withTransportRetry(async () => { + const factory = await sshFactory(params.installationId, params.knot, params.repoDid) + return runSpliceDelete(factory, params.ref) + }) } /** The delete exchange over an open session. Split out for the wire test. */ diff --git a/test/unit/transient-transport.spec.ts b/test/unit/transient-transport.spec.ts new file mode 100644 index 0000000..371530a --- /dev/null +++ b/test/unit/transient-transport.spec.ts @@ -0,0 +1,63 @@ +import { describe, expect, it, vi } from 'vitest' +import { isTransientTransport, RemoteRejectedError, WireError } from '../../server/utils/git-wire/errors' +import { withTransportRetry } from '../../server/utils/splice' + +describe('isTransientTransport', () => { + it('matches the knot handshake/reset failures we want to retry in-job', () => { + const messages = [ + 'receive-pack: no advertisement (stderr: ssh error: read ECONNRESET)', + 'receive-pack: no advertisement (stderr: ssh error: Timed out while waiting for handshake)', + 'kex_exchange_identification: Connection reset by peer', + 'receive-pack: no advertisement (stderr: ssh error: Connection lost before handshake)', + 'receive-pack: stdin closed before pack finished streaming', + 'receive-pack exited 128 (stderr: empty)', + 'receive-pack: empty report-status (stderr: empty)', + ] + for (const m of messages) { + expect(isTransientTransport(new WireError(m))).toBe(true) + } + }) + + it('never retries a protocol verdict (RemoteRejectedError)', () => { + for (const reason of ['stale-old-sha', 'repo-gone', 'auth-rejected', 'too-big'] as const) { + expect(isTransientTransport(new RemoteRejectedError('nope', reason))).toBe(false) + } + }) + + it('does not retry unrelated wire errors or plain errors', () => { + expect(isTransientTransport(new WireError('knot receive-pack does not advertise report-status'))).toBe(false) + expect(isTransientTransport(new Error('read ECONNRESET'))).toBe(false) + expect(isTransientTransport('read ECONNRESET')).toBe(false) + }) +}) + +describe('withTransportRetry', () => { + it('retries a transient transport failure and eventually succeeds', async () => { + const failures = [ + new WireError('ssh error: read ECONNRESET'), + new WireError('ssh error: Timed out while waiting for handshake'), + ] + let calls = 0 + const op = vi.fn<() => Promise>(async () => { + const err = failures[calls++] + if (err) throw err + return 'ok' + }) + const result = await withTransportRetry(op, 0) + expect(result).toBe('ok') + expect(op).toHaveBeenCalledTimes(3) + }) + + it('re-throws a protocol verdict immediately without retrying', async () => { + const op = vi.fn<() => Promise>().mockRejectedValue(new RemoteRejectedError('lost', 'stale-old-sha')) + await expect(withTransportRetry(op, 0)).rejects.toBeInstanceOf(RemoteRejectedError) + expect(op).toHaveBeenCalledTimes(1) + }) + + it('gives up after the retry budget on persistent transport failure', async () => { + const op = vi.fn<() => Promise>().mockRejectedValue(new WireError('ssh error: read ECONNRESET')) + await expect(withTransportRetry(op, 0)).rejects.toBeInstanceOf(WireError) + // initial attempt + 3 retries + expect(op).toHaveBeenCalledTimes(4) + }) +}) -- 2.51.2