Merge pull request #88 from entireio/fix/session-resource-leaks

52aa33b→main·

Soph·4w ago·2 files·+76 added/-13 removed

Release session resources on partial-setup and error paths

Changes

• internal/syncer

• Ameasurement_test.go+46

  • Msyncer.go+30/-13

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46

package syncer

import (
    "runtime"
    "testing"
    "time"
)

// The measurement closure must stop its ticker goroutine when invoked, and be
// safe to invoke more than once — finish() now calls it on every path
// (including error paths) in addition to the happy-path Result builder, so a
// double-call must neither panic (double close) nor change the reported value.
func TestStartMeasurementStopsGoroutineAndIsIdempotent(t *testing.T) {
    before := runtime.NumGoroutine()

done := startMeasurement(true)

m1 := done()
    if !m1.Enabled {
        t.Fatalf("expected an enabled measurement, got %+v", m1)
    }
    m2 := done() // second call (the finish() path) must be a safe no-op
    if m1 != m2 {
        t.Fatalf("measurement changed across calls: %+v vs %+v", m1, m2)
    }

// The ticker goroutine must have exited; poll to avoid races with the
    // scheduler tearing it down.
    deadline := time.Now().Add(2 * time.Second)
    for runtime.NumGoroutine() > before {
        if time.Now().After(deadline) {
            t.Fatalf("measurement goroutine leaked: %d goroutines, baseline %d",
                runtime.NumGoroutine(), before)
        }
        time.Sleep(10 * time.Millisecond)
    }
}

// When disabled, no goroutine is started and the closure is still safe to call.
func TestStartMeasurementDisabledIsInert(t *testing.T) {
    done := startMeasurement(false)
    if m := done(); m.Enabled {
        t.Fatalf("disabled measurement should not be enabled: %+v", m)
    }
    _ = done() // idempotent
}

Ainternal/syncer/measurement_test.go+46

585 unmodified lines

586
587
588
589
590
591
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
58 unmodified lines

663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
37 unmodified lines

716
717
718
719
720
721
722
723
724
6 unmodified lines

731
732
733
715
716
717
718
719
720
721
722
723
724
734
735
736
737
738
739
740
741
742
743
744
23 unmodified lines

768
769
770
771
772
773
774

585 unmodified lines

rejections map[plumbing.ReferenceName]string
}

// finish releases any resources owned by the session — currently the live
// progress ticker. Idempotent and safe to call from defer in callers that
// also produce results in the happy path.
// finish releases the resources owned by the session: the live progress
// ticker, the memory-measurement ticker goroutine, and the source/target
// transports (SSH transports spawn processes). Idempotent and safe to call
// from defer in callers that also produce results in the happy path —
// measurementDone is sync.Once-guarded, so an error path that never built a
// Result still stops its goroutine without disturbing the happy-path value.
func (s *syncSession) finish() {
    if s.progress != nil {
        s.progress.terminate()
    }
    if s.measurementDone != nil {
        _ = s.measurementDone()
    }
    if s.sourceConn != nil {
        _ = s.sourceConn.Close()
    }
58 unmodified lines

stats:           newStats(cfg.ShowStats),
        measurementDone: startMeasurement(cfg.MeasureMemory),
    }
    // startMeasurement spawned a ticker goroutine and the steps below open
    // transports (SSH spawns a process). If we return an error partway through
    // setup the caller has no session to finish(), so release everything here
    // unless we hand the session back.
    success := false
    defer func() {
        if !success {
            s.finish()
        }
    }()
    var warnedSSHStats bool
    warnSSHStats := func(sourceConn, targetConn gitproto.Conn) {
        if warnedSSHStats {
37 unmodified lines

if err != nil {
            return nil, fmt.Errorf("create target transport: %w", err)
        }
        // Hand the conn to the session immediately so the deferred cleanup
        // closes it even if a ref-listing step below fails.
        s.target = &targetSession{conn: targetConn}
        targetConn.SetProgressWriter(&sessionStderr{s: s})
        warnSSHStats(s.sourceConn, targetConn)
        targetAdv, err := gitproto.AdvertisedRefsV1(ctx, targetConn, transport.ReceivePackService)
6 unmodified lines

}
        targetRefMap := gitproto.RefHashMap(targetRefSlice)
        targetFeatures := gitproto.TargetFeaturesFromAdvRefs(targetAdv)
        s.target = &targetSession{
            conn:     targetConn,
            adv:      targetAdv,
            refMap:   targetRefMap,
            features: targetFeatures,
            policy: planner.RelayTargetPolicy{
                CapabilitiesKnown: targetFeatures.Known,
                NoThin:            targetFeatures.NoThin,
            },
            pusher: gitproto.NewPusher(targetConn, targetAdv, cfg.Verbose),
        s.target.adv = targetAdv
        s.target.refMap = targetRefMap
        s.target.features = targetFeatures
        s.target.policy = planner.RelayTargetPolicy{
            CapabilitiesKnown: targetFeatures.Known,
            NoThin:            targetFeatures.NoThin,
            }
        s.target.pusher = gitproto.NewPusher(targetConn, targetAdv, cfg.Verbose)
        if cfg.BestEffort {
            s.rejections = make(map[plumbing.ReferenceName]string)
            s.target.pusher.OnRejection = func(name plumbing.ReferenceName, status string) {
23 unmodified lines

s
        }
        }
    }

success = true
    return s, nil
}

Minternal/syncer/syncer.go+30/-13