mirror your GitHub repos to tangled.org automatically
synchub.to
mirror
sync
tangled
github
git
fix: surface mid-push knot disconnects as transient errors
Daniel Roe
(Jul 10, 2026, 12:41 PM +0200)
fd8e4181
12576164
+221
-6
+40
-4
server/utils/git-wire/receive-pack.ts
+40
-4
server/utils/git-wire/receive-pack.ts
···
95
95
stdin.pipe(channel)
96
96
channel.pipe(stdout)
97
97
channel.stderr.on('data', appendStderr)
98
+
// A channel-level error (peer reset, transport dying mid-stream) would
99
+
// otherwise be lost: the write side is a PassThrough whose 'error' we
100
+
// swallow, and a half-sent pack reaches the knot as a corrupt stream
101
+
// it reports back as an `unpack` failure. Fold it into the stderr band
102
+
// so push() can tell a truncated push apart from a clean one.
103
+
channel.on('error', (err: Error) => { if (!killed) connError = err })
98
104
channel.on('exit', code => settle(typeof code === 'number' ? code : null))
99
105
channel.on('close', () => { client.end(); stdout.end() })
100
106
})
···
128
134
username: 'git',
129
135
privateKey: target.privateKey,
130
136
readyTimeout: 8_000,
137
+
// The channel sits idle while we fetch the thin pack from GitHub (we need
138
+
// the knot's advertised tips as haves before we can ask for the delta).
139
+
// Keepalives stop the knot or an intermediary from dropping that idle
140
+
// channel out from under us, which otherwise surfaces as a truncated pack
141
+
// (`unpack` failure) once we resume streaming.
142
+
keepaliveInterval: 5_000,
143
+
keepaliveCountMax: 6,
131
144
hostVerifier: () => true,
132
145
})
133
146
···
208
221
else this.proc.stdin.end()
209
222
210
223
const report = await this.reader.readUntilFlush()
224
+
// If the transport died while we were streaming the pack, the knot saw a
225
+
// truncated stream and reports it as an `unpack` failure (e.g. "pack
226
+
// signature mismatch"). That reads like a terminal protocol error but is
227
+
// really a transient network fault, and a naive retry replays the same
228
+
// broken pipe. Check the channel/exit state first and surface it as a
229
+
// plain transient WireError so the queue retries with a fresh session.
230
+
const exitCode = await this.proc.done.catch(() => null)
231
+
const transport = classifySshStderr(this.proc.stderr())
232
+
if (transport) throw transport
233
+
if (exitCode !== null && exitCode !== 0) {
234
+
throw new WireError(`receive-pack exited ${exitCode} (stderr: ${this.proc.stderr().trim() || 'empty'})`)
235
+
}
211
236
parseReportStatus(report ?? [], updates, this.proc.stderr())
212
237
}
213
238
finally {
···
272
297
async function pipePack(stdin: NodeJS.WritableStream, packStream: AsyncIterable<Buffer>): Promise<void> {
273
298
const src = Readable.from(packStream)
274
299
await new Promise<void>((resolve, reject) => {
275
-
src.on('error', reject)
276
-
stdin.on('error', reject)
300
+
let settled = false
301
+
const done = (err?: Error) => {
302
+
if (settled) return
303
+
settled = true
304
+
if (err) reject(err)
305
+
else resolve()
306
+
}
307
+
// A `close` before `finish` means the write side went away before the pack
308
+
// was fully flushed (the knot's channel reset mid-stream). Resolving there
309
+
// would hand a truncated pack to report-status parsing; reject so push()
310
+
// treats it as the transient transport failure it is.
311
+
src.on('error', done)
312
+
stdin.on('error', done)
313
+
stdin.on('finish', () => done())
314
+
stdin.on('close', () => done(new WireError('receive-pack: stdin closed before pack finished streaming')))
277
315
src.pipe(stdin, { end: true })
278
-
stdin.on('finish', resolve)
279
-
stdin.on('close', resolve)
280
316
})
281
317
}
+49
-2
test/unit/receive-pack.spec.ts
+49
-2
test/unit/receive-pack.spec.ts
···
1
1
import { execFileSync } from 'node:child_process'
2
+
import { PassThrough } from 'node:stream'
3
+
import { Buffer } from 'node:buffer'
2
4
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
3
-
import { RemoteRejectedError } from '../../server/utils/git-wire/errors'
4
-
import { ReceivePackSession } from '../../server/utils/git-wire/receive-pack'
5
+
import { RemoteRejectedError, WireError } from '../../server/utils/git-wire/errors'
6
+
import { ReceivePackSession, type ReceivePackProcess } from '../../server/utils/git-wire/receive-pack'
7
+
import { encodePktLine, flushPkt } from '../../server/utils/git-wire/pkt-line'
5
8
import { ZERO_SHA } from '../../server/utils/git-wire/refs'
6
9
import { fakeGithubFetch, GitFixture, localReceivePackFactory } from '../utils/git-wire'
7
10
import { fetchPack } from '../../server/utils/git-wire/upload-pack'
···
123
126
}
124
127
}
125
128
await expect(ReceivePackSession.open(stalled, 50)).rejects.toThrow(/end of stream|advertisement/)
129
+
})
130
+
131
+
it('surfaces a mid-push channel death as a transient WireError, not a report parse', async () => {
132
+
// Advertise one ref so open() succeeds, then model the transport dying
133
+
// while the pack streams: stdin closes early and the process exits non-zero
134
+
// with no report-status. The knot in production would report a truncated
135
+
// pack as an `unpack` failure; the session must instead throw a plain
136
+
// (transient, retryable) WireError.
137
+
const factory = () => {
138
+
const stdin = new PassThrough()
139
+
const advertisement = Buffer.concat([
140
+
encodePktLine(`${ZERO_SHA} refs/heads/main\0report-status\n`),
141
+
flushPkt,
142
+
])
143
+
const stdout = new PassThrough()
144
+
stdout.write(advertisement)
145
+
let resolveDone: (code: number | null) => void
146
+
const done = new Promise<number | null>(r => { resolveDone = r })
147
+
// Let stdin drain (so writeAll + pipePack complete their writes), but end
148
+
// stdout without a report-status flush and exit non-zero, as a reset ssh
149
+
// channel would after eating a partial pack.
150
+
stdin.resume()
151
+
stdin.on('finish', () => {
152
+
stdout.end()
153
+
resolveDone(128)
154
+
})
155
+
const proc: ReceivePackProcess = {
156
+
stdin,
157
+
stdout,
158
+
stderr: () => '',
159
+
kill: () => { stdout.end(); resolveDone(128) },
160
+
done,
161
+
}
162
+
return proc
163
+
}
164
+
165
+
async function* pack(): AsyncGenerator<Buffer> {
166
+
yield Buffer.from('PACK')
167
+
yield Buffer.alloc(1024)
168
+
}
169
+
170
+
const session = await ReceivePackSession.open(factory)
171
+
await expect(session.push([{ ref: 'refs/heads/main', old: ZERO_SHA, next: 'a'.repeat(40) }], pack()))
172
+
.rejects.toBeInstanceOf(WireError)
126
173
})
127
174
128
175
it('pushes an annotated tag', async () => {
+88
test/unit/splice-negotiation.spec.ts
+88
test/unit/splice-negotiation.spec.ts
···
1
+
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
2
+
import { fetchPack } from '../../server/utils/git-wire/upload-pack'
3
+
import { fakeGithubFetch, GitFixture, localReceivePackFactory } from '../utils/git-wire'
4
+
import { ReceivePackSession } from '../../server/utils/git-wire/receive-pack'
5
+
import { ZERO_SHA } from '../../server/utils/git-wire/refs'
6
+
7
+
async function drain(gen: AsyncGenerator<Buffer>): Promise<Buffer> {
8
+
const parts: Buffer[] = []
9
+
for await (const c of gen) parts.push(c)
10
+
return Buffer.concat(parts)
11
+
}
12
+
13
+
describe('fetchPack negotiation edge cases', () => {
14
+
let fx: GitFixture
15
+
let realFetch: typeof globalThis.fetch
16
+
17
+
beforeEach(() => {
18
+
fx = new GitFixture()
19
+
realFetch = globalThis.fetch
20
+
})
21
+
22
+
afterEach(() => {
23
+
globalThis.fetch = realFetch
24
+
fx.cleanup()
25
+
})
26
+
27
+
function firstBytes(buf: Buffer, n = 4): string {
28
+
return buf.subarray(0, n).toString('ascii')
29
+
}
30
+
31
+
it('pack begins with the PACK signature when a have is a real common commit', async () => {
32
+
const gh = fx.initBare('gh.git')
33
+
const work = fx.initWork('work')
34
+
const base = fx.commit(work, 'a.txt', 'hello')
35
+
fx.pushTo(work, gh, 'HEAD:refs/heads/main')
36
+
const head = fx.commit(work, 'b.txt', 'world')
37
+
fx.pushTo(work, gh, 'HEAD:refs/heads/main')
38
+
39
+
globalThis.fetch = fakeGithubFetch(new Map([['owner/repo', gh]])) as unknown as typeof globalThis.fetch
40
+
const { pack } = await fetchPack({ repoFullName: 'owner/repo', token: 't', want: head, haves: [base], maxBytes: 1 << 30 })
41
+
expect(firstBytes(await drain(pack))).toBe('PACK')
42
+
})
43
+
44
+
it('pack begins with PACK when a have is unknown to GitHub (diverged knot ref)', async () => {
45
+
const gh = fx.initBare('gh.git')
46
+
const work = fx.initWork('work')
47
+
const base = fx.commit(work, 'a.txt', 'hello')
48
+
fx.pushTo(work, gh, 'HEAD:refs/heads/main')
49
+
const head = fx.commit(work, 'b.txt', 'world')
50
+
fx.pushTo(work, gh, 'HEAD:refs/heads/main')
51
+
52
+
// A commit GitHub has never seen (exists only on a throwaway repo). The
53
+
// knot could advertise such a tip; we must not let it corrupt the stream.
54
+
const other = fx.initWork('other')
55
+
const stranger = fx.commit(other, 'x.txt', 'stranger')
56
+
57
+
globalThis.fetch = fakeGithubFetch(new Map([['owner/repo', gh]])) as unknown as typeof globalThis.fetch
58
+
const { pack } = await fetchPack({ repoFullName: 'owner/repo', token: 't', want: head, haves: [stranger, base], maxBytes: 1 << 30 })
59
+
expect(firstBytes(await drain(pack))).toBe('PACK')
60
+
})
61
+
62
+
it('pushes into a knot that advertises a stranger tip alongside the base', async () => {
63
+
const gh = fx.initBare('gh.git')
64
+
const work = fx.initWork('work')
65
+
const base = fx.commit(work, 'a.txt', 'hello')
66
+
fx.pushTo(work, gh, 'HEAD:refs/heads/main')
67
+
const head = fx.commit(work, 'b.txt', 'world')
68
+
fx.pushTo(work, gh, 'HEAD:refs/heads/main')
69
+
70
+
// Knot has the base on main plus an unrelated tip on another ref, so its
71
+
// advertisement (and therefore our have-set) mixes a real common commit
72
+
// with one GitHub can't resolve.
73
+
const knot = fx.initBare('knot.git')
74
+
fx.pushTo(work, knot, `${base}:refs/heads/main`)
75
+
const other = fx.initWork('other')
76
+
const stranger = fx.commit(other, 'x.txt', 'stranger')
77
+
fx.pushTo(other, knot, `${stranger}:refs/heads/wip`)
78
+
79
+
globalThis.fetch = fakeGithubFetch(new Map([['owner/repo', gh]])) as unknown as typeof globalThis.fetch
80
+
81
+
const session = await ReceivePackSession.open(localReceivePackFactory(knot))
82
+
const haves = [...new Set(session.tips.values())].filter(s => s !== ZERO_SHA)
83
+
const { pack } = await fetchPack({ repoFullName: 'owner/repo', token: 't', want: head, haves, maxBytes: 1 << 30 })
84
+
await session.push([{ ref: 'refs/heads/main', old: base, next: head }], pack)
85
+
86
+
expect(fx.revParse(knot, 'refs/heads/main')).toBe(head)
87
+
})
88
+
})
+44
test/unit/splice-thin-pack.spec.ts
+44
test/unit/splice-thin-pack.spec.ts
···
1
+
import { afterEach, beforeEach, describe, expect, it } from 'vitest'
2
+
import { runSplice } from '../../server/utils/splice'
3
+
import { fakeGithubFetch, GitFixture, localReceivePackFactory } from '../utils/git-wire'
4
+
5
+
describe('runSplice thin-pack negotiation (knot already holds the base)', () => {
6
+
let fx: GitFixture
7
+
let realFetch: typeof globalThis.fetch
8
+
9
+
beforeEach(() => {
10
+
fx = new GitFixture()
11
+
realFetch = globalThis.fetch
12
+
})
13
+
14
+
afterEach(() => {
15
+
globalThis.fetch = realFetch
16
+
fx.cleanup()
17
+
})
18
+
19
+
it('pushes an incremental update where the knot advertises the base commit as a have', async () => {
20
+
// GitHub side: two commits on main (base -> head).
21
+
const gh = fx.initBare('gh.git')
22
+
const work = fx.initWork('work')
23
+
const base = fx.commit(work, 'a.txt', 'hello')
24
+
fx.pushTo(work, gh, 'HEAD:refs/heads/main')
25
+
const head = fx.commit(work, 'b.txt', 'world')
26
+
fx.pushTo(work, gh, 'HEAD:refs/heads/main')
27
+
28
+
// Knot side: already has `base` on main (the state after the first sync).
29
+
const knot = fx.initBare('knot.git')
30
+
fx.pushTo(work, knot, `${base}:refs/heads/main`)
31
+
32
+
globalThis.fetch = fakeGithubFetch(new Map([['owner/repo', gh]])) as unknown as typeof globalThis.fetch
33
+
34
+
const result = await runSplice(localReceivePackFactory(knot), {
35
+
repoFullName: 'owner/repo',
36
+
ref: 'refs/heads/main',
37
+
want: head,
38
+
token: 't',
39
+
})
40
+
41
+
expect(result.status).toBe('synced')
42
+
expect(fx.revParse(knot, 'refs/heads/main')).toBe(head)
43
+
})
44
+
})