mirror your GitHub repos to tangled.org automatically
synchub.to
mirror
sync
tangled
github
git
fix: retry transient knot disconnects within the sync job
Daniel Roe
(Jul 10, 2026, 1:01 PM +0200)
060a55e6
6575d143
+131
-4
+27
server/utils/git-wire/errors.ts
+27
server/utils/git-wire/errors.ts
···
34
34
}
35
35
36
36
/**
37
+
* True for failures that are worth retrying immediately within the same job: a
38
+
* flaky knot SSH endpoint that resets or times out the connection at the
39
+
* handshake, or drops the channel mid-stream. These clear within seconds, so a
40
+
* short in-job retry usually succeeds where waiting out the queue's minutes-long
41
+
* backoff would leave the mirror stale.
42
+
*
43
+
* A `RemoteRejectedError` is never transient-transport: it's a protocol-level
44
+
* verdict (CAS loss, repo gone, auth refused, pack too big) that a reconnect
45
+
* won't change, so those fall through to the caller's own handling.
46
+
*/
47
+
export function isTransientTransport(err: unknown): boolean {
48
+
if (!(err instanceof WireError) || err instanceof RemoteRejectedError) return false
49
+
const lc = err.message.toLowerCase()
50
+
return (
51
+
lc.includes('econnreset')
52
+
|| lc.includes('connection reset')
53
+
|| lc.includes('timed out')
54
+
|| lc.includes('handshake')
55
+
|| lc.includes('no advertisement')
56
+
|| lc.includes('connection lost')
57
+
|| lc.includes('stdin closed')
58
+
|| lc.includes('receive-pack exited')
59
+
|| lc.includes('empty report-status')
60
+
)
61
+
}
62
+
63
+
/**
37
64
* Classify ssh / sshd / knot stderr (the child process's stderr band, since
38
65
* we deliberately do not request side-band multiplexing). Returns null when
39
66
* nothing matches so the caller can fall back to a generic transient error.
+41
-4
server/utils/splice.ts
+41
-4
server/utils/splice.ts
···
1
+
import { isTransientTransport } from './git-wire/errors'
1
2
import {
2
3
type ReceivePackFactory,
3
4
ReceivePackSession,
···
13
14
/** Cap haves so a repo with thousands of refs can't bloat the negotiation. */
14
15
const MAX_HAVES = 256
15
16
17
+
/**
18
+
* In-job retry for the flaky knot SSH endpoint. A burst of connection resets at
19
+
* the handshake typically clears within a second or two, so retrying a few
20
+
* times inside the job lands the push now rather than after the queue's
21
+
* minutes-long backoff (or, worse, after the attempt ceiling gives up and
22
+
* leaves the mirror stale until the next push). Only transient transport
23
+
* failures are retried; protocol verdicts re-throw immediately.
24
+
*/
25
+
const TRANSPORT_RETRIES = 3
26
+
const TRANSPORT_RETRY_DELAY_MS = 1_500
27
+
28
+
export async function withTransportRetry<T>(op: () => Promise<T>, delayMs = TRANSPORT_RETRY_DELAY_MS): Promise<T> {
29
+
let lastErr: unknown
30
+
for (let attempt = 0; attempt <= TRANSPORT_RETRIES; attempt++) {
31
+
try {
32
+
// eslint-disable-next-line no-await-in-loop -- sequential retries by design
33
+
return await op()
34
+
}
35
+
catch (err) {
36
+
if (!isTransientTransport(err)) throw err
37
+
lastErr = err
38
+
if (attempt < TRANSPORT_RETRIES) {
39
+
// eslint-disable-next-line no-await-in-loop -- deliberate backoff between retries
40
+
await new Promise(resolve => setTimeout(resolve, delayMs))
41
+
}
42
+
}
43
+
}
44
+
throw lastErr
45
+
}
46
+
16
47
function maxPackBytes(): number {
17
48
const raw = process.env.NUXT_MAX_PACK_BYTES
18
49
if (!raw) return DEFAULT_MAX_PACK_BYTES
···
60
91
* keeps the knot's advertised tip as the authoritative compare-and-swap base.
61
92
*/
62
93
export async function splicePush(params: SplicePushParams): Promise<SplicePushResult> {
63
-
const factory = await sshFactory(params.installationId, params.knot, params.repoDid)
64
-
return runSplice(factory, params)
94
+
return withTransportRetry(async () => {
95
+
// Fresh factory (and therefore fresh SSH connection) per attempt: a reset
96
+
// connection can't be reused.
97
+
const factory = await sshFactory(params.installationId, params.knot, params.repoDid)
98
+
return runSplice(factory, params)
99
+
})
65
100
}
66
101
67
102
/** The fetch + push exchange over an open session. Split out for the wire test. */
···
118
153
repoDid: string
119
154
ref: string
120
155
}): Promise<SpliceDeleteResult> {
121
-
const factory = await sshFactory(params.installationId, params.knot, params.repoDid)
122
-
return runSpliceDelete(factory, params.ref)
156
+
return withTransportRetry(async () => {
157
+
const factory = await sshFactory(params.installationId, params.knot, params.repoDid)
158
+
return runSpliceDelete(factory, params.ref)
159
+
})
123
160
}
124
161
125
162
/** The delete exchange over an open session. Split out for the wire test. */
+63
test/unit/transient-transport.spec.ts
+63
test/unit/transient-transport.spec.ts
···
1
+
import { describe, expect, it, vi } from 'vitest'
2
+
import { isTransientTransport, RemoteRejectedError, WireError } from '../../server/utils/git-wire/errors'
3
+
import { withTransportRetry } from '../../server/utils/splice'
4
+
5
+
describe('isTransientTransport', () => {
6
+
it('matches the knot handshake/reset failures we want to retry in-job', () => {
7
+
const messages = [
8
+
'receive-pack: no advertisement (stderr: ssh error: read ECONNRESET)',
9
+
'receive-pack: no advertisement (stderr: ssh error: Timed out while waiting for handshake)',
10
+
'kex_exchange_identification: Connection reset by peer',
11
+
'receive-pack: no advertisement (stderr: ssh error: Connection lost before handshake)',
12
+
'receive-pack: stdin closed before pack finished streaming',
13
+
'receive-pack exited 128 (stderr: empty)',
14
+
'receive-pack: empty report-status (stderr: empty)',
15
+
]
16
+
for (const m of messages) {
17
+
expect(isTransientTransport(new WireError(m))).toBe(true)
18
+
}
19
+
})
20
+
21
+
it('never retries a protocol verdict (RemoteRejectedError)', () => {
22
+
for (const reason of ['stale-old-sha', 'repo-gone', 'auth-rejected', 'too-big'] as const) {
23
+
expect(isTransientTransport(new RemoteRejectedError('nope', reason))).toBe(false)
24
+
}
25
+
})
26
+
27
+
it('does not retry unrelated wire errors or plain errors', () => {
28
+
expect(isTransientTransport(new WireError('knot receive-pack does not advertise report-status'))).toBe(false)
29
+
expect(isTransientTransport(new Error('read ECONNRESET'))).toBe(false)
30
+
expect(isTransientTransport('read ECONNRESET')).toBe(false)
31
+
})
32
+
})
33
+
34
+
describe('withTransportRetry', () => {
35
+
it('retries a transient transport failure and eventually succeeds', async () => {
36
+
const failures = [
37
+
new WireError('ssh error: read ECONNRESET'),
38
+
new WireError('ssh error: Timed out while waiting for handshake'),
39
+
]
40
+
let calls = 0
41
+
const op = vi.fn<() => Promise<string>>(async () => {
42
+
const err = failures[calls++]
43
+
if (err) throw err
44
+
return 'ok'
45
+
})
46
+
const result = await withTransportRetry(op, 0)
47
+
expect(result).toBe('ok')
48
+
expect(op).toHaveBeenCalledTimes(3)
49
+
})
50
+
51
+
it('re-throws a protocol verdict immediately without retrying', async () => {
52
+
const op = vi.fn<() => Promise<never>>().mockRejectedValue(new RemoteRejectedError('lost', 'stale-old-sha'))
53
+
await expect(withTransportRetry(op, 0)).rejects.toBeInstanceOf(RemoteRejectedError)
54
+
expect(op).toHaveBeenCalledTimes(1)
55
+
})
56
+
57
+
it('gives up after the retry budget on persistent transport failure', async () => {
58
+
const op = vi.fn<() => Promise<never>>().mockRejectedValue(new WireError('ssh error: read ECONNRESET'))
59
+
await expect(withTransportRetry(op, 0)).rejects.toBeInstanceOf(WireError)
60
+
// initial attempt + 3 retries
61
+
expect(op).toHaveBeenCalledTimes(4)
62
+
})
63
+
})