| … | … | @@ -160,9 +160,9 @@ async function handleConnect(service, serverKey, repoName, httpsUrl, stdout, std |
| 160 | 160 | process.stdin.removeAllListeners('data') |
| 161 | 161 | process.stdin.removeAllListeners('end') |
| 162 | 162 | |
| 163 | // SecretStream has built-in timeout that auto-refreshes on incoming data. | |
| 164 | // If no data arrives within 45s, it calls destroy() which fires 'close'. | |
| 165 | conn.setTimeout(45000) | |
| 163 | // No connection timeout — large repos (1+ GB) can transfer for minutes. | |
| 164 | // Backpressure keeps data flowing; stale connections are detected by | |
| 165 | // the stream's error/close events. | |
| 166 | 166 | |
| 167 | 167 | // Tell git we're ready FIRST (before any data forwarding) |
| 168 | 168 | // This ensures '\n' precedes ref advertisement on stdout |
| … | … | @@ -182,9 +182,18 @@ async function handleConnect(service, serverKey, repoName, httpsUrl, stdout, std |
| 182 | 182 | // streamx pipe() unconditionally calls dest.end() when source ends, |
| 183 | 183 | // and SecretStream._final() closes both directions of the noise stream. |
| 184 | 184 | // This kills push (server can't send status after client sends pack). |
| 185 | // | |
| 186 | // Backpressure: when write() returns false, pause the source. | |
| 187 | // Resume on 'drain'. Without this, large repos overflow the send | |
| 188 | // buffer and cause "unexpected disconnect" / "bad pack header". | |
| 185 | 189 | let stdinEnded = false |
| 186 | 190 | process.stdin.on('data', (chunk) => { |
| 187 | conn.write(chunk) | |
| 191 | if (!conn.write(chunk)) { | |
| 192 | process.stdin.pause() | |
| 193 | } | |
| 194 | }) | |
| 195 | conn.on('drain', () => { | |
| 196 | if (!stdinEnded) process.stdin.resume() | |
| 188 | 197 | }) |
| 189 | 198 | |
| 190 | 199 | process.stdin.on('end', () => { |
| … | … | @@ -196,8 +205,11 @@ async function handleConnect(service, serverKey, repoName, httpsUrl, stdout, std |
| 196 | 205 | }) |
| 197 | 206 | |
| 198 | 207 | conn.on('data', (chunk) => { |
| 199 | stdout.write(chunk) | |
| 208 | if (!stdout.write(chunk)) { | |
| 209 | conn.pause() | |
| 210 | } | |
| 200 | 211 | }) |
| 212 | stdout.on('drain', () => conn.resume()) | |
| 201 | 213 | |
| 202 | 214 | conn.on('close', () => cleanup(0)) |
| 203 | 215 |
| … | … | @@ -54,8 +54,9 @@ export class GitP2PHandler { |
| 54 | 54 | const connId = peerInfo.publicKey.toString('hex').slice(0, 8) |
| 55 | 55 | console.log(`[GitP2P] Connection from ${connId}`) |
| 56 | 56 | |
| 57 | // Auto-destroy stale connections — SecretStream refreshes on incoming data | |
| 58 | if (typeof conn.setTimeout === 'function') conn.setTimeout(60000) | |
| 57 | // No connection-level timeout — git transfers can run for minutes on large | |
| 58 | // repos. Backpressure keeps data flowing; connection drops are detected by | |
| 59 | // the stream's built-in error/close events. | |
| 59 | 60 | |
| 60 | 61 | let headerReceived = false |
| 61 | 62 | let buffer = Buffer.alloc(0) |
| … | … | @@ -789,10 +790,17 @@ export class GitP2PHandler { |
| 789 | 790 | // Manual bidirectional proxy instead of .pipe(). |
| 790 | 791 | // Explicit resume() after adding data listener — removing the |
| 791 | 792 | // headerHandler may leave the noise stream paused in streamx. |
| 793 | // | |
| 794 | // Backpressure: when write() returns false, pause the source. | |
| 795 | // Resume on 'drain'. Without this, large repos (1+ GB) overflow | |
| 796 | // the send buffer and cause "unexpected disconnect" / "bad pack header". | |
| 792 | 797 | conn.on('data', (chunk) => { |
| 793 | 798 | console.log(`[GitP2P] conn->git: ${chunk.length} bytes`) |
| 794 | gitProcess.stdin.write(chunk) | |
| 799 | if (!gitProcess.stdin.write(chunk)) { | |
| 800 | conn.pause() | |
| 801 | } | |
| 795 | 802 | }) |
| 803 | gitProcess.stdin.on('drain', () => conn.resume()) | |
| 796 | 804 | |
| 797 | 805 | let firstChunk = true |
| 798 | 806 | gitProcess.stdout.on('data', (chunk) => { |
| … | … | @@ -803,8 +811,11 @@ export class GitP2PHandler { |
| 803 | 811 | console.log(`[GitP2P] REFS: ${preview}`) |
| 804 | 812 | firstChunk = false |
| 805 | 813 | } |
| 806 | conn.write(chunk) | |
| 814 | if (!conn.write(chunk)) { | |
| 815 | gitProcess.stdout.pause() | |
| 816 | } | |
| 807 | 817 | }) |
| 818 | conn.on('drain', () => gitProcess.stdout.resume()) | |
| 808 | 819 | |
| 809 | 820 | // Ensure conn is in flowing mode after headerHandler was removed |
| 810 | 821 | if (typeof conn.resume === 'function') conn.resume() |
| … | … | @@ -37,11 +37,18 @@ export function createGitPipe (service, repoPath, dataEntry, opts = {}) { |
| 37 | 37 | const stderrChunks = [] |
| 38 | 38 | |
| 39 | 39 | // -- git stdout → channel.message.send() -- |
| 40 | // NOTE: Protomux message.send() does not return a backpressure signal. | |
| 41 | // The underlying mux stream handles buffering. If the remote can't keep | |
| 42 | // up, data accumulates in the mux write buffer. For very large transfers, | |
| 43 | // the direct punch:// path (git-p2p-handler) should be preferred over | |
| 44 | // this Protomux path. | |
| 40 | 45 | child.stdout.on('data', (chunk) => { |
| 41 | 46 | dataEntry.message.send(chunk) |
| 42 | 47 | }) |
| 43 | 48 | |
| 44 | 49 | // -- channel data → git stdin (wire up onmessage) -- |
| 50 | // Protomux delivers messages synchronously — we can't pause delivery, | |
| 51 | // but we can let Node/Bare buffer in git's stdin pipe. | |
| 45 | 52 | dataEntry.message.onmessage = (data) => { |
| 46 | 53 | if (child.stdin.writable !== false) { |
| 47 | 54 | child.stdin.write(data) |
| … | … | @@ -583,7 +583,8 @@ class PearGitServer { |
| 583 | 583 | console.log(`[Metadata] Created new repo from peer: ${remote.id}`) |
| 584 | 584 | |
| 585 | 585 | // Register it |
| 586 | await this.repoRegistry.register(remote.id, { | |
| 586 | await this.repoRegistry.register({ | |
| 587 | name: remote.id, | |
| 587 | 588 | description: remote.description, |
| 588 | 589 | visibility: remote.visibility |
| 589 | 590 | }) |