# gitproto: spool materialized push body to avoid mid-stream stall

`ca362ee`→[main](/content/gh/entireio/git-sync/commits/main/index.html)·  Soph·1mo ago·3 files·+184 added/-14 removed

Buffer the receive-pack body (update-request header + pack) to a temp file in the materialized push path so the POST goes out in one continuous burst, instead of streaming the body incrementally as the pack encoder produces it.

The cause we're working around: go-git's encoder runs delta selection synchronously before writing any pack bytes, which on big repos can take tens of seconds while the HTTP request body sits idle waiting for the next chunk. CDN edges like Cloudflare's enforce an idle-write timeout on the request body and close the connection on a stall that long — surfacing as "use of closed network connection" mid-upload, with no server response. Spooling collapses "encode" and "write" into a single sequential phase, so once bytes start flowing they don't stop until the body is done. Bootstrap-relay didn't have this problem because source pack bytes flow steadily from the upstream upload-pack response — no internal stall, nothing to engineer around.

Side effect: the spooled body has a known length, so the POST goes out with Content-Length instead of Transfer-Encoding: chunked (matching upstream git's smart-HTTP transport behaviour), and req.GetBody lets Go's transport replay the body on transient connection failures. These are nice-to-haves; the stall-avoidance is the actual fix.

Scoped to materialized only. The materialized strategy already requires the full source object closure to be local before encoding begins, so spooling on upload doesn't change its fundamental shape. Relay paths (PushPack) keep streaming source bytes through to target with chunked encoding, and preserve the "streaming proxy" property git-sync is built around.

## Changes

3

- internal/gitproto
 
  - Mpush.go+47/-14

- Mpush_test.go+64

- Msmarthttp.go+73

```
153 unmodified lines

154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
39 unmodified lines

214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
13 unmodified lines

255
256
257
221
222
223
224
225
258
259
260
261
262
263
227
228
264
265
230
231
232
233
234
235
266
267
268
269
270
239
271
272
273
274
275

153 unmodified lines

if packData != nil {
    body = io.MultiReader(body, packData)
}
return postReceivePack(ctx, conn, req, body, verbose, onRejection)

// postReceivePack POSTs an already-built receive-pack request body and
// decodes the response. Split from sendReceivePack so the materialized
// push path can construct a spooled body (header + pack in one temp
// file) and reuse the response handling.
func postReceivePack(
    ctx context.Context,
    conn Conn,
    req *packp.UpdateRequests,
    body io.Reader,
    verbose bool,
    onRejection func(plumbing.ReferenceName, string),
) error {
    reader, err := PostRPCStreamBody(ctx, conn, transport.ReceivePackService, body, false, "receive-pack push")
    if err != nil {
        return fmt.Errorf("target receive-pack: %w", err)
}
39 unmodified lines

// PushObjects pushes locally-materialized objects to the target.
//
// The receive-pack body (update-request header + pack) is written to a
// temp file before the POST so the upload goes out in one continuous
// burst. go-git's encoder runs delta selection synchronously before
// writing any pack bytes, which on big repos stalls the request body
// for tens of seconds — long enough for CDN edges like Cloudflare's to
// hit their idle-write timeout and close the connection mid-upload.
// Spooling collapses encoding and writing into one phase from the
// network's point of view, so the body bytes stream out without gaps.
//
// As a side benefit the spooled body carries a known length, so the
// POST sends Content-Length instead of Transfer-Encoding: chunked
// (matching upstream git's smart-HTTP transport), and req.GetBody lets
// Go's transport retry transient connection failures.
//
// The materialized strategy already requires the full source object
// closure to be local before encoding begins, so a temp file on upload
// doesn't change its fundamental shape. Relay paths (PushPack) keep
// streaming source bytes through to target with chunked encoding —
// source pack data flows steadily, there's no stall to engineer
// around, and the "streaming proxy" property git-sync is built around
// is preserved.
func PushObjects(
    ctx context.Context,
    conn Conn,
    adv *packp.AdvRefs,
    cmds []PushCommand,
    store storage.Storage,
    verbose bool,
    onRejection func(plumbing.ReferenceName, string),
) error {
    
    useRefDeltas := !adv.Capabilities.Supports(capability.OFSDelta)
    pr, pw := io.Pipe()
    done := make(chan error, 1)

go func() {
        enc := packfile.NewEncoder(pw, store, useRefDeltas)
        spooled, cleanup, err := NewSpooledBody(func(w io.Writer) error {
            if err := req.Encode(w); err != nil {
                return fmt.Errorf("encode update-request: %w", err)
            }
            enc := packfile.NewEncoder(w, store, useRefDeltas)
            if _, err := enc.Encode(hashes, 10); err != nil {
                done <- pw.CloseWithError(fmt.Errorf("encode packfile: %w", err))
                return
            }
            done <- pw.Close()
        })

err = sendReceivePack(ctx, conn, req, pr, verbose, onRejection)
        _ = pr.Close()
        encodeErr := <-done
        return nil
    })

if err != nil {
        return err
    }
    return encodeErr
    defer cleanup()
    return postReceivePack(ctx, conn, req, spooled, verbose, onRejection)
}

// PushPack pushes a pack stream (relay) to the target.
```

Minternal/gitproto/push.go+47/-14

```

14 unmodified lines

15
16
17
18
19
20
21
206 unmodified lines

228
229
230
231
232
233
234
235
236
237
238
43 unmodified lines

282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345

14 unmodified lines

"github.com/go-git/go-git/v6/plumbing/protocol/capability"
	"github.com/go-git/go-git/v6/plumbing/protocol/packp"
	"github.com/go-git/go-git/v6/plumbing/transport"
	"github.com/go-git/go-git/v6/storage/memory"
	"github.com/stretchr/testify/require"
}

// TestPushPackStartsHTTPBeforePackFullyRead asserts that PushPack — the
// relay path — keeps streaming source pack bytes through to the target
// with chunked encoding. The "streaming proxy" property is the whole
// point of relay; spooling would erase it. Materialized push is the
// path that buffers (see TestPushObjectsBuffersBody).
func TestPushPackStartsHTTPBeforePackFullyRead(t *testing.T) {
    started := make(chan struct{}, 1)
    release := make(chan struct{})
43 unmodified lines

// TestPushObjectsBuffersBody asserts the materialized push path
// (PushObjects) sends a non-chunked request with an explicit
// Content-Length, by spooling the receive-pack body to a temp file
// before the POST. This works around servers (e.g. Cloudflare's git
// frontend) that close the connection on chunked receive-pack uploads.
func TestPushObjectsBuffersBody(t *testing.T) {
    type observation struct {
        transferEncoding []string
        contentLength    int64
        bodyLen          int64
    }
    observed := make(chan observation, 1)
    
    srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        n, err := io.Copy(io.Discard, r.Body)
        if err != nil {
            t.Logf("drain request body: %v", err)
        }
        _ = r.Body.Close()
        observed <- observation{
            transferEncoding: r.TransferEncoding,
            contentLength:    r.ContentLength,
            bodyLen:          n,
        }
        w.WriteHeader(http.StatusOK)
    }))
    defer srv.Close()

conn := connForServer(t, srv)
    adv := &packp.AdvRefs{}
    adv.Capabilities.Set(capability.OFSDelta)

err := PushObjects(context.Background(), conn, adv, []PushCommand{[]{
        Name: "refs/heads/main",
        New:  plumbing.NewHash("aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"),
} }, memory.NewStorage(), nil, false, nil)
    if err != nil {
        t.Fatalf("PushObjects: %v", err)
    }

var obs observation
    select {
    case obs = <-observed:
    case <-time.After(2 * time.Second):
        t.Fatal("server did not receive request")
    }

if len(obs.transferEncoding) != 0 {
        t.Errorf("Transfer-Encoding = %v, want empty (no chunked)", obs.transferEncoding)
    }
    if obs.contentLength <= 0 {
        t.Errorf("Content-Length = %d, want > 0", obs.contentLength)
    }
    if obs.bodyLen != obs.contentLength {
        t.Errorf("body length %d != Content-Length %d", obs.bodyLen, obs.contentLength)
    }
}

func TestBuildUpdateRequest(t *testing.T) {
    adv := &packp.AdvRefs{}
    adv.Capabilities.Set(capability.ReportStatus)
}

Minternal/gitproto/push_test.go+64

```

377 unmodified lines

378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
18 unmodified lines

427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487

377 unmodified lines

// PostRPCStreamBody sends a POST to the given service using a streaming request body.
// Caller must close the returned ReadCloser.
//
// The body is sent as-is. Streaming bodies (io.MultiReader, io.PipeReader)
// produce a chunked request — that's the right shape for relay paths,
// where source pack bytes flow steadily from source through to target.
// Callers whose body would otherwise stall mid-stream (e.g. the
// materialized push path, where the encoder's delta-selection phase
// produces no bytes for tens of seconds) spool the full payload first
// and pass a *SpooledBody; PostRPCStreamBody sets req.ContentLength and
// req.GetBody from its fields so the upload goes out in one continuous
// burst and Go's transport can auto-retry transient connection failures.
func (c *HTTPConn) PostRPCStreamBody(ctx context.Context, service string, body io.Reader, v2 bool, phase string) (io.ReadCloser, error) {
    reqURL := fmt.Sprintf("%s/%s", c.EndpointURL.String(), service)
    ctx = withHTTPTrace(ctx, "POST "+service)

req, err := http.NewRequestWithContext(ctx, http.MethodPost, reqURL, body)
    if err != nil {
        return nil, fmt.Errorf("create RPC request: %w", err)
    }
    if spooled, ok := body.(*SpooledBody); ok {
        req.ContentLength = spooled.size
        path := spooled.path
        req.GetBody = func() (io.ReadCloser, error) {
            return os.Open(path)
        }
    }
    req.Header.Set("Content-Type", fmt.Sprintf("application/x-%s-request", service))
    req.Header.Set("Accept", fmt.Sprintf("application/x-%s-result", service))
    req.Header.Set("User-Agent", capability.DefaultAgent())
    
    return res.Body, nil
}

// SpooledBody is a temp-file-backed request body with a known length.
// PostRPCStreamBody type-asserts on it and sets req.ContentLength /
// req.GetBody so the request body goes out in one continuous burst
// (no mid-stream idle gap) and is replayable on transient connection
// failures.
//
// Used by the materialized push path, where the full payload has to be
// produced locally before any bytes can flow — go-git's encoder runs
// delta selection synchronously before writing the pack, which on big
// repos stalls the request body for tens of seconds. CDN edges like
// Cloudflare's enforce an idle-write timeout on request bodies and
// close the connection on a stall that long; spooling first eliminates
// the gap entirely. The closure walk already requires a local store,
// so the temp file doesn't change the strategy's fundamental shape.
//
// Relay paths intentionally don't use this — source pack bytes flow
// steadily from the upstream upload-pack response, so there's no stall
to engineer around. Keeping relay streaming is the whole point of
// the relay shape.
type SpooledBody struct {
    io.ReadCloser
    path string
    size int64
}

// NewSpooledBody creates a temp file, writes write(f) into it, rewinds
// it, and returns a SpooledBody plus a cleanup that removes the temp
// file. The cleanup is always non-nil; call it (typically via defer)
// regardless of error.
func NewSpooledBody(write func(io.Writer) error) (*SpooledBody, func(), error) {
    f, err := os.CreateTemp("", "git-sync-rpc-*")
    if err != nil {
        return nil, func() {}, fmt.Errorf("create temp file: %w", err)
    }
    path := f.Name()
    cleanup := func() {
        _ = f.Close()
        _ = os.Remove(path)
    }
    if err := write(f); err != nil {
        cleanup()
        return nil, func() {}, err
    }
    size, err := f.Seek(0, io.SeekCurrent)
    if err != nil {
        cleanup()
        return nil, func() {}, fmt.Errorf("get spooled body size: %w", err)
    }
    if _, err := f.Seek(0, io.SeekStart); err != nil {
        cleanup()
        return nil, func() {}, fmt.Errorf("rewind spooled body: %w", err)
    }
    return &SpooledBody{ReadCloser: f, path: path, size: size}, cleanup, nil
}

// ApplyAuth applies the given auth method to an HTTP request. Errors from
// the Authorizer (e.g. transient signing failures) are surfaced as request
// failures by leaving the Authorization header unset; the upstream server

... (and so on) ... 
 
// TestBuildUpdateRequest asserts the functionality of building an update request.
func TestBuildUpdateRequest(t *testing.T) {
    adv := &packp.AdvRefs{}
    adv.Capabilities.Set(capability.ReportStatus)
}