Cover batched pack stream interruption cleanup

5d1679a→main· Soph·3mo ago·2 files·+78 added/-1 removed

Sessions

f5c817f253a9View transcript

Changes

2

• docs

  • Mrewrite-issue-list.md+1

• internal/strategy/bootstrap

  • Mbootstrap_test.go+77/-1

158 unmodified lines

159
160
161
162
163
164
165

158 unmodified lines

- Ownership of stream lifecycle is clearer than on `main`, and the rewrite now has direct tests for key pack-stream close behavior on success and error paths.
- Direct strategy-level error-path tests now verify that relay bootstrap and incremental paths close source pack streams when pushes fail.
- Batched integration coverage now also exercises a failed checkpoint pack push followed by a resume-from-temp-ref retry.
- Batched bootstrap strategy tests now also cover an actual mid-read checkpoint stream interruption where the target pusher returns the read error without closing the stream.
- Lower-level `gitproto.PushPack` rejection paths now also close the provided pack stream instead of leaking it on preflight command errors.
- `gitproto.PushPack` now also has direct closure coverage for cancellation, server-side receive-pack errors, and success.
- `gitproto` fetch tests now verify response-body closure symmetry for both v1 and v2 decode-failure paths.

Mdocs/rewrite-issue-list.md+1

415 unmodified lines

416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
160 unmodified lines

604
605
606
585
607
608
609
610
6 unmodified lines

617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676

415 unmodified lines

return nil
}

type interruptedReadCloser struct {
    first  []byte
    err    error
    stage  int
    closed bool
}

func (r *interruptedReadCloser) Read(p []byte) (int, error) {
    switch r.stage {
    case 0:
        r.stage = 1
        return copy(p, r.first), nil
    default:
        return 0, r.err
    }
}

func (r *interruptedReadCloser) Close() error {
    r.closed = true
    return nil
}

func (f fakeBootstrapPusher) PushPack(ctx context.Context, cmds []gitproto.PushCommand, pack io.ReadCloser) error {
    return f.pushPack(ctx, cmds, pack)
}
415 unmodified lines

},
    }
    TargetRefs:   map[plumbing.ReferenceName]plumbing.Hash{},
    BatchMaxPack: 1,
    BatchMaxPack: 10,
    }, "empty target")
    if err == nil || !strings.Contains(err.Error(), "push bootstrap batch") {
        t.Fatalf("unexpected error: %v", err)
    }
415 unmodified lines

}
}

func TestExecuteBatchedClosesCheckpointPackOnReadInterruption(t *testing.T) {
    mainRef := plumbing.NewBranchReferenceName("main")
    hashes := makeLinearCommitChain(t, 1)
    pack := &interruptedReadCloser{first: []byte("PACK"), err: io.ErrUnexpectedEOF}
    packFetches := 0

_, err := Execute(context.Background(), Params{
        SourceService: fakeBootstrapSource{
            fetchCommitGraph: func(_ context.Context, store storer.Storer, _ *gitproto.Conn, _ gitproto.DesiredRef) error {
                writeLinearCommitChain(t, store, 1)
                return nil
            },
            fetchPack: func(_ context.Context, _ *gitproto.Conn, desired map[plumbing.ReferenceName]gitproto.DesiredRef, _ map[plumbing.ReferenceName]plumbing.Hash) (io.ReadCloser, error) {
                packFetches++
                for _, ref := range desired {
                    if ref.SourceHash == hashes[len(hashes)-1] {
                        if packFetches == 1 {
                            return io.NopCloser(bytes.NewReader(nil)), nil
                        }
                        return pack, nil
                    }
                }
                return io.NopCloser(bytes.NewReader(nil)), nil
            },
        },
        TargetPusher: fakeBootstrapPusher{
            pushPack: func(_ context.Context, _ []gitproto.PushCommand, pack io.ReadCloser) error {
                _, err := io.Copy(io.Discard, pack)
                return err
            },
        },
        DesiredRefs: map[plumbing.ReferenceName]planner.DesiredRef{
            mainRef: {
                SourceRef:  mainRef,
                TargetRef:  mainRef,
                SourceHash: hashes[len(hashes)-1],
                Kind:       planner.RefKindBranch,
                Label:      "main",
            },
        },
        TargetRefs:   map[plumbing.ReferenceName]plumbing.Hash{},
        BatchMaxPack: 10,
    }, "empty target")
    if !errors.Is(err, io.ErrUnexpectedEOF) {
        t.Fatalf("expected interrupted read error, got %v", err)
    }
    if packFetches < 2 {
        t.Fatalf("expected separate probe and execution fetches, got %d", packFetches)
    }
    if !pack.closed {
        t.Fatal("expected strategy to close checkpoint pack after read interruption")
    }
}

func TestExecuteRequiresTargetPusherBeforeFetch(t *testing.T) {
    mainRef := plumbing.NewBranchReferenceName("main")
    mainHash := plumbing.NewHash("bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb")