1package replica
2
3import (
4 "context"
5 "errors"
6 "net"
7 "strconv"
8 "sync"
9 "sync/atomic"
10 "time"
11
12 "github.com/go-git/go-git/v5"
13 "github.com/go-git/go-git/v5/plumbing"
14 "github.com/go-git/go-git/v5/plumbing/transport"
15)
16
17var (
18 ErrFetchIdle = errors.New("replica: whole-history fetch idle bound exceeded")
19 ErrFetchTotal = errors.New("replica: whole-history fetch total bound exceeded")
20 activeFetches sync.Map
21 fetchSequence atomic.Uint64
22)
23
24type fetchLimit struct {
25 ctx context.Context
26 idle time.Duration
27 total time.Duration
28 timer *time.Timer
29}
30
31type fetchConn struct {
32 net.Conn
33 limit *fetchLimit
34 stop func() bool
35}
36
37func (c *fetchConn) Read(p []byte) (int, error) {
38 n, err := c.Conn.Read(p)
39 if n > 0 && c.limit.timer != nil {
40 c.limit.timer.Reset(c.limit.idle)
41 }
42 return n, err
43}
44
45func (c *fetchConn) Write(p []byte) (int, error) {
46 n, err := c.Conn.Write(p)
47 if n > 0 && c.limit.timer != nil {
48 c.limit.timer.Reset(c.limit.idle)
49 }
50 return n, err
51}
52
53func (c *fetchConn) Close() error {
54 c.stop()
55 return c.Conn.Close()
56}
57
58func FetchHistory(ctx context.Context, repo *git.Repository, url string, auth transport.AuthMethod, branch plumbing.ReferenceName) (plumbing.Hash, error) {
59 a, ok := auth.(*SSH)
60 if !ok {
61 return Fetch(ctx, repo, url, auth, branch)
62 }
63 idle, total := a.FetchIdleTimeout, a.FetchTotalTimeout
64 if idle <= 0 {
65 idle = DefaultFetchIdleTimeout
66 }
67 if total <= 0 {
68 total = DefaultFetchTotalTimeout
69 }
70 ctx, cancel := context.WithCancelCause(ctx)
71 defer cancel(nil)
72 limit := &fetchLimit{ctx: ctx, idle: idle}
73 limit.timer = time.AfterFunc(idle, func() { cancel(ErrFetchIdle) })
74 defer limit.timer.Stop()
75 totalTimer := time.AfterFunc(total, func() { cancel(ErrFetchTotal) })
76 defer totalTimer.Stop()
77 copy := *a
78 copy.Timeout = total
79 auth, release := registerFetch(©, limit)
80 defer release()
81 h, err := Fetch(ctx, repo, url, auth, branch)
82 if cause := context.Cause(ctx); cause != nil {
83 return plumbing.ZeroHash, cause
84 }
85 return h, err
86}
87
88func registerFetch(auth *SSH, limit *fetchLimit) (*SSH, func()) {
89 copy := *auth
90 copy.fetch = strconv.FormatUint(fetchSequence.Add(1), 10)
91 activeFetches.Store(copy.fetch, limit)
92 return ©, func() { activeFetches.Delete(copy.fetch) }
93}